satiregram/HealthDataOps-Env

1#!/usr/bin/env python32"""3openenv validate — local validation script for HealthDataOps-Env.4Checks that all 3 tasks generate valid observations, that the graders5return floats in [0, 1], and that the YAML manifest is readable.6"""7 8import sys9import os10import yaml11 12sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))13 14PASS = "\033[92m✔\033[0m"15FAIL = "\033[91m✘\033[0m"16errors = []17 18def check(label: str, condition: bool, detail: str = ""):19    if condition:20        print(f"  {PASS} {label}")21    else:22        print(f"  {FAIL} {label}" + (f": {detail}" if detail else ""))23        errors.append(label)24 25print("\n══════════════════════════════════════════")26print("  openenv validate  ·  HealthDataOps-Env")27print("══════════════════════════════════════════\n")28 29# ── 1. YAML manifest ──────────────────────────────────────────────────────────30print("[1/4] Validating openenv.yaml …")31try:32    with open("openenv.yaml") as f:33        cfg = yaml.safe_load(f)34    check("name field present", "name" in cfg)35    check("version field present", "version" in cfg)36    check("3 tasks declared", len(cfg.get("tasks", [])) == 3)37    check("entry_point declared", "entry_point" in cfg)38    check("openenv tag present", "openenv" in cfg.get("tags", []))39except Exception as e:40    check("openenv.yaml readable", False, str(e))41 42# ── 2. Models importable ──────────────────────────────────────────────────────43print("\n[2/4] Validating Pydantic models …")44try:45    from env.models import T1Observation, T1Action, T2Observation, T2Action, T3Observation, T3Action46    check("T1Observation importable", True)47    check("T2Observation importable", True)48    check("T3Observation importable", True)49except Exception as e:50    check("Models importable", False, str(e))51 52# ── 3. Task generation ────────────────────────────────────────────────────────53print("\n[3/4] Validating task generation …")54try:55    from env.tasks.task1_diet_validator import DietValidatorTask56    t1 = DietValidatorTask()57    d = t1.generate_task()58    check("Task 1 generates observation", "observation" in d)59    check("Task 1 has ground_truth_errors", "ground_truth_errors" in d)60except Exception as e:61    check("Task 1 generation", False, str(e))62 63try:64    from env.tasks.task2_ehr_triage import EHRTriageTask65    t2 = EHRTriageTask()66    d = t2.generate_task()67    check("Task 2 generates observation", "observation" in d)68    check("Task 2 batch has 10 records", len(d["observation"]["batch_records"]) <= 10)69    check("Task 2 ground_truth is dict", isinstance(d["ground_truth"], dict))70except Exception as e:71    check("Task 2 generation", False, str(e))72 73try:74    from env.tasks.task3_weather_health import WeatherHealthTask75    t3 = WeatherHealthTask()76    d = t3.generate_task()77    check("Task 3 generates observation", "observation" in d)78    check("Task 3 has weather_aqi", "weather_aqi" in d["observation"])79    check("Task 3 has patient_conditions", "patient_conditions" in d["observation"])80except Exception as e:81    check("Task 3 generation", False, str(e))82 83# ── 4. End-to-end environment step ───────────────────────────────────────────84print("\n[4/4] Validating environment step loop …")85try:86    from env.health_data_ops_env import HealthDataOpsEnv87 88    env = HealthDataOpsEnv()89    obs, info = env.reset()90    check("env.reset() returns obs dict", isinstance(obs, dict))91 92    # Task 1 — dummy valid action93    action1 = {"is_valid": True, "issues": []}94    obs2, r1, done1, info2 = env.step(action1)95    check("Task 1 reward in [0,1]", 0.0 <= r1 <= 1.0, f"got {r1}")96 97    # Task 2 — dummy action98    batch_ids = [rec["record_id"] for rec in obs2.get("batch_records", [])]99    action2 = {"anomalies": [100        {"record_id": rid, "anomaly_type": "none", "severity": "low",101         "suggested_action": "no action", "confidence_score": 0.9}102        for rid in batch_ids103    ]}104    obs3, r2, done2, info3 = env.step(action2)105    check("Task 2 reward in [0,1]", 0.0 <= r2 <= 1.0, f"got {r2}")106 107    # Task 3 — dummy action108    action3 = {109        "risk_assessment": "Low risk day.",110        "dietary_adjustments": "Eat balanced meals with low sodium.",111        "medication_timing_adjustments": None,112        "safe_activity_recommendation": "Light indoor exercise recommended."113    }114    obs4, r3, done3, info4 = env.step(action3)115    check("Task 3 reward in [0,1]", 0.0 <= r3 <= 1.0, f"got {r3}")116    check("Environment terminates after 3 tasks", done3 is True)117 118except Exception as e:119    check("Environment step loop", False, str(e))120 121# ── Summary ───────────────────────────────────────────────────────────────────122print("\n══════════════════════════════════════════")123if errors:124    print(f"  \033[91mFAILED\033[0m — {len(errors)} check(s) did not pass:")125    for e in errors:126        print(f"    • {e}")127    sys.exit(1)128else:129    print("  \033[92mAll checks passed.\033[0m openenv validate: OK ✔")130print("══════════════════════════════════════════\n")131