bigaxon/kinetic

1# model_runner.py2# 🔬 Kinetic Pathology — production-safe core (no UI, no launch)3 4import warnings, os5import numpy as np, pandas as pd, matplotlib; matplotlib.use("Agg")6import matplotlib.pyplot as plt7from typing import Optional, List8from datetime import datetime9from scipy.spatial.distance import cdist10from scipy.optimize import linear_sum_assignment11from skimage import io as skio, exposure, filters, morphology, measure, color12from skimage.restoration import denoise_tv_chambolle13 14warnings.filterwarnings("ignore")15 16try:17    import cv2  # optional18    CV2_AVAILABLE = True19except Exception:20    CV2_AVAILABLE = False21 22_TMP_DIR = "/tmp/kinetic"23os.makedirs(_TMP_DIR, exist_ok=True)24 25def _safe_path(name: str) -> str:26    return os.path.join(_TMP_DIR, name)27 28def _blank_png(path: str, text: str):29    fig = plt.figure(figsize=(4, 2.5), dpi=150)30    ax = fig.add_subplot(111)31    ax.text(0.5, 0.5, text, ha="center", va="center", fontsize=9)32    ax.axis("off")33    fig.tight_layout()34    fig.savefig(path, bbox_inches="tight")35    plt.close(fig)36 37def _empty_csv(path: str, cols: List[str]):38    import csv39    with open(path, "w", newline="") as f:40        csv.writer(f).writerow(cols)41 42def _fmt_sig2(x) -> str:43    try:44        xv = float(x)45    except Exception:46        return "nan"47    if not np.isfinite(xv) or xv == 0:48        return "0"49    s = f"{xv:.2g}"50    if ("e" in s or "E" in s) and 1e-2 <= abs(xv) < 1e4:51        mag = int(np.floor(np.log10(abs(xv))))52        norm = xv / (10 ** mag)53        s = f"{norm:.2f}e{mag:+d}"54    return s55 56def _df_preview(df: Optional[pd.DataFrame], n: int = 10) -> pd.DataFrame:57    if df is None: return pd.DataFrame()58    try: return df.head(n).copy()59    except Exception: return pd.DataFrame()60 61def run(62    video_path: str,63    max_frames: int,64    tv_weight: float,65    min_area: int,66    max_area: int,67    opening_radius: int,68    max_link_distance: float,69    use_um: bool,70    px_um: float,71    show_ids: bool,72    show_trails: bool,73    trail_len: int,74    circle_sz: int,75    do_emergent: bool,76    auto_center: bool,77    lesion_cx: Optional[float],78    lesion_cy: Optional[float],79    neighbor_radius: float,80):81    """Return exactly 11 outputs for the Gradio UI."""82    if not video_path or not os.path.exists(video_path):83        summary = "Please upload a valid video file."84        report_path   = _safe_path("report_placeholder.png");  _blank_png(report_path, "No video")85        counts_path   = _safe_path("counts_placeholder.png");  _blank_png(counts_path, "Counts")86        traj_path     = _safe_path("traj_placeholder.png");    _blank_png(traj_path, "Trajectories")87        origin_path   = _safe_path("origin_placeholder.png");  _blank_png(origin_path, "Origin")88        tracks_csv    = _safe_path("tracks.csv");              _empty_csv(tracks_csv, ["track_id","t","x","y"])89        feats_csv     = _safe_path("features.csv");            _empty_csv(feats_csv, ["cell_id","area","speed"])90        return (summary, report_path, "", "",91                counts_path, traj_path, origin_path,92                tracks_csv, feats_csv, pd.DataFrame(), pd.DataFrame())93 94    try:95        # ====== STUB ANALYSIS (replace later with real code) ======96        n_frames_scanned = 097        means = []98        if CV2_AVAILABLE:99            cap = None100            try:101                import cv2102                cap = cv2.VideoCapture(video_path)103                while n_frames_scanned < min(max_frames, 30):104                    ok, frame = cap.read()105                    if not ok: break106                    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)107                    means.append(float(gray.mean()))108                    n_frames_scanned += 1109            finally:110                if cap is not None: cap.release()111        avg_intensity = np.mean(means) if means else np.nan112        unit = "µm/s" if use_um else "px/s"113 114        summary = (115            f"Kinetic Pathology (stub)\n"116            f"- Frames scanned: {n_frames_scanned}/{max_frames}\n"117            f"- Avg intensity: {_fmt_sig2(avg_intensity)}\n"118            f"- Units: {unit}\n"119        )120 121        ts = datetime.now().strftime("%Y%m%d_%H%M%S")122        report_path = _safe_path(f"report_{ts}.png"); _blank_png(report_path, "Report (stub)")123        counts_path = _safe_path(f"counts_{ts}.png"); _blank_png(counts_path, "Counts (stub)")124        traj_path   = _safe_path(f"traj_{ts}.png");   _blank_png(traj_path, "Trajectories (stub)")125        origin_path = _safe_path(f"origin_{ts}.png"); _blank_png(origin_path, "Origin (stub)")126 127        orig_vid_path = video_path128        over_vid_path = ""  # supply when you produce an overlay129 130        tracks_csv = _safe_path(f"tracks_{ts}.csv");  _empty_csv(tracks_csv, ["track_id","t","x","y"])131        feats_csv  = _safe_path(f"features_{ts}.csv"); _empty_csv(feats_csv, ["cell_id","area","speed"])132        feats_prev  = pd.DataFrame({"cell_id":[], "area":[], "speed":[]})133        tracks_prev = pd.DataFrame({"track_id":[], "t":[], "x":[], "y":[]})134        # ====== END STUB ======135 136        return (summary, report_path, orig_vid_path, over_vid_path,137                counts_path, traj_path, origin_path,138                tracks_csv, feats_csv, _df_preview(feats_prev), _df_preview(tracks_prev))139 140    except Exception as e:141        err = f"Analysis failed: {type(e).__name__}: {e}"142        report_path = _safe_path("report_error.png"); _blank_png(report_path, "Error")143        counts_path = _safe_path("counts_error.png"); _blank_png(counts_path, "Counts error")144        traj_path   = _safe_path("traj_error.png");   _blank_png(traj_path, "Traj error")145        origin_path = _safe_path("origin_error.png"); _blank_png(origin_path, "Origin error")146        tracks_csv  = _safe_path("tracks_error.csv"); _empty_csv(tracks_csv, ["track_id","t","x","y"])147        feats_csv   = _safe_path("features_error.csv"); _empty_csv(feats_csv, ["cell_id","area","speed"])148        return (err, report_path, "", "",149                counts_path, traj_path, origin_path,150                tracks_csv, feats_csv, pd.DataFrame(), pd.DataFrame())151