Spaces:
Sleeping
Sleeping
| # mapping_rules.py | |
| import re, datetime | |
| KNOWN_HEADWORKS = ["Tarbela","Kalabagh","Chashma","Taunsa","Taunsa","Guddu","Sukkur","Kotri","Mangla","Rasul","Marala", | |
| "Khanki","Qadirabad","Trimmu","Panjnad","Balloki","Sidhnai","Sulemanki","Islam",] | |
| def parse_number(s): | |
| if s is None: return None | |
| s = str(s).strip() | |
| if s == "": return None | |
| s = s.replace(',', '').replace('O','0').replace('o','0').replace('I','1').replace('l','1').replace('S','5') | |
| neg = False | |
| if re.search(r'^\(.*\)$', s): | |
| neg = True | |
| s = s.strip('()') | |
| m = re.search(r'[-+]?\d+(\.\d+)?', s) | |
| if not m: return None | |
| try: | |
| val = float(m.group(0)) | |
| except: | |
| return None | |
| return -val if neg else val | |
| def parse_date(s): | |
| if not s: return None | |
| s = str(s).strip() | |
| for fmt in ("%d-%m-%Y","%d/%m/%Y","%Y-%m-%d","%d-%m-%y","%d %b %Y","%d %B %Y"): | |
| try: | |
| return datetime.datetime.strptime(s, fmt).date().isoformat() | |
| except: | |
| pass | |
| m = re.search(r'(\d{1,2})[\/\-](\d{1,2})[\/\-](\d{2,4})', s) | |
| if m: | |
| d,mo,y = m.groups() | |
| if len(y)==2: y = '20'+y | |
| return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}" | |
| return None | |
| def find_known_headworks(text): | |
| if not text: return None | |
| for h in KNOWN_HEADWORKS: | |
| if h.lower() in text.lower(): | |
| return h | |
| return None | |
| def detect_unit(text): | |
| if not text: return None | |
| t = str(text) | |
| if re.search(r'\b(cs|Cs|CS)\b', t): return 'Cs' | |
| if re.search(r'\b(ft|FT)\b', t): return 'ft' | |
| if re.search(r'\b(ind|IND)\b', t): return 'IND' | |
| return None | |
| def map_row(cells, confidences=None, header_tokens=None): | |
| """ | |
| cells: list of strings (OCR per column) | |
| confidences: list of floats (0..1) | |
| header_tokens: optional list of header strings if row is header | |
| returns: dict with canonical fields | |
| """ | |
| if confidences is None: | |
| confidences = [0.95]*len(cells) | |
| joined = " | ".join([c for c in cells if c]) | |
| header = " ".join(header_tokens).lower() if header_tokens else "" | |
| out = { | |
| "Date": None, "Headworks": None, "Canal": None, "Site": None, | |
| "A.F.S.Q": None, "Gauge": None, "Indent": None, "Actual": None, | |
| "Unit": None, "SourceRowText": joined, "OCRConfidence": round(sum(confidences)/len(confidences),3) | |
| } | |
| # Date | |
| for c in cells: | |
| d = parse_date(c) | |
| if d: | |
| out["Date"] = d | |
| break | |
| # Headworks | |
| if header and "headwork" in header: | |
| for t in header_tokens: | |
| h = find_known_headworks(t) | |
| if h: | |
| out["Headworks"] = h | |
| break | |
| if not out["Headworks"]: | |
| for c in cells[:3]: | |
| h = find_known_headworks(c) | |
| if h: | |
| out["Headworks"] = h | |
| break | |
| # Gauge detection (decimal-like) | |
| gauge_idx = None | |
| for i,c in enumerate(cells): | |
| if c and re.search(r'\d+\.\d{1,2}', str(c)): | |
| gauge_idx = i | |
| out["Gauge"] = parse_number(c) | |
| break | |
| # Numeric candidates | |
| nums = [(i, parse_number(c)) for i,c in enumerate(cells) if parse_number(c) is not None] | |
| nums = [(i,v) for (i,v) in nums if i!=gauge_idx] | |
| nums_sorted = sorted(nums, key=lambda x: abs(x[1]) if x[1] is not None else 0, reverse=True) | |
| if len(nums_sorted)>=1: | |
| out["Indent"] = nums_sorted[0][1] | |
| if len(nums_sorted)>=2: | |
| out["Actual"] = nums_sorted[1][1] | |
| # A.F.S.Q detection (small decimal or single-digit) | |
| for c in cells: | |
| try: | |
| val = float(str(c).strip()) | |
| if val>0 and val<30: | |
| out["A.F.S.Q"] = val | |
| break | |
| except: | |
| continue | |
| # Unit detection | |
| for c in cells: | |
| u = detect_unit(c) | |
| if u: | |
| out["Unit"] = u | |
| break | |
| # Canal and Site heuristics | |
| for c in cells: | |
| if c and any(k in str(c).lower() for k in ["canal","link","feeder","escape","branch","rd","rd-","tail","head"]): | |
| if not out["Canal"]: | |
| out["Canal"] = str(c).strip().title() | |
| elif not out["Site"]: | |
| out["Site"] = str(c).strip().upper() | |
| # Site tokens U/S D/S HEAD TAIL | |
| for c in cells: | |
| if c and any(k in str(c).upper() for k in ["U/S","D/S","HEAD","TAIL","RD","RD-"]): | |
| out["Site"] = str(c).strip().upper() | |
| break | |
| return out | |