Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| from datetime import datetime, date, time, timedelta | |
| from pathlib import Path | |
| import os | |
| # ===== Policy ===== | |
| POLICY_CHANGE = date(2025, 7, 7) # inclusive | |
| BASELINE_BEFORE = time(17, 30) # BEFORE 2025-07-07 | |
| BASELINE_AFTER = time(17, 0) # FROM 2025-07-07 | |
| VIRTUAL_BEFORE = time(18, 0) # Case 4 floor (BEFORE) | |
| VIRTUAL_AFTER = time(17, 30) # Case 4 floor (FROM) | |
| IN_TOKENS = {"P10","IN","INN","CHECKIN","CHECK-IN"} | |
| OUT_TOKENS = {"P20","OUT","CHECKOUT","CHECK-OUT"} | |
| SMALL_GAP_MERGE_SEC = 30 * 60 # merge OUT→IN gaps under 30 minutes | |
| # ===== Helpers ===== | |
| def parse_yyyymmdd(s): | |
| s = str(s).strip() | |
| for fmt in ("%Y%m%d","%Y-%m-%d","%d/%m/%Y","%m/%d/%Y"): | |
| try: return datetime.strptime(s, fmt).date() | |
| except Exception: pass | |
| return pd.NaT | |
| def parse_hms(s): | |
| s = str(s).strip() | |
| if s.isdigit(): | |
| s = s.zfill(6)[:6] | |
| hh,mm,ss = s[:2], s[2:4], s[4:6] | |
| try: return time(int(hh), int(mm), int(ss)) | |
| except Exception: return pd.NaT | |
| for fmt in ("%H:%M:%S","%H:%M"): | |
| try: return datetime.strptime(s, fmt).time() | |
| except Exception: pass | |
| return pd.NaT | |
| def combine_dt(d, t): | |
| if pd.isna(d) or pd.isna(t): return pd.NaT | |
| return datetime.combine(d, t) | |
| def from_policy(d): return (not pd.isna(d)) and (d >= POLICY_CHANGE) | |
| def baseline_cutoff(d): return BASELINE_AFTER if from_policy(d) else BASELINE_BEFORE | |
| def virtual_checkin(d): return VIRTUAL_AFTER if from_policy(d) else VIRTUAL_BEFORE | |
| def detect_event_token(v): | |
| u = str(v).upper() | |
| is_in = any(tok == u or tok in u for tok in IN_TOKENS) | |
| is_out = any(tok == u or tok in u for tok in OUT_TOKENS) | |
| if is_in and not is_out: return "in" | |
| if is_out and not is_in: return "out" | |
| return np.nan | |
| def uniq_times_str(ts): | |
| return ", ".join(sorted({t.strftime("%H:%M:%S") for t in ts.dropna()})) | |
| def fmt_time(x): | |
| if pd.isna(x): return "" | |
| if isinstance(x, datetime): return x.strftime("%H:%M:%S") | |
| if isinstance(x, time): return x.strftime("%H:%M:%S") | |
| return str(x) | |
| # Collapse runs of identical events: | |
| # - keep earliest time of an IN-run | |
| # - keep latest time of an OUT-run | |
| def compress_events(events): | |
| # events: list[(ts, 'in'|'out')] ascending | |
| if not events: | |
| return [] | |
| out = [] | |
| for ts, ev in events: | |
| if not out: | |
| out.append([ts, ev]) | |
| continue | |
| last_ts, last_ev = out[-1] | |
| if ev == last_ev: | |
| if ev == "out": | |
| out[-1][0] = ts # extend OUT-run to latest OUT | |
| # IN-run: keep earliest | |
| else: | |
| out.append([ts, ev]) | |
| return [(ts, ev) for ts, ev in out] | |
| # Merge small OFF gaps: remove OUT→IN pairs with gap < threshold | |
| def merge_small_gaps(events, threshold_sec=SMALL_GAP_MERGE_SEC): | |
| # events must be compressed already. | |
| merged = [] | |
| i = 0 | |
| n = len(events) | |
| while i < n: | |
| ts, ev = events[i] | |
| if ev == "out" and merged and merged[-1][1] == "in" and (i + 1) < n and events[i+1][1] == "in": | |
| gap = (events[i+1][0] - ts).total_seconds() | |
| if gap < threshold_sec: | |
| i += 2 | |
| continue | |
| merged.append([ts, ev]) | |
| i += 1 | |
| return [(ts, ev) for ts, ev in merged] | |
| # Summarize OT between start_time and end_time on a clean event stream | |
| def summarize_ot(start_time, end_time, events, start_state): | |
| # returns: (work_total_sec, list_internal_off_gaps_sec, segments_count) | |
| work_total = 0.0 | |
| segments = 0 | |
| off_gaps = [] | |
| cur_state = start_state | |
| cur = start_time | |
| off_start = None | |
| for ts, ev in events: | |
| if ts <= start_time: continue | |
| if ts > end_time: break | |
| if cur_state == "in": | |
| work_total += (ts - cur).total_seconds() | |
| if ev == "out": | |
| segments += 1 | |
| off_start = ts | |
| else: | |
| if ev == "in" and off_start is not None: | |
| off_gaps.append((ts - off_start).total_seconds()) | |
| off_start = None | |
| cur_state = ev | |
| cur = ts | |
| if cur < end_time and cur_state == "in": | |
| work_total += (end_time - cur).total_seconds() | |
| return work_total, off_gaps, segments | |
| # ===== Core computation ===== | |
| def compute_ot(df_raw: pd.DataFrame) -> pd.DataFrame: | |
| df = df_raw.copy() | |
| df.columns = [str(c).strip().lower() for c in df.columns] | |
| # detect 3 required columns heuristically | |
| def find_col(df, candidates, fallback=None): | |
| for col in df.columns: | |
| if any(pat in col for pat in candidates): | |
| return col | |
| return fallback | |
| emp_col = find_col(df, ["emp","employee","staff","id","person"], df.columns[0] if len(df.columns)>0 else None) | |
| date_col = find_col(df, ["date","yyyymmdd","ngay"], df.columns[1] if len(df.columns)>1 else None) | |
| time_col = find_col(df, ["time","gio","hh","clock"], df.columns[2] if len(df.columns)>2 else None) | |
| if any(c is None for c in [emp_col, date_col, time_col]): | |
| raise gr.Error(f"Expected 3 columns: employee, yyyymmdd, hh:mm:ss (or hhmmss). Got {list(df.columns)}") | |
| # status column (optional) | |
| evt_col = None | |
| for c in df.columns: | |
| vals = df[c].astype(str).str.upper() | |
| if vals.isin(["P10","P20"]).any() or vals.str.contains(r"\bP10\b|\bP20\b", regex=True, na=False).any(): | |
| evt_col = c | |
| break | |
| df["__emp"] = df[emp_col].astype(str).str.strip() | |
| df["__date"] = df[date_col].apply(parse_yyyymmdd) | |
| df["__time"] = df[time_col].apply(parse_hms) | |
| df["__ts"] = [combine_dt(d, t) for d, t in zip(df["__date"], df["__time"])] | |
| if evt_col: | |
| df["__event"] = df[evt_col].apply(detect_event_token) | |
| else: | |
| def scan_row(row): | |
| for c in row.index: | |
| ev = detect_event_token(row[c]) | |
| if pd.notna(ev): return ev | |
| return np.nan | |
| df["__event"] = df.apply(scan_row, axis=1) | |
| df = df.dropna(subset=["__ts"]).sort_values(["__emp","__date","__ts"]).reset_index(drop=True) | |
| df["__baseline_dt"] = [combine_dt(d, baseline_cutoff(d)) for d in df["__date"]] | |
| df["__virtual_dt"] = [combine_dt(d, virtual_checkin(d)) for d in df["__date"]] | |
| rows = [] | |
| for (emp, d), day in df.groupby(["__emp","__date"], sort=True): | |
| if pd.isna(d): | |
| continue | |
| day = day.sort_values("__ts") | |
| baseline_dt = day["__baseline_dt"].iloc[0] | |
| virtual_dt = day["__virtual_dt"].iloc[0] | |
| pre = day[day["__ts"] < baseline_dt] | |
| after = day[day["__ts"] >= baseline_dt] | |
| if after.empty: | |
| continue | |
| last_pre_event = pre.iloc[-1]["__event"] if not pre.empty else np.nan | |
| ins_after = after.loc[after["__event"].eq("in"), "__ts"] | |
| outs_after = after.loc[after["__event"].eq("out"), "__ts"] | |
| in_times_list = uniq_times_str(ins_after) | |
| out_times_list = uniq_times_str(outs_after) | |
| code = "" | |
| ot_hours = np.nan | |
| virtual_used_str = "" | |
| # Need at least one OUT after cutoff | |
| if len(outs_after) == 0: | |
| if len(ins_after) > 0: | |
| code = "alert" | |
| continue | |
| end_dt = outs_after.max() | |
| # Build + clean events after cutoff up to last OUT | |
| events_df = after.loc[(after["__ts"] <= end_dt) & (after["__event"].isin(["in","out"])), ["__ts","__event"]] | |
| events_raw = list(events_df.itertuples(index=False, name=None)) | |
| events = merge_small_gaps(compress_events(events_raw), threshold_sec=SMALL_GAP_MERGE_SEC) | |
| if len(ins_after) == 0: | |
| # No IN after cutoff | |
| if last_pre_event == "in": | |
| # Case 4: inside at cutoff; baseline minus OFF gaps, floored by virtual | |
| work_secs, _, _ = summarize_ot(baseline_dt, end_dt, events, start_state="in") | |
| eff_start = end_dt - timedelta(seconds=work_secs) | |
| start_dt = max(eff_start, virtual_dt) | |
| ot_hours = round(max(0.0, (end_dt - start_dt).total_seconds())/3600.0, 2) | |
| virtual_used_str = fmt_time(start_dt) | |
| else: | |
| # Case 2: from baseline | |
| ot_hours = round(max(0.0, (end_dt - baseline_dt).total_seconds())/3600.0, 2) | |
| else: | |
| # Have both IN and OUT after cutoff | |
| first_after_ev = after.iloc[0]["__event"] | |
| inside_at_cutoff = (last_pre_event == "in") or (first_after_ev == "out") | |
| if inside_at_cutoff: | |
| # Case 4: baseline minus gaps, floored by virtual | |
| work_secs, _, _ = summarize_ot(baseline_dt, end_dt, events, start_state="in") | |
| eff_start = end_dt - timedelta(seconds=work_secs) | |
| start_dt = max(eff_start, virtual_dt) | |
| ot_hours = round(max(0.0, (end_dt - start_dt).total_seconds())/3600.0, 2) | |
| virtual_used_str = fmt_time(start_dt) | |
| else: | |
| # Case 1: earliest IN after cutoff, subtract OFF gaps | |
| start_dt = ins_after.min() | |
| work_secs, _, _ = summarize_ot(start_dt, end_dt, events, start_state="in") | |
| ot_hours = round(max(0.0, work_secs)/3600.0, 2) | |
| rows.append({ | |
| "employee_id": emp, | |
| "date": d.strftime("%Y%m%d"), | |
| "ot_hours": ot_hours, | |
| "code": code, | |
| "in_times": in_times_list, | |
| "out_times": out_times_list, | |
| "virtual_in_time": virtual_used_str | |
| }) | |
| res = pd.DataFrame(rows, columns=[ | |
| "employee_id","date","ot_hours","code","in_times","out_times","virtual_in_time" | |
| ]) | |
| if not res.empty: | |
| res = res.sort_values(["employee_id","date"]).reset_index(drop=True) | |
| return res | |
| # ===== I/O wrappers ===== | |
| def load_dataframe(upload): | |
| # Gradio v4 passes a FileData object with .path; also accept str/Path | |
| if upload is None: | |
| raise gr.Error("Please upload a file.") | |
| path = None | |
| if isinstance(upload, (str, Path)): | |
| path = str(upload) | |
| elif hasattr(upload, "path"): | |
| path = upload.path | |
| elif hasattr(upload, "name") and os.path.exists(upload.name): | |
| path = upload.name | |
| if not path or not os.path.exists(path): | |
| raise gr.Error("Could not resolve uploaded file path.") | |
| lower = path.lower() | |
| if lower.endswith((".xlsx",".xls")): | |
| df = pd.read_excel(path, dtype=str) | |
| elif lower.endswith(".csv"): | |
| try: | |
| df = pd.read_csv(path, dtype=str) | |
| except Exception: | |
| df = pd.read_csv(path, dtype=str, sep=";") | |
| else: | |
| raise gr.Error("Unsupported file type. Upload .xlsx/.xls/.csv.") | |
| return df | |
| def predict(upload): | |
| df_in = load_dataframe(upload) | |
| res = compute_ot(df_in) | |
| out_path = Path("ot_summary.xlsx") | |
| try: | |
| import xlsxwriter; engine = "xlsxwriter" | |
| except Exception: | |
| engine = "openpyxl" | |
| with pd.ExcelWriter(out_path, engine=engine) as w: | |
| res.to_excel(w, index=False, sheet_name="OT") | |
| return res, str(out_path) | |
| # Tiny example so users can test without uploading | |
| example_path = Path("example.csv") | |
| if not example_path.exists(): | |
| example_path.write_text( | |
| "emp,date,time,status\n" | |
| "3809273,20250704,125718,P10\n" | |
| "3809273,20250704,174514,P20\n" | |
| "3809273,20250704,174553,P10\n" | |
| "3809273,20250704,174608,P20\n" | |
| "3809273,20250704,174631,P10\n" | |
| "3809273,20250704,220230,P20\n", | |
| encoding="utf-8" | |
| ) | |
| # ===== Gradio app ===== | |
| iface = gr.Interface( | |
| fn=predict, | |
| inputs=gr.File(label="Upload Excel/CSV", file_types=[".xlsx",".xls",".csv"]), | |
| outputs=[ | |
| gr.Dataframe(label="Result preview", row_count=(12, "dynamic")), | |
| gr.File(label="Download OT Excel (.xlsx)") | |
| ], | |
| title="OT Calculator (P10/P20)", | |
| description=( | |
| "Policy: cutoff 17:30 before 2025-07-07, 17:00 from 2025-07-07. " | |
| "Case 4 floors start at 18:00 (before) or 17:30 (from). " | |
| "Consecutive duplicate events are collapsed; OUT→IN gaps < 30′ are merged." | |
| ), | |
| flagging_mode="never", | |
| examples=[[str(example_path)]], | |
| ) | |
| if __name__ == "__main__": | |
| iface.launch() | |