"""Clinical audit report generator for regulatory compliance.
Generates structured HTML reports summarizing safety validation results,
model performance, and Fitzpatrick equity analysis for clinical review.
Reports include:
- Safety validation pass/fail summary per patient
- Aggregate statistics by procedure and Fitzpatrick type
- Flagged cases for manual review
- Model version and configuration provenance
Usage:
from landmarkdiff.audit import AuditReporter, AuditCase
reporter = AuditReporter(model_version="0.3.2")
reporter.add_case(AuditCase(
case_id="P001",
procedure="rhinoplasty",
safety_passed=True,
identity_sim=0.87,
fitzpatrick_type="III",
))
reporter.generate_report("audit_report.html")
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
@dataclass
class AuditCase:
"""A single patient case for audit reporting."""
case_id: str
procedure: str
safety_passed: bool
identity_sim: float = 0.0
intensity: float = 65.0
fitzpatrick_type: str = ""
warnings: list[str] = field(default_factory=list)
failures: list[str] = field(default_factory=list)
metrics: dict[str, float] = field(default_factory=dict)
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
@dataclass
class AuditSummary:
"""Aggregate statistics for an audit report."""
total_cases: int = 0
passed_cases: int = 0
failed_cases: int = 0
flagged_cases: int = 0
pass_rate: float = 0.0
mean_identity_sim: float = 0.0
by_procedure: dict[str, dict[str, Any]] = field(default_factory=dict)
by_fitzpatrick: dict[str, dict[str, Any]] = field(default_factory=dict)
class AuditReporter:
"""Generate clinical audit reports from safety validation results.
Args:
model_version: Model version string for provenance.
report_title: Title for generated reports.
"""
def __init__(
self,
model_version: str = "0.3.2",
report_title: str = "LandmarkDiff Clinical Audit Report",
) -> None:
self.model_version = model_version
self.report_title = report_title
self.cases: list[AuditCase] = []
def add_case(self, case: AuditCase) -> None:
"""Add a case to the audit report."""
self.cases.append(case)
def add_cases(self, cases: list[AuditCase]) -> None:
"""Add multiple cases."""
self.cases.extend(cases)
def clear(self) -> None:
"""Clear all cases."""
self.cases.clear()
def compute_summary(self) -> AuditSummary:
"""Compute aggregate statistics from all cases."""
if not self.cases:
return AuditSummary()
total = len(self.cases)
passed = sum(1 for c in self.cases if c.safety_passed)
failed = total - passed
flagged = sum(1 for c in self.cases if not c.safety_passed or c.warnings)
id_sims = [c.identity_sim for c in self.cases if c.identity_sim > 0]
mean_id = sum(id_sims) / len(id_sims) if id_sims else 0.0
# By procedure
by_proc: dict[str, dict[str, Any]] = {}
for case in self.cases:
proc = case.procedure
if proc not in by_proc:
by_proc[proc] = {"total": 0, "passed": 0, "id_sims": []}
by_proc[proc]["total"] += 1
if case.safety_passed:
by_proc[proc]["passed"] += 1
if case.identity_sim > 0:
by_proc[proc]["id_sims"].append(case.identity_sim)
for _proc, stats in by_proc.items():
stats["pass_rate"] = stats["passed"] / max(stats["total"], 1)
stats["mean_identity_sim"] = (
sum(stats["id_sims"]) / len(stats["id_sims"])
if stats["id_sims"]
else 0.0
)
del stats["id_sims"]
# By Fitzpatrick type
by_fitz: dict[str, dict[str, Any]] = {}
for case in self.cases:
ft = case.fitzpatrick_type or "Unknown"
if ft not in by_fitz:
by_fitz[ft] = {"total": 0, "passed": 0, "id_sims": []}
by_fitz[ft]["total"] += 1
if case.safety_passed:
by_fitz[ft]["passed"] += 1
if case.identity_sim > 0:
by_fitz[ft]["id_sims"].append(case.identity_sim)
for _ft, stats in by_fitz.items():
stats["pass_rate"] = stats["passed"] / max(stats["total"], 1)
stats["mean_identity_sim"] = (
sum(stats["id_sims"]) / len(stats["id_sims"])
if stats["id_sims"]
else 0.0
)
del stats["id_sims"]
return AuditSummary(
total_cases=total,
passed_cases=passed,
failed_cases=failed,
flagged_cases=flagged,
pass_rate=passed / total,
mean_identity_sim=mean_id,
by_procedure=by_proc,
by_fitzpatrick=by_fitz,
)
def flagged_cases(self) -> list[AuditCase]:
"""Return cases that need manual review (failed or have warnings)."""
return [c for c in self.cases if not c.safety_passed or c.warnings]
def to_json(self) -> str:
"""Export audit data as JSON."""
summary = self.compute_summary()
data = {
"report_title": self.report_title,
"model_version": self.model_version,
"generated_at": datetime.now(timezone.utc).isoformat(),
"summary": {
"total_cases": summary.total_cases,
"passed_cases": summary.passed_cases,
"failed_cases": summary.failed_cases,
"flagged_cases": summary.flagged_cases,
"pass_rate": round(summary.pass_rate, 4),
"mean_identity_sim": round(summary.mean_identity_sim, 4),
},
"by_procedure": {
k: {kk: round(vv, 4) if isinstance(vv, float) else vv for kk, vv in v.items()}
for k, v in summary.by_procedure.items()
},
"by_fitzpatrick": {
k: {kk: round(vv, 4) if isinstance(vv, float) else vv for kk, vv in v.items()}
for k, v in summary.by_fitzpatrick.items()
},
"cases": [
{
"case_id": c.case_id,
"procedure": c.procedure,
"safety_passed": c.safety_passed,
"identity_sim": round(c.identity_sim, 4),
"intensity": c.intensity,
"fitzpatrick_type": c.fitzpatrick_type,
"warnings": c.warnings,
"failures": c.failures,
"metrics": {k: round(v, 4) for k, v in c.metrics.items()},
"timestamp": c.timestamp,
}
for c in self.cases
],
}
return json.dumps(data, indent=2)
def generate_report(self, output_path: str | Path) -> Path:
"""Generate an HTML audit report.
Args:
output_path: Path to save the HTML report.
Returns:
Path to the generated report.
"""
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
summary = self.compute_summary()
html = self._render_html(summary)
output_path.write_text(html)
return output_path
def _render_html(self, summary: AuditSummary) -> str:
"""Render the audit report as HTML."""
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
status = "PASS" if summary.failed_cases == 0 else "REQUIRES REVIEW"
status_color = "#28a745" if summary.failed_cases == 0 else "#dc3545"
# Build procedure rows
proc_rows = ""
for proc, stats in sorted(summary.by_procedure.items()):
rate = stats["pass_rate"]
rate_color = "#28a745" if rate >= 0.95 else "#ffc107" if rate >= 0.8 else "#dc3545"
proc_rows += (
f"
"
f"
{proc.title()}
"
f"
{stats['total']}
"
f"
{stats['passed']}
"
f'
{rate:.1%}
'
f"
{stats['mean_identity_sim']:.4f}
"
f"
\n"
)
# Build Fitzpatrick rows
fitz_rows = ""
for ft, stats in sorted(summary.by_fitzpatrick.items()):
rate = stats["pass_rate"]
rate_color = "#28a745" if rate >= 0.95 else "#ffc107" if rate >= 0.8 else "#dc3545"
fitz_rows += (
f"
"
f"
{ft}
"
f"
{stats['total']}
"
f"
{stats['passed']}
"
f'
{rate:.1%}
'
f"
{stats['mean_identity_sim']:.4f}
"
f"
\n"
)
# Build flagged cases
flagged = self.flagged_cases()
flagged_rows = ""
for c in flagged:
issues = "; ".join(c.failures + [f"WARN: {w}" for w in c.warnings])
bg = "#fff3cd" if c.safety_passed else "#f8d7da"
flagged_rows += (
f'
'
f"
{c.case_id}
"
f"
{c.procedure.title()}
"
f"
{c.fitzpatrick_type}
"
f"
{c.identity_sim:.4f}
"
f'
{"WARN" if c.safety_passed else "FAIL"}
'
f"
{issues}
"
f"
\n"
)
return f"""
{self.report_title}
{self.report_title}
Generated: {now} | Model version: {self.model_version}
| Overall status: {status}
Disclaimer: This report is for research and development purposes only.
LandmarkDiff predictions are AI-generated visualizations and do not constitute medical advice
or guarantee surgical outcomes. All predictions should be reviewed by qualified clinical professionals.