Create features.py

engine/features.py

@@ -0,0 +1,142 @@
+# engine/features.py
+import json
+import numpy as np
+import re
+from typing import Dict, List, Any
+
+
+# ------------------------------------------------------------------
+# Load schema once
+# ------------------------------------------------------------------
+
+_FEATURE_SCHEMA_PATH = "data/feature_schema.json"
+
+with open(_FEATURE_SCHEMA_PATH, "r", encoding="utf-8") as f:
+    SCHEMA = json.load(f)
+
+FEATURES = SCHEMA["features"]
+
+
+# ------------------------------------------------------------------
+# Helper mappings
+# ------------------------------------------------------------------
+
+PNV_MAP = {
+    "positive": 1.0,
+    "negative": -1.0,
+    "variable": 0.5,
+    "unknown": 0.0,
+    None: 0.0
+}
+
+SHAPE_MAP = {
+    "cocci": 1.0,
+    "rods": 2.0,
+    "short rods": 2.5,
+    "spiral": 3.0,
+    "yeast": 4.0,
+    "variable": 0.5,
+    "unknown": 0.0
+}
+
+OXYGEN_MAP = {
+    "aerobic": 1.0,
+    "anaerobic": 2.0,
+    "facultative anaerobe": 3.0,
+    "microaerophilic": 4.0,
+    "capnophilic": 5.0,
+    "unknown": 0.0
+}
+
+
+# ------------------------------------------------------------------
+# Normalisation helpers
+# ------------------------------------------------------------------
+
+def _norm(s: Any) -> str:
+    if not s:
+        return "unknown"
+    return str(s).strip().lower()
+
+
+def _map_pnv(x: Any) -> float:
+    return PNV_MAP.get(_norm(x), 0.0)
+
+
+def _map_shape(x: Any) -> float:
+    return SHAPE_MAP.get(_norm(x), 0.0)
+
+
+def _map_oxygen(x: Any) -> float:
+    return OXYGEN_MAP.get(_norm(x), 0.0)
+
+
+def _growth_minmax(v: str):
+    """
+    Convert '30//37' → (30, 37)
+    If missing, return (0, 0)
+    """
+    if not v:
+        return (0.0, 0.0)
+    m = re.match(r"^\s*(\d+)\s*//\s*(\d+)\s*$", v)
+    if not m:
+        return (0.0, 0.0)
+    return (float(m.group(1)), float(m.group(2)))
+
+
+def _media_flag(media_field: str, medium: str) -> float:
+    """
+    Return 1.0 if medium appears in media list, else 0.0.
+    """
+    if not media_field:
+        return 0.0
+    mf = media_field.lower()
+    return 1.0 if medium.lower() in mf else 0.0
+
+
+# ------------------------------------------------------------------
+# Main public function
+# ------------------------------------------------------------------
+
+def extract_feature_vector(fused_fields: Dict[str, Any]) -> np.ndarray:
+    """
+    Convert fused tri-fusion fields into a fixed-length numeric vector.
+    Ordered exactly according to feature_schema.json.
+    Unknowns → 0.0.
+    """
+    vec: List[float] = []
+
+    for f in FEATURES:
+        name = f["name"]
+        kind = f["kind"]
+
+        value = fused_fields.get(name, "Unknown")
+
+        if kind == "pnv":
+            vec.append(_map_pnv(value))
+
+        elif kind == "shape":
+            vec.append(_map_shape(value))
+
+        elif kind == "oxygen":
+            vec.append(_map_oxygen(value))
+
+        elif kind == "numeric_from_growth_temp":
+            low, high = _growth_minmax(value)
+            vec.append(low)
+            vec.append(high)
+            # IMPORTANT: skip the next schema feature
+            # (schema should include two entries but model expects two values)
+            continue
+
+        elif kind == "media_flag":
+            # Each media entry in schema specifies the medium name
+            # e.g. "MacConkey Growth"
+            medium = name.replace("Growth", "").strip()
+            vec.append(_media_flag(fused_fields.get("Media Grown On"), medium))
+
+        else:
+            # Unknown kind → default numeric 0
+            vec.append(0.0)
+
+    return np.array(vec, dtype=float)