Spaces:
Sleeping
Sleeping
| # File: logic/care_gap_engine.py | |
| import pandas as pd | |
| from datetime import datetime | |
| def normalize_bool(value): | |
| if isinstance(value, str): | |
| value = value.strip().lower() | |
| if value in ["yes", "true", "1"]: | |
| return True | |
| elif value in ["no", "false", "0"]: | |
| return False | |
| elif isinstance(value, (int, float)): | |
| return bool(value) | |
| return False | |
| def normalize_gender(value): | |
| if isinstance(value, str): | |
| v = value.strip().lower() | |
| if v in ["f", "female", "2"]: | |
| return "F" | |
| elif v in ["m", "male", "1"]: | |
| return "M" | |
| elif isinstance(value, int): | |
| return "F" if value == 2 else "M" | |
| return "U" # Unknown | |
| def normalize_float(value, default=0.0): | |
| try: | |
| return float(value) | |
| except: | |
| return default | |
| def normalize_readmissions(value): | |
| try: | |
| return int(value) if int(value) >= 0 else 0 | |
| except: | |
| return 0 | |
| def evaluate_care_gaps(df: pd.DataFrame, config): | |
| today = datetime.today() | |
| rules = config["care_gap_rules"] | |
| results = [] | |
| for _, row in df.iterrows(): | |
| gaps = [] | |
| gender = normalize_gender(row.get('gender')) | |
| age = normalize_float(row.get('age')) | |
| systolic_bp = normalize_float(row.get('systolic_bp')) | |
| hba1c_value = normalize_float(row.get('hba1c_value')) | |
| last_mammo = pd.to_datetime(row.get('last_mammogram', None), errors='coerce') | |
| last_colono = pd.to_datetime(row.get('last_colonoscopy', None), errors='coerce') | |
| if gender == 'F' and rules['Breast Cancer Screening']['min_age'] <= age <= rules['Breast Cancer Screening']['max_age']: | |
| if pd.isna(last_mammo) or (today - last_mammo).days > rules['Breast Cancer Screening']['interval_days']: | |
| gaps.append("Breast Cancer Screening") | |
| if rules['Colorectal Cancer Screening']['min_age'] <= age <= rules['Colorectal Cancer Screening']['max_age']: | |
| if pd.isna(last_colono) or (today - last_colono).days > rules['Colorectal Cancer Screening']['interval_days']: | |
| gaps.append("Colorectal Cancer Screening") | |
| if systolic_bp > rules['Blood Pressure Control']['bp_threshold']: | |
| gaps.append("Blood Pressure Control") | |
| if hba1c_value >= 9: | |
| gaps.append("Diabetes: Poor HbA1c Control") | |
| if not normalize_bool(row.get('FollowUp_Scheduled')) or not normalize_bool(row.get('Primary_Care_Established')): | |
| gaps.append("Follow-Up Care") | |
| if normalize_readmissions(row.get('Previous_Readmissions')) >= 3: | |
| gaps.append("Readmission Risk") | |
| results.append({ | |
| 'patient_id': row['patient_id'], | |
| 'care_gaps': gaps | |
| }) | |
| return pd.DataFrame(results) | |