| """ |
| Orchestration layer: run the full QC analysis on one image or a batch, and |
| assemble the artefacts (scores, maps, descriptors) the UI needs. |
| """ |
|
|
| from __future__ import annotations |
| import numpy as np |
| import cv2 |
| from .fov import detect_fov |
| from .qc_metrics import compute_all_metrics, METRIC_WEIGHTS |
| from .quality_score import composite_score |
| from .failure_analysis import composite_problem_map, fov_overlay, per_axis_heatmap |
| from .batch_effects import descriptor_from_metrics |
|
|
| METRIC_NAMES = list(METRIC_WEIGHTS.keys()) |
|
|
|
|
| def _to_rgb(img): |
| a = np.asarray(img) |
| if a.ndim == 2: |
| a = cv2.cvtColor(a, cv2.COLOR_GRAY2RGB) |
| if a.shape[-1] == 4: |
| a = a[..., :3] |
| return np.ascontiguousarray(a.astype(np.uint8)) |
|
|
|
|
| def degradation_map_from_metrics(shape, metrics): |
| """Aggregate the spatial maps of failing/borderline axes into one 0-1 map.""" |
| dmap = np.zeros(shape[:2], np.float32) |
| for m in metrics: |
| if m.get("_map") is not None and m["score"] < 0.66: |
| w = (0.66 - m["score"]) |
| dmap = np.maximum(dmap, m["_map"].astype(np.float32) * w) |
| if dmap.max() > 0: |
| dmap /= dmap.max() |
| return dmap |
|
|
|
|
| def analyze_image(img, name="image", with_vessels=True): |
| """Full single-image analysis. Returns a dict bundle.""" |
| rgb = _to_rgb(img) |
| fov = detect_fov(rgb) |
| metrics = compute_all_metrics(rgb, fov) |
| summary = composite_score(metrics) |
| problem_overlay, problem_caption = composite_problem_map(rgb, fov, metrics, summary) |
| descriptor = descriptor_from_metrics(rgb, metrics, fov) |
| dmap = degradation_map_from_metrics(rgb.shape, metrics) |
| bundle = dict( |
| name=name, rgb=rgb, fov=fov, metrics=metrics, summary=summary, |
| problem_overlay=problem_overlay, problem_caption=problem_caption, |
| fov_overlay=fov_overlay(rgb, fov), descriptor=descriptor, |
| degradation_map=dmap, |
| ) |
| if with_vessels: |
| try: |
| from .vessels import analyze_vessels, vessel_gradability_score |
| v = analyze_vessels(rgb, fov) |
| v["gradability"], v["parts"] = vessel_gradability_score(v) |
| bundle["vessels"] = v |
| except Exception as e: |
| bundle["vessels_error"] = str(e) |
| return bundle |
|
|
|
|
| def analyze_batch(images, names=None, progress=None): |
| """Analyse a list of images. `images` may be file paths or arrays.""" |
| from PIL import Image |
| results = [] |
| n = len(images) |
| for i, im in enumerate(images): |
| if progress is not None: |
| progress((i + 1) / max(n, 1), desc=f"Analysing {i+1}/{n}") |
| if isinstance(im, str): |
| nm = names[i] if names else im.split("/")[-1] |
| arr = np.array(Image.open(im).convert("RGB")) |
| else: |
| nm = names[i] if names else f"image_{i:03d}" |
| arr = _to_rgb(im) |
| try: |
| results.append(analyze_image(arr, nm)) |
| except Exception as e: |
| results.append(dict(name=nm, error=str(e), rgb=_to_rgb(arr))) |
| return results |
|
|
|
|
| def results_to_dataframe(results): |
| import pandas as pd |
| rows = [] |
| for r in results: |
| if "error" in r: |
| rows.append(dict(image=r["name"], composite=np.nan, verdict="ERROR")) |
| continue |
| row = dict(image=r["name"], |
| composite=round(r["summary"]["composite"], 1), |
| verdict=r["summary"]["verdict"], |
| band=r["summary"]["band"], |
| primary_reason=r["summary"]["primary_reason"]) |
| for m in r["metrics"]: |
| row[m["name"]] = round(m["score"], 3) |
| rows.append(row) |
| return pd.DataFrame(rows) |
|
|
|
|
| def metric_table(result): |
| """Per-axis table for one image (list of rows).""" |
| rows = [] |
| for m in result["metrics"]: |
| rows.append([m["name"], f'{m["value"]:.2f} {m["unit"]}', |
| f'{m["score"]:.2f}', m["status"].upper(), m["reason"]]) |
| return rows |
|
|
