---
license: apache-2.0
language:
- en
- zh
pipeline_tag: robotics
---
#!/usr/bin/env python3
# forensic_oil_complete_reform.py
# 完整改良版单文件：法证级多模态油脂包浆鉴定（一次性全面改革）
# 目标：一次性解决之前所有失败与不足，提供工程化、可测试、可部署的单文件实现。
# 说明：
# - 生产必须替换 TODO: REPLACE 的厂商解析器与提供真实 CS-14C 标定数据
# - 请通过 KMS/Secret 管理 PROVENANCE_SECRET，不要在环境中明文存放
# - 运行示例：
# python forensic_oil_complete_reform.py --mode build-dummy
# python forensic_oil_complete_reform.py --mode train --manifest train_manifest.json --model-dir models
# python forensic_oil_complete_reform.py --mode infer-file --sample-file sample.json --model-dir models
# python forensic_oil_complete_reform.py --mode auto
# uvicorn forensic_oil_complete_reform:app --host 0.0.0.0 --port 8000
import os
import io
import sys
import json
import time
import uuid
import math
import hmac
import yaml
import base64
import hashlib
import logging
import datetime
from typing import Dict, Any, List, Tuple, Optional
from threading import Thread, Event
from queue import Queue, Empty
from contextlib import asynccontextmanager
# numpy compatibility
try:
import numpy as np
if not hasattr(np, 'bool'):
np.bool = np.bool_
except Exception:
raise
import pandas as pd
from scipy import signal, optimize, stats
from scipy.ndimage import gaussian_filter
from sklearn.isotonic import IsotonicRegression
from sklearn.metrics import roc_auc_score, precision_recall_fscore_support
from sklearn.model_selection import train_test_split
import xgboost as xgb
# Optional heavy deps (graceful fallback)
try:
import pymzml
except Exception:
pymzml = None
try:
import tifffile
except Exception:
tifffile = None
try:
import pymc3 as pm
import arviz as az
except Exception:
pm = None
try:
import shap
except Exception:
shap = None
try:
from PIL import Image, UnidentifiedImageError
except Exception:
Image = None
UnidentifiedImageError = Exception
# Web API optional
try:
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel
except Exception:
FastAPI = None
# Logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger("forensic_oil_complete_reform")
# ----------------------------
# Default configuration
# ----------------------------
DEFAULT_CONFIG = {
"model_version": "5.0.0-complete-reform",
"provenance_secret_env": "PROVENANCE_SECRET",
"audit_dir": "audit_logs",
"model_dir": "models",
"incoming_dir": "incoming_samples",
"min_sample_mass_mg": 0.2,
"gcms": {
"binned_bins": 1024,
"marker_windows": [[270,290],[300,320],[340,360],[400,420]],
"topk": 256,
"peak_prominence": 1e3,
"peak_width": (1, 200)
},
"spectrum": {"binned_bins": 512, "marker_peaks": [1700,1740,2850,2920,1650]},
"image": {"patch_size": 256, "hist_bins": 128},
"afm": {"features": ["max_force","min_force","adhesion","slope","auc"]},
"fusion": {"confidence_high": 0.92, "confidence_low": 0.35},
"training": {"xgb_rounds": 200, "random_seed": 42},
"blind_test": {"n_min": 200, "auc_target": 0.95, "fpr_target": 0.05},
"bayes": {"mcmc_draws": 1000, "mcmc_tune": 500},
"auto": {
"watch_dir": "incoming_samples",
"poll_interval_s": 5,
"max_workers": 2,
"human_review_confidence_threshold": 0.75,
"auto_accept_confidence": 0.92,
"max_queue_size": 1000
}
}
# ----------------------------
# Utilities
# ----------------------------
def now_iso() -> str:
return datetime.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
def ensure_dir(p: str):
os.makedirs(p, exist_ok=True)
def canonical_json(obj: Any) -> str:
return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(',',':'))
def compute_provenance(meta: Dict[str, Any], files_bytes: List[bytes], secret: str) -> Dict[str,str]:
h = hashlib.sha256()
h.update(canonical_json(meta).encode('utf-8'))
for fb in files_bytes:
if fb:
h.update(fb)
digest = h.hexdigest()
mac = hmac.new(secret.encode('utf-8'), digest.encode('utf-8'), hashlib.sha256).hexdigest()
return {"hash": digest, "hmac": mac}
# ----------------------------
# Error codes & helpers
# ----------------------------
ERROR_CODES = {
"PARSE_GCMS_FAIL": "E1001",
"PARSE_SPECTRUM_FAIL": "E1002",
"PARSE_IMAGE_FAIL": "E1003",
"PARSE_AFM_FAIL": "E1004",
"MISSING_MODALITY": "E2001",
"MODEL_LOAD_FAIL": "E3001",
"INFERENCE_FAIL": "E3002"
}
def audit_error(audit_dir: str, sample_id: str, code: str, message: str, raw_bytes: Optional[bytes]=None):
ensure_dir(audit_dir)
rec = {"sample_id": sample_id, "error_code": code, "message": message, "timestamp": now_iso()}
if raw_bytes:
rec['raw_hash'] = hashlib.sha256(raw_bytes).hexdigest()
fname = os.path.join(audit_dir, f"error_{sample_id}_{int(time.time())}_{uuid.uuid4().hex[:6]}.json")
with open(fname, 'w', encoding='utf-8') as f:
json.dump(rec, f, ensure_ascii=False, indent=2)
logger.debug("Audit error written: %s", fname)
# ----------------------------
# Robust parsing utilities (comprehensive)
# - Handles placeholders, single-column CSVs, non-image bytes, and tries multiple fallbacks
# ----------------------------
PLACEHOLDER_MARKERS = {b"mzml_placeholder", b"spectrum_placeholder", b"tiff_placeholder", b"afm_placeholder", b"cs14c_placeholder", b"tiff", b"image", b"placeholder"}
def is_placeholder(b: Optional[bytes]) -> bool:
if not b:
return True
if isinstance(b, bytes) and len(b) < 1024:
low = b.strip().lower()
for m in PLACEHOLDER_MARKERS:
if m in low:
return True
return False
def try_read_csv_bytes(b: bytes, sep_guesses=[',','\t',';',' ']) -> pd.DataFrame:
s = b.decode('utf-8', errors='ignore')
# quick sniff: if looks like JSON, raise
if s.strip().startswith('{') or s.strip().startswith('['):
raise ValueError("Not CSV")
# try pandas default
try:
df = pd.read_csv(io.StringIO(s))
if df.shape[1] >= 2:
return df
except Exception:
pass
# try guesses
for sep in sep_guesses:
try:
df = pd.read_csv(io.StringIO(s), sep=sep)
if df.shape[1] >= 2:
return df
except Exception:
continue
# try single-column splitting heuristics
lines = [ln.strip() for ln in s.splitlines() if ln.strip()]
rows = []
for ln in lines:
parts = ln.replace(',', ' ').replace('\t',' ').split()
if len(parts) >= 2:
rows.append(parts[:2])
if rows:
df = pd.DataFrame(rows)
return df
# fallback: raise
raise ValueError("CSV parse failed or single-column without numeric pairs")
# ----------------------------
# Parsers (robust, with synthetic fallback)
# ----------------------------
def parse_mzml_precise(mzml_bytes: Optional[bytes]) -> Dict[str, Any]:
"""
Return {'peaks': DataFrame with columns ['mz','intensity','rt'], 'meta': {...}, 'note':...}
Robust behavior:
- If placeholder or None -> return empty peaks and note 'placeholder'
- If pymzml available and valid mzML -> parse
- Else try CSV heuristics
- On failure return empty peaks and audit error
"""
meta = {}
if is_placeholder(mzml_bytes):
return {'peaks': pd.DataFrame(columns=['mz','intensity','rt']), 'meta': {'parser':'placeholder'}, 'note':'placeholder'}
if pymzml is not None:
try:
with io.BytesIO(mzml_bytes) as bio:
run = pymzml.run.Reader(bio)
peaks = []
for spec in run:
if getattr(spec, 'ms_level', 1) == 1:
mzs = np.array(getattr(spec, 'mz', []))
ints = np.array(getattr(spec, 'i', []))
try:
rt = spec.scan_time_in_minutes()
except Exception:
rt = np.nan
if len(mzs) == 0:
continue
for m, it in zip(mzs, ints):
peaks.append((float(m), float(it), float(rt)))
if len(peaks) == 0:
raise ValueError("No MS1 peaks found")
df = pd.DataFrame(peaks, columns=['mz','intensity','rt'])
return {'peaks': df, 'meta': {'parser':'pymzml'}, 'note':'pymzml'}
except Exception as e:
logger.warning("pymzml parse failed: %s. Will try CSV fallback.", e)
# CSV fallback
try:
df = try_read_csv_bytes(mzml_bytes)
# normalize column names
cols = [c.lower() for c in df.columns]
if 'mz' in cols and 'intensity' in cols:
df2 = df.copy()
df2.columns = cols
if 'rt' not in df2.columns:
df2['rt'] = np.nan
return {'peaks': df2[['mz','intensity','rt']].copy(), 'meta': {'parser':'csv_fallback'}, 'note':'csv_fallback'}
# if first two columns numeric, treat as mz,intensity
numeric_cols = []
for c in df.columns[:2]:
try:
pd.to_numeric(df[c])
numeric_cols.append(c)
except Exception:
pass
if len(numeric_cols) >= 2:
df2 = df.copy()
df2.columns = ['mz','intensity'] + list(df.columns[2:])
if 'rt' not in df2.columns:
df2['rt'] = np.nan
return {'peaks': df2[['mz','intensity','rt']].copy(), 'meta': {'parser':'csv_guess'}, 'note':'csv_guess'}
except Exception as e:
logger.exception("parse_mzml_precise failed: %s", e)
# final fallback: empty
return {'peaks': pd.DataFrame(columns=['mz','intensity','rt']), 'meta': {'parser':'failed'}, 'note':'failed'}
def parse_jcamp_precise(b: Optional[bytes]) -> Dict[str, Any]:
if is_placeholder(b):
return {'wn': np.array([], dtype=np.float32), 'intensity': np.array([], dtype=np.float32), 'meta': {'parser':'placeholder'}, 'note':'placeholder'}
s = b.decode('utf-8', errors='ignore')
if '##' in s and 'JCAMP' in s.upper():
header = {}
data = []
in_xy = False
for ln in s.splitlines():
ln = ln.strip()
if ln.startswith('##'):
parts = ln[2:].split('=',1)
if len(parts) == 2:
header[parts[0].strip()] = parts[1].strip()
if ln.upper().startswith('##XYDATA'):
in_xy = True
continue
if in_xy:
for token in ln.split():
try:
data.append(float(token))
except:
pass
if len(data) >= 4 and len(data) % 2 == 0:
wn = np.array(data[0::2], dtype=np.float32)
inten = np.array(data[1::2], dtype=np.float32)
return {'wn': wn, 'intensity': inten, 'meta': header, 'note':'jcamp'}
# CSV fallback
try:
df = try_read_csv_bytes(b)
if df.shape[1] >= 2:
wn = pd.to_numeric(df.iloc[:,0], errors='coerce').fillna(0).values.astype(np.float32)
inten = pd.to_numeric(df.iloc[:,1], errors='coerce').fillna(0).values.astype(np.float32)
return {'wn': wn, 'intensity': inten, 'meta': {'parser':'csv'}, 'note':'csv'}
except Exception as e:
logger.exception("parse_jcamp_precise failed: %s", e)
return {'wn': np.array([], dtype=np.float32), 'intensity': np.array([], dtype=np.float32), 'meta': {'parser':'failed'}, 'note':'failed'}
def parse_sem_tiff_precise(b: Optional[bytes]) -> Tuple[np.ndarray, Dict[str,Any]]:
meta = {}
if is_placeholder(b):
# return synthetic blank patch
arr = np.zeros((DEFAULT_CONFIG['image']['patch_size'], DEFAULT_CONFIG['image']['patch_size']), dtype=np.uint8)
return arr, {'parser':'placeholder'}
if tifffile is not None:
try:
with io.BytesIO(b) as bio:
tf = tifffile.TiffFile(bio)
page = tf.pages[0]
arr = page.asarray()
tags = {}
try:
for k,v in page.tags.items():
tags[k] = str(v.value)
meta['tiff_tags'] = tags
desc = tags.get('ImageDescription','')
if isinstance(desc, str) and 'nm' in desc:
import re
m = re.search(r'([\d\.]+)\s*nm', desc)
if m:
meta['pixel_size_nm'] = float(m.group(1))
except Exception:
pass
return arr, meta
except Exception as e:
logger.warning("tifffile parse failed: %s. Trying PIL fallback.", e)
# PIL fallback
try:
if Image is None:
raise RuntimeError("PIL not available")
with io.BytesIO(b) as bio:
img = Image.open(bio)
arr = np.array(img)
return arr, meta
except UnidentifiedImageError as e:
logger.warning("PIL cannot identify image: %s", e)
except Exception as e:
logger.exception("parse_sem_tiff_precise failed: %s", e)
# final fallback: synthetic blank
arr = np.zeros((DEFAULT_CONFIG['image']['patch_size'], DEFAULT_CONFIG['image']['patch_size']), dtype=np.uint8)
return arr, {'parser':'failed_fallback'}
def parse_afm_precise(b: Optional[bytes]) -> Dict[str,Any]:
if is_placeholder(b):
return {'distance': np.array([], dtype=np.float32), 'force': np.array([], dtype=np.float32), 'meta': {'parser':'placeholder'}}
s = b.decode('utf-8', errors='ignore')
try:
df = try_read_csv_bytes(b)
# heuristics: find numeric columns
numeric_cols = []
for c in df.columns:
try:
pd.to_numeric(df[c])
numeric_cols.append(c)
except Exception:
pass
if len(numeric_cols) >= 2:
dist_col = numeric_cols[0]; force_col = numeric_cols[1]
return {'distance': pd.to_numeric(df[dist_col], errors='coerce').fillna(0).values.astype(np.float32),
'force': pd.to_numeric(df[force_col], errors='coerce').fillna(0).values.astype(np.float32),
'meta': {'parser':'csv'}}
# if only one numeric column, assume it's force and synthesize distance
if len(numeric_cols) == 1:
force = pd.to_numeric(df[numeric_cols[0]], errors='coerce').fillna(0).values.astype(np.float32)
distance = np.linspace(0.0, 1.0, len(force)).astype(np.float32)
return {'distance': distance, 'force': force, 'meta': {'parser':'csv_singlecol_synth_distance'}}
except Exception as e:
logger.exception("parse_afm_precise failed: %s", e)
return {'distance': np.array([], dtype=np.float32), 'force': np.array([], dtype=np.float32), 'meta': {'parser':'failed'}}
def parse_cs14c_precise(b: Optional[bytes]) -> Dict[str,Any]:
if is_placeholder(b):
return {'records': [], 'meta': {'parser':'placeholder'}}
try:
df = try_read_csv_bytes(b)
cols = [c.lower() for c in df.columns]
required = ['sample_id','compound_id','age_bp','age_sigma']
if all(r in cols for r in required):
df.columns = cols
return {'records': df.to_dict(orient='records'), 'meta': {'parser':'csv'}}
# try to map common variants
mapping = {}
for r in required:
for c in cols:
if r in c:
mapping[r] = c
break
if len(mapping) == len(required):
df2 = df.rename(columns={v:k for k,v in mapping.items()})
return {'records': df2.to_dict(orient='records'), 'meta': {'parser':'csv_mapped'}}
except Exception as e:
logger.exception("parse_cs14c_precise failed: %s", e)
return {'records': [], 'meta': {'parser':'failed'}}
# ----------------------------
# Signal processing & feature helpers (unchanged but robust)
# ----------------------------
def detect_peaks(mz: np.ndarray, inten: np.ndarray, prominence: float, width: Tuple[float,float]) -> np.ndarray:
if len(mz) < 3:
return np.array([], dtype=int)
try:
inten_s = signal.savgol_filter(inten, window_length=11 if len(inten)>=11 else max(3,len(inten)//2*2+1), polyorder=2)
peaks_idx, props = signal.find_peaks(inten_s, prominence=prominence, width=width)
return peaks_idx
except Exception:
return np.array([], dtype=int)
def fit_gaussian_peak(x: np.ndarray, y: np.ndarray) -> Tuple[float,float,float,float]:
if len(x) < 3:
return (float(np.mean(x)) if len(x)>0 else 0.0, 0.0, 0.0, float(np.max(y)) if len(y)>0 else 0.0)
def gauss(x, A, mu, sigma):
return A * np.exp(-0.5*((x-mu)/sigma)**2)
p0 = [float(np.max(y)), float(x[np.argmax(y)]), max(1e-6, float((x.max()-x.min())/6.0))]
try:
popt, _ = optimize.curve_fit(gauss, x, y, p0=p0, maxfev=5000)
A, mu, sigma = popt
area = A * sigma * math.sqrt(2*math.pi)
height = float(A)
return float(mu), float(sigma), float(area), float(height)
except Exception:
return (float(np.mean(x)) if len(x)>0 else 0.0, 0.0, float(np.trapz(y,x)) if len(x)>0 else 0.0, float(np.max(y)) if len(y)>0 else 0.0)
def bin_generic(x: np.ndarray, y: np.ndarray, bins: int = 256, xmin: float = None, xmax: float = None) -> np.ndarray:
if len(x) == 0:
return np.zeros(bins, dtype=np.float32)
if xmin is None: xmin = float(np.min(x))
if xmax is None: xmax = float(np.max(x))
if xmin == xmax:
return np.zeros(bins, dtype=np.float32)
edges = np.linspace(xmin, xmax, bins+1)
binned = np.zeros(bins, dtype=np.float32)
idx = np.digitize(x, edges) - 1
for i in range(len(x)):
j = min(max(int(idx[i]), 0), bins-1)
binned[j] += float(y[i])
s = binned.sum()
if s > 0:
binned = binned / s
return binned
def asls_baseline(intensity: np.ndarray, lam: float = 1e5, p: float = 0.01, niter: int = 10) -> np.ndarray:
L = len(intensity)
if L < 3:
return intensity
D = np.diff(np.eye(L), 2)
w = np.ones(L)
for i in range(niter):
W = np.diag(w)
Z = W + lam * D.T.dot(D)
z = np.linalg.solve(Z, w * intensity)
w = p * (intensity > z) + (1-p) * (intensity < z)
corrected = intensity - z
corrected[corrected < 0] = 0.0
return corrected
# ----------------------------
# Image & AFM quantification (robust)
# ----------------------------
def pixel_size_from_meta(meta: Dict[str,Any]) -> Optional[float]:
if not meta:
return None
if 'pixel_size_nm' in meta:
try:
return float(meta['pixel_size_nm'])
except Exception:
pass
tags = meta.get('tiff_tags', {})
for k,v in tags.items():
if isinstance(v, str) and 'nm' in v:
import re
m = re.search(r'([\d\.]+)\s*nm', v)
if m:
try:
return float(m.group(1))
except:
pass
return None
def penetration_proxy(img: np.ndarray) -> float:
if img is None or img.size == 0:
return 0.0
if img.ndim == 3:
gray = np.mean(img.astype(np.float32), axis=2)
else:
gray = img.astype(np.float32)
gray = gray - gray.min()
if gray.max() > 0:
gray = gray / gray.max()
h, w = gray.shape[:2]
cy, cx = h//2, w//2
ys, xs = np.indices((h,w))
r = np.sqrt((xs-cx)**2 + (ys-cy)**2).astype(np.int32)
maxr = min(cx, cy)
profile = [float(np.mean(gray[r==ri])) if np.any(r==ri) else 0.0 for ri in range(maxr)]
if len(profile) < 5:
return 0.0
prof = np.array(profile)
x = np.arange(len(prof))
slope, intercept, r_val, p_val, std_err = stats.linregress(x, prof)
return float(-slope)
def estimate_penetration_nm(img: np.ndarray, meta: Dict[str,Any], calibration_curve: Optional[Tuple[float,float,float]] = None) -> Tuple[float,float]:
proxy = penetration_proxy(img)
pixel_nm = pixel_size_from_meta(meta)
if calibration_curve:
slope, intercept, sigma = calibration_curve
penetration_nm = slope * proxy + intercept
uncertainty = abs(sigma) + 0.05 * penetration_nm
return float(max(0.0, penetration_nm)), float(uncertainty)
elif pixel_nm:
h, w = img.shape[:2]
mean_dim_nm = ((h + w) / 2.0) * pixel_nm
penetration_nm = proxy * mean_dim_nm
uncertainty = 0.2 * penetration_nm
return float(max(0.0, penetration_nm)), float(uncertainty)
else:
return float(proxy), float(proxy * 0.5 + 1.0)
def image_features(img: np.ndarray, meta: Dict[str,Any], cfg: Dict[str,Any]) -> np.ndarray:
if img is None or img.size == 0:
return np.zeros(cfg['hist_bins'] + 8, dtype=np.float32)
if img.ndim == 3:
arr = np.mean(img.astype(np.float32), axis=2)
else:
arr = img.astype(np.float32)
arr = arr - arr.min()
if arr.max() > 0:
arr = arr / arr.max()
target = (cfg['patch_size'], cfg['patch_size'])
try:
pil = Image.fromarray((arr*255).astype(np.uint8))
pil = pil.resize(target, Image.BILINEAR)
arr2 = np.array(pil).astype(np.float32)/255.0
except Exception:
arr2 = gaussian_filter(arr, sigma=1.0)
arr2 = arr2[:target[0], :target[1]]
if arr2.shape != target:
arr2 = np.resize(arr2, target)
mean = float(arr2.mean())
std = float(arr2.std())
grad_x = float(np.mean(np.abs(np.diff(arr2, axis=1))))
grad_y = float(np.mean(np.abs(np.diff(arr2, axis=0))))
hist, _ = np.histogram(arr2, bins=cfg['hist_bins'], range=(0,1))
hist = hist.astype(np.float32) / (hist.sum()+1e-9)
tail_ratio = float(np.sum(arr2 < 0.1) / (arr2.size + 1e-9))
penetration_nm, penetration_unc = estimate_penetration_nm(img, meta, calibration_curve=None)
return np.concatenate([[mean, std, grad_x, grad_y, tail_ratio, penetration_nm, penetration_unc], hist])
def calibrate_afm(force_raw: np.ndarray, probe_k_N_per_m: Optional[float], sensitivity_nm_per_V: Optional[float]) -> np.ndarray:
if probe_k_N_per_m and sensitivity_nm_per_V:
deflection_nm = force_raw * sensitivity_nm_per_V
force_N = probe_k_N_per_m * (deflection_nm * 1e-9)
return force_N * 1e9 # nN
else:
return force_raw
def afm_features(distance: np.ndarray, force_raw: np.ndarray, probe_k: Optional[float], sensitivity: Optional[float]) -> Tuple[np.ndarray, Dict[str,Any]]:
if distance is None or force_raw is None or len(force_raw)==0:
return np.zeros(5, dtype=np.float32), {'probe_k': probe_k, 'sensitivity': sensitivity}
force_nN = calibrate_afm(force_raw, probe_k, sensitivity)
meta = {'probe_k': probe_k, 'sensitivity': sensitivity}
if len(force_nN) < 3:
return np.zeros(5, dtype=np.float32), meta
max_f = float(np.max(force_nN))
min_f = float(np.min(force_nN))
adhesion = float(-min_f) if min_f < 0 else 0.0
slope = float((force_nN[-1] - force_nN[0]) / (distance[-1] - distance[0] + 1e-9))
auc = float(np.trapz(force_nN, distance))
return np.array([max_f, min_f, adhesion, slope, auc], dtype=np.float32), meta
# ----------------------------
# Dataset & feature assembly (robust with synthetic minimal features)
# ----------------------------
class MultimodalSample:
def __init__(self, sample_id: str, meta: Dict[str,Any],
gcms_bytes: Optional[bytes]=None,
spectrum_bytes: Optional[bytes]=None,
image_bytes: Optional[bytes]=None,
afm_bytes: Optional[bytes]=None,
cs14c_bytes: Optional[bytes]=None,
label: Optional[int]=None):
self.sample_id = sample_id
self.meta = meta or {}
self.gcms_bytes = gcms_bytes
self.spectrum_bytes = spectrum_bytes
self.image_bytes = image_bytes
self.afm_bytes = afm_bytes
self.cs14c_bytes = cs14c_bytes
self.label = label
class MultimodalDataset:
def __init__(self, samples: List[MultimodalSample], cfg: Dict[str,Any]):
self.samples = samples
self.cfg = cfg
def __len__(self):
return len(self.samples)
def __getitem__(self, idx):
s = self.samples[idx]
feats = {}
# GC-MS
try:
parsed = parse_mzml_precise(s.gcms_bytes)
peaks = parsed.get('peaks', pd.DataFrame(columns=['mz','intensity','rt']))
feats['gcms'] = extract_gcms_features(peaks, self.cfg['gcms'])
except Exception as e:
logger.exception("GCMS error %s: %s", s.sample_id, e)
audit_error(self.cfg.get('audit_dir','audit_logs'), s.sample_id, ERROR_CODES['PARSE_GCMS_FAIL'], str(e), s.gcms_bytes)
feats['gcms'] = synthesize_gcms_minimal(self.cfg['gcms'])
# Spectrum
try:
sp = parse_jcamp_precise(s.spectrum_bytes)
feats['spectrum'] = spectrum_features(sp.get('wn', np.array([])), sp.get('intensity', np.array([])), self.cfg['spectrum']['marker_peaks'], self.cfg['spectrum']['binned_bins'])
except Exception as e:
logger.exception("Spectrum error %s: %s", s.sample_id, e)
audit_error(self.cfg.get('audit_dir','audit_logs'), s.sample_id, ERROR_CODES['PARSE_SPECTRUM_FAIL'], str(e), s.spectrum_bytes)
feats['spectrum'] = synthesize_spectrum_minimal(self.cfg['spectrum'])
# Image
try:
img, img_meta = parse_sem_tiff_precise(s.image_bytes)
feats['image'] = image_features(img, img_meta, self.cfg['image'])
except Exception as e:
logger.exception("Image error %s: %s", s.sample_id, e)
audit_error(self.cfg.get('audit_dir','audit_logs'), s.sample_id, ERROR_CODES['PARSE_IMAGE_FAIL'], str(e), s.image_bytes)
feats['image'] = synthesize_image_minimal(self.cfg['image'])
# AFM
try:
afm = parse_afm_precise(s.afm_bytes)
afm_feat, afm_meta = afm_features(afm.get('distance', np.array([])), afm.get('force', np.array([])), s.meta.get('afm_probe_k_N_per_m'), s.meta.get('afm_sensitivity_nm_per_V'))
feats['afm'] = afm_feat
except Exception as e:
logger.exception("AFM error %s: %s", s.sample_id, e)
audit_error(self.cfg.get('audit_dir','audit_logs'), s.sample_id, ERROR_CODES['PARSE_AFM_FAIL'], str(e), s.afm_bytes)
feats['afm'] = synthesize_afm_minimal()
# meta features
env = s.meta.get('environment', {})
temp = float(env.get('temperature_c', 20.0))
rh = float(env.get('relative_humidity', 40.0))
material = s.meta.get('object_material', 'unknown')
material_code = 0.0
if isinstance(material, str):
if material.lower() in ['ceramic','pottery','porcelain']:
material_code = 1.0
elif material.lower() in ['bronze','metal']:
material_code = 2.0
meta_feat = np.array([temp, rh, material_code], dtype=np.float32)
x = np.concatenate([feats['gcms'], feats['spectrum'], feats['image'], feats['afm'], meta_feat])
y = np.array([s.label], dtype=np.int64) if s.label is not None else np.array([-1], dtype=np.int64)
return {'x': x, 'y': y, 'sample_id': s.sample_id, 'meta': s.meta}
# ----------------------------
# Synthetic minimal feature generators (when modality missing)
# ----------------------------
def synthesize_gcms_minimal(cfg: Dict[str,Any]) -> np.ndarray:
dim = cfg['binned_bins'] + len(cfg.get('marker_windows',[])) + cfg.get('topk',256)
# small random noise stable seed
return np.zeros(dim, dtype=np.float32)
def synthesize_spectrum_minimal(cfg: Dict[str,Any]) -> np.ndarray:
dim = cfg['binned_bins'] + len(cfg.get('marker_peaks',[])) + 1
return np.zeros(dim, dtype=np.float32)
def synthesize_image_minimal(cfg: Dict[str,Any]) -> np.ndarray:
dim = cfg['hist_bins'] + 8
return np.zeros(dim, dtype=np.float32)
def synthesize_afm_minimal() -> np.ndarray:
return np.zeros(5, dtype=np.float32)
# ----------------------------
# Spectrum features helper
# ----------------------------
def spectrum_features(wn: np.ndarray, inten: np.ndarray, marker_peaks: List[float], bins: int) -> np.ndarray:
if len(wn) == 0 or len(inten) == 0:
return synthesize_spectrum_minimal({'binned_bins': bins, 'marker_peaks': marker_peaks})
inten_corr = asls_baseline(inten)
binned = bin_generic(wn, inten_corr, bins=bins, xmin=wn.min(), xmax=wn.max())
markers = []
for pk in marker_peaks:
idx = np.argmin(np.abs(wn - pk))
markers.append(float(inten_corr[idx]) / (inten_corr.sum()+1e-9))
peak_count = float(np.sum(inten_corr > (np.mean(inten_corr) + 2*np.std(inten_corr))))
return np.concatenate([binned, np.array(markers, dtype=np.float32), np.array([peak_count], dtype=np.float32)])
# ----------------------------
# GC-MS feature extraction (precise)
# ----------------------------
def extract_gcms_features(peaks_df: pd.DataFrame, cfg: Dict[str,Any]) -> np.ndarray:
if peaks_df is None or len(peaks_df)==0:
return synthesize_gcms_minimal(cfg)
try:
mz = peaks_df['mz'].values.astype(np.float64)
inten = peaks_df['intensity'].values.astype(np.float64)
except Exception:
return synthesize_gcms_minimal(cfg)
binned = bin_generic(mz, inten, bins=cfg['binned_bins'], xmin=mz.min(), xmax=mz.max())
marker_vals = []
for w in cfg.get('marker_windows', []):
lo, hi = w
mask = (mz >= lo) & (mz <= hi)
val = float(inten[mask].sum()) / (inten.sum()+1e-9)
marker_vals.append(val)
K = cfg.get('topk', 256)
try:
order = np.argsort(inten)[-K:] if len(inten) >= K else np.argsort(inten)
top_mz = np.sort(mz[order])
top_vec = np.zeros(K, dtype=np.float32)
for i, val in enumerate(top_mz[:K]):
top_vec[i] = float((val % 1000) / 1000.0)
except Exception:
top_vec = np.zeros(K, dtype=np.float32)
feat = np.concatenate([binned.astype(np.float32), np.array(marker_vals, dtype=np.float32), top_vec])
return feat
# ----------------------------
# Submodels (XGBoost)
# ----------------------------
class XGBModel:
def __init__(self, params: Dict[str,Any] = None):
self.params = params or {'objective':'binary:logistic','eval_metric':'auc','use_label_encoder':False}
self.model = None
def train(self, X: np.ndarray, y: np.ndarray, num_round: int = 200):
dtrain = xgb.DMatrix(X, label=y)
self.model = xgb.train(self.params, dtrain, num_round)
def predict(self, X: np.ndarray) -> np.ndarray:
if self.model is None:
raise RuntimeError("Model not trained")
d = xgb.DMatrix(X)
return self.model.predict(d)
def save(self, path: str):
ensure_dir(os.path.dirname(path) or '.')
self.model.save_model(path)
def load(self, path: str):
self.model = xgb.Booster()
self.model.load_model(path)
# ----------------------------
# Fusion & Bayesian age posterior (MCMC template + conjugate fallback)
# ----------------------------
def fusion_posterior(ps: Dict[str,float], weights: Dict[str,float], prior: float = 0.5) -> float:
eps = 1e-9
def logit(p): return math.log((p+eps)/(1-p+eps))
logits = np.array([logit(ps.get(k,0.5)) for k in ['gcms','image','spectrum','afm']])
w = np.array([weights.get(k,0.25) for k in ['gcms','image','spectrum','afm']])
combined = np.dot(w, logits) + logit(prior)
posterior = 1.0 / (1.0 + math.exp(-combined))
return posterior
def bayesian_age_mcmc(prior_mu: float, prior_sigma: float, evidence: List[Tuple[float,float]], cs14c_records: Optional[List[Dict[str,Any]]] = None, draws: int = 1000, tune: int = 500):
if pm is None:
mu, sigma, ci = bayesian_age_conjugate((prior_mu, prior_sigma), evidence, cs14c_records)
return {'mu': mu, 'sigma': sigma, 'ci95': ci, 'method': 'conjugate_fallback'}
with pm.Model() as model:
age = pm.Normal('age', mu=prior_mu, sigma=prior_sigma)
for i, (mu_i, sigma_i) in enumerate(evidence):
pm.Normal(f'evidence_{i}', mu=age, sigma=max(1.0, sigma_i), observed=mu_i)
if cs14c_records:
for j, rec in enumerate(cs14c_records):
mu_c = float(rec.get('age_bp'))
sigma_c = float(rec.get('age_sigma', 20.0))
pm.Normal(f'cs14c_{j}', mu=age, sigma=max(1.0, sigma_c), observed=mu_c)
trace = pm.sample(draws=draws, tune=tune, chains=2, cores=1, return_inferencedata=True, progressbar=False)
summary = az.summary(trace, hdi_prob=0.95)
mu_post = float(summary.loc['age','mean'])
sd_post = float(summary.loc['age','sd'])
hdi = az.hdi(trace, var_names=['age'], hdi_prob=0.95)
ci95 = (float(hdi['age'][0]), float(hdi['age'][1]))
return {'mu': mu_post, 'sigma': sd_post, 'ci95': ci95, 'method': 'mcmc'}
def bayesian_age_conjugate(prior: Tuple[float,float], evidence: List[Tuple[float,float]], cs14c_records: Optional[List[Dict[str,Any]]] = None):
mu0, s0 = prior
var0 = s0**2
mu_n = mu0
var_n = var0
for mu_i, s_i in evidence:
var_i = max(1.0, s_i**2)
var_n = 1.0 / (1.0/var_n + 1.0/var_i)
mu_n = var_n * (mu_n/var_n + mu_i/var_i)
if cs14c_records:
for rec in cs14c_records:
mu_c = float(rec.get('age_bp'))
sigma_c = float(rec.get('age_sigma', 20.0))
var_c = sigma_c**2
var_n = 1.0 / (1.0/var_n + 1.0/var_c)
mu_n = var_n * (mu_n/var_n + mu_c/var_c)
sigma_n = math.sqrt(var_n)
ci95 = (mu_n - 1.96*sigma_n, mu_n + 1.96*sigma_n)
return mu_n, sigma_n, ci95
# ----------------------------
# Calibration & evaluation
# ----------------------------
def calibrate_probs(probs: np.ndarray, labels: np.ndarray) -> Dict[str,Any]:
iso = IsotonicRegression(out_of_bounds='clip')
try:
iso.fit(probs, labels)
return {'type':'isotonic','model':iso}
except Exception:
return {'type':'identity'}
def apply_calib(calib: Dict[str,Any], probs: np.ndarray) -> np.ndarray:
if calib is None or calib.get('type') == 'identity':
return probs
if calib.get('type') == 'isotonic':
iso = calib['model']
return iso.transform(probs)
return probs
# ----------------------------
# Training pipeline (robust)
# ----------------------------
def train_pipeline(samples: List[MultimodalSample], cfg: Dict[str,Any], out_dir: str):
ensure_dir(out_dir)
ds = MultimodalDataset(samples, cfg)
X = []
Y = []
for i in range(len(ds)):
item = ds[i]
X.append(item['x'])
Y.append(int(item['y'][0]))
X = np.stack(X).astype(np.float32)
Y = np.array(Y).astype(np.int64)
X_train, X_val, y_train, y_val = train_test_split(X, Y, test_size=0.2, stratify=Y, random_state=cfg['training']['random_seed'])
# modality dims
gcms_dim = cfg['gcms']['binned_bins'] + len(cfg['gcms'].get('marker_windows',[])) + cfg['gcms'].get('topk',256)
spec_dim = cfg['spectrum']['binned_bins'] + len(cfg['spectrum'].get('marker_peaks',[])) + 1
image_dim = cfg['image']['hist_bins'] + 8
afm_dim = 5
meta_dim = 3
idx = 0
gcms_idx = (idx, idx+gcms_dim); idx += gcms_dim
spec_idx = (idx, idx+spec_dim); idx += spec_dim
image_idx = (idx, idx+image_dim); idx += image_dim
afm_idx = (idx, idx+afm_dim); idx += afm_dim
meta_idx = (idx, idx+meta_dim); idx += meta_dim
# train submodels
model_g = XGBModel(); model_g.train(X_train[:, gcms_idx[0]:gcms_idx[1]], y_train, num_round=cfg['training']['xgb_rounds'])
model_s = XGBModel(); model_s.train(X_train[:, spec_idx[0]:spec_idx[1]], y_train, num_round=cfg['training']['xgb_rounds'])
model_i = XGBModel(); model_i.train(X_train[:, image_idx[0]:image_idx[1]], y_train, num_round=cfg['training']['xgb_rounds'])
model_a = XGBModel(); model_a.train(X_train[:, afm_idx[0]:afm_idx[1]], y_train, num_round=max(50, cfg['training']['xgb_rounds']//4))
# validation preds
pg = model_g.predict(X_val[:, gcms_idx[0]:gcms_idx[1]])
ps = model_s.predict(X_val[:, spec_idx[0]:spec_idx[1]])
pi = model_i.predict(X_val[:, image_idx[0]:image_idx[1]])
pa = model_a.predict(X_val[:, afm_idx[0]:afm_idx[1]])
# weights from AUCs
def auc_safe(y_true, y_score):
try:
return roc_auc_score(y_true, y_score)
except:
return 0.5
auc_g = auc_safe(y_val, pg)
auc_s = auc_safe(y_val, ps)
auc_i = auc_safe(y_val, pi)
auc_a = auc_safe(y_val, pa)
raw = np.array([max(0.01, auc_g), max(0.01, auc_i), max(0.01, auc_s), max(0.01, auc_a)])
weights = (raw / raw.sum()).tolist()
weights_dict = {'gcms': weights[0], 'image': weights[1], 'spectrum': weights[2], 'afm': weights[3]}
logger.info("Modality weights: %s", weights_dict)
fused = []
for i in range(len(pg)):
psd = {'gcms': float(pg[i]), 'image': float(pi[i]), 'spectrum': float(ps[i]), 'afm': float(pa[i])}
fused.append(fusion_posterior(psd, weights_dict, prior=0.5))
fused = np.array(fused)
calib = calibrate_probs(fused, y_val)
fused_cal = apply_calib(calib, fused)
auc_fused = auc_safe(y_val, fused_cal)
preds = (fused_cal >= 0.5).astype(int)
precision, recall, f1, _ = precision_recall_fscore_support(y_val, preds, average='binary', zero_division=0)
logger.info("Fused metrics: AUC=%.4f Precision=%.4f Recall=%.4f F1=%.4f", auc_fused, precision, recall, f1)
# save
ensure_dir(out_dir)
model_g.save(os.path.join(out_dir, "gcms.xgb"))
model_s.save(os.path.join(out_dir, "spectrum.xgb"))
model_i.save(os.path.join(out_dir, "image.xgb"))
model_a.save(os.path.join(out_dir, "afm.xgb"))
meta = {'weights': weights_dict, 'calibration': {'type': calib.get('type', 'identity')}, 'cfg': cfg, 'model_version': cfg.get('model_version')}
with open(os.path.join(out_dir, "fusion_meta.json"), "w", encoding='utf-8') as f:
json.dump(meta, f, ensure_ascii=False, indent=2)
import pickle
with open(os.path.join(out_dir, "calibration.pkl"), "wb") as f:
pickle.dump(calib, f)
logger.info("Training complete. Models saved to %s", out_dir)
return {'auc': auc_fused, 'precision': precision, 'recall': recall, 'f1': f1, 'weights': weights_dict}
# ----------------------------
# Inference engine (robust)
# ----------------------------
class InferenceEngine:
def __init__(self, model_dir: str, cfg: Dict[str,Any], secret: str):
self.model_dir = model_dir
self.cfg = cfg
self.secret = secret
self._load()
def _load(self):
self.model_g = XGBModel(); self.model_s = XGBModel(); self.model_i = XGBModel(); self.model_a = XGBModel()
try:
self.model_g.load(os.path.join(self.model_dir, "gcms.xgb"))
self.model_s.load(os.path.join(self.model_dir, "spectrum.xgb"))
self.model_i.load(os.path.join(self.model_dir, "image.xgb"))
self.model_a.load(os.path.join(self.model_dir, "afm.xgb"))
except Exception:
logger.warning("Model files not found; inference will use defaults")
try:
with open(os.path.join(self.model_dir, "fusion_meta.json"), "r", encoding='utf-8') as f:
meta = json.load(f); self.weights = meta.get('weights', {'gcms':0.4,'image':0.3,'spectrum':0.2,'afm':0.1})
except Exception:
self.weights = {'gcms':0.4,'image':0.3,'spectrum':0.2,'afm':0.1}
try:
import pickle
with open(os.path.join(self.model_dir, "calibration.pkl"), "rb") as f:
self.calib = pickle.load(f)
except Exception:
self.calib = {'type':'identity'}
logger.info("Engine loaded from %s", self.model_dir)
def infer(self, sample: MultimodalSample, return_explain: bool = False) -> Dict[str,Any]:
try:
ds = MultimodalDataset([sample], self.cfg)
item = ds[0]
x = item['x'].reshape(1,-1)
# indices
gcms_dim = self.cfg['gcms']['binned_bins'] + len(self.cfg['gcms'].get('marker_windows',[])) + self.cfg['gcms'].get('topk',256)
spec_dim = self.cfg['spectrum']['binned_bins'] + len(self.cfg['spectrum'].get('marker_peaks',[])) + 1
image_dim = self.cfg['image']['hist_bins'] + 8
afm_dim = 5
idx = 0
gcms_idx = (idx, idx+gcms_dim); idx += gcms_dim
spec_idx = (idx, idx+spec_dim); idx += spec_dim
image_idx = (idx, idx+image_dim); idx += image_dim
afm_idx = (idx, idx+afm_dim); idx += afm_dim
try:
p_g = float(self.model_g.predict(x[:, gcms_idx[0]:gcms_idx[1]])[0])
except Exception:
p_g = 0.5
try:
p_s = float(self.model_s.predict(x[:, spec_idx[0]:spec_idx[1]])[0])
except Exception:
p_s = 0.5
try:
p_i = float(self.model_i.predict(x[:, image_idx[0]:image_idx[1]])[0])
except Exception:
p_i = 0.5
try:
p_a = float(self.model_a.predict(x[:, afm_idx[0]:afm_idx[1]])[0])
except Exception:
p_a = 0.5
psd = {'gcms': p_g, 'image': p_i, 'spectrum': p_s, 'afm': p_a}
posterior = fusion_posterior(psd, self.weights, prior=0.5)
posterior_cal = apply_calib(self.calib, np.array([posterior]))[0]
verdict = "ancient_oil" if posterior_cal >= self.cfg['fusion']['confidence_high'] else ("modern_oil" if posterior_cal < self.cfg['fusion']['confidence_low'] else "inconclusive")
cs14c_recs = None
if sample.cs14c_bytes:
try:
cs14c = parse_cs14c_precise(sample.cs14c_bytes)
cs14c_recs = cs14c.get('records', [])
except Exception:
cs14c_recs = None
evidence_age = [] # placeholder for submodel age proxies
age_post = bayesian_age_mcmc(200.0, 200.0, evidence_age, cs14c_recs, draws=self.cfg['bayes']['mcmc_draws'], tune=self.cfg['bayes']['mcmc_tune']) if pm is not None else {'mu': None, 'sigma': None, 'ci95': (None,None), 'method':'conjugate_fallback'}
explain = {}
if shap is not None:
try:
explainer = shap.TreeExplainer(self.model_g.model)
shap_vals = explainer.shap_values(x[:, gcms_idx[0]:gcms_idx[1]])
explain['gcms_shap_mean_abs'] = float(np.mean(np.abs(shap_vals)))
except Exception:
explain['gcms_shap_mean_abs'] = None
else:
def logit(p): return math.log((p+1e-9)/(1-p+1e-9))
contrib = {k: float(self.weights.get(k,0.25) * abs(logit(psd.get(k,0.5)))) for k in psd}
explain['modality_contrib_proxy'] = contrib
files_bytes = [sample.gcms_bytes or b'', sample.spectrum_bytes or b'', sample.image_bytes or b'', sample.afm_bytes or b'', sample.cs14c_bytes or b'']
prov = compute_provenance(sample.meta, files_bytes, self.secret)
report = {
"sample_id": sample.sample_id,
"verdict": verdict,
"confidence": float(posterior_cal),
"age_posterior": age_post,
"evidence_probs": psd,
"explainability": explain,
"modality_weights": self.weights,
"model_version": self.cfg.get('model_version'),
"provenance": prov,
"timestamp": now_iso(),
"notes": "Algorithmic result. For high-stakes use, require CS-14C compound-specific dating and expert review."
}
audit_dir = self.cfg.get('audit_dir', DEFAULT_CONFIG['audit_dir'])
ensure_dir(audit_dir)
fname = os.path.join(audit_dir, f"{sample.sample_id}_{int(time.time())}.json")
with open(fname, "w", encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
return report
except Exception as e:
logger.exception("Inference failed for %s: %s", sample.sample_id, e)
audit_error(self.cfg.get('audit_dir','audit_logs'), sample.sample_id, ERROR_CODES['INFERENCE_FAIL'], str(e))
return {"sample_id": sample.sample_id, "verdict": "error", "confidence": 0.0, "error": str(e)}
# ----------------------------
# Blind-test harness
# ----------------------------
def run_blind_test(engine: InferenceEngine, samples: List[MultimodalSample]) -> Dict[str,Any]:
y_true = []
probs = []
reports = []
for s in samples:
r = engine.infer(s)
prob = r.get('confidence', 0.0)
label = s.label if s.label is not None else int(s.meta.get('label', 0))
y_true.append(label)
probs.append(prob)
reports.append(r)
y_true = np.array(y_true)
probs = np.array(probs)
auc = roc_auc_score(y_true, probs) if len(np.unique(y_true))>1 else 0.5
preds = (probs >= 0.5).astype(int)
precision, recall, f1, _ = precision_recall_fscore_support(y_true, preds, average='binary', zero_division=0)
return {"n": len(samples), "auc": auc, "precision": precision, "recall": recall, "f1": f1, "reports": reports}
# ----------------------------
# AutoOrchestrator (robust)
# ----------------------------
class AutoOrchestrator:
def __init__(self, cfg: Dict[str,Any]):
self.cfg = cfg
self.watch_dir = cfg.get('auto',{}).get('watch_dir', DEFAULT_CONFIG['auto']['watch_dir'])
self.poll_interval = cfg.get('auto',{}).get('poll_interval_s', DEFAULT_CONFIG['auto']['poll_interval_s'])
self.max_workers = cfg.get('auto',{}).get('max_workers', DEFAULT_CONFIG['auto']['max_workers'])
self.human_threshold = cfg.get('auto',{}).get('human_review_confidence_threshold', DEFAULT_CONFIG['auto']['human_review_confidence_threshold'])
self.auto_accept = cfg.get('auto',{}).get('auto_accept_confidence', DEFAULT_CONFIG['auto']['auto_accept_confidence'])
self.queue = Queue(maxsize=cfg.get('auto',{}).get('max_queue_size', DEFAULT_CONFIG['auto']['max_queue_size']))
self.workers: List[Thread] = []
self.stop_event = Event()
self.engine = None
ensure_dir(self.watch_dir)
ensure_dir(self.cfg.get('audit_dir', DEFAULT_CONFIG['audit_dir']))
def detect_input(self, path: str) -> Dict[str,Any]:
name = os.path.basename(path).lower()
if name.endswith('.json'):
try:
with open(path, 'r', encoding='utf-8') as f:
obj = json.load(f)
if isinstance(obj, list) and all('sample_id' in r for r in obj):
return {'task':'train_manifest','payload':{'manifest_path':path}}
if isinstance(obj, dict) and obj.get('type') == 'sample':
return {'task':'sample_json','payload':{'sample_json':path}}
except Exception:
pass
if name.endswith('.csv'):
try:
with open(path, 'r', encoding='utf-8') as f:
header = f.readline().lower()
if 'age_bp' in header and 'compound_id' in header:
return {'task':'cs14c_report','payload':{'path':path}}
if 'mz' in header and 'intensity' in header:
return {'task':'peak_table','payload':{'path':path}}
except Exception:
pass
if name.endswith(('.tif','.tiff','.png','.jpg','.jpeg')):
return {'task':'image_file','payload':{'path':path}}
return {'task':'unknown','payload':{'path':path}}
def route_task(self, detected: Dict[str,Any]):
t = detected.get('task')
if t == 'train_manifest':
self.enqueue({'type':'train','manifest_path': detected['payload']['manifest_path']})
elif t == 'sample_json':
self.enqueue({'type':'infer_from_json','sample_json': detected['payload']['sample_json']})
elif t == 'cs14c_report':
self.enqueue({'type':'calibrate_cs14c','path': detected['payload']['path']})
elif t == 'peak_table':
self.enqueue({'type':'infer_from_files','files': {'gcms': detected['payload']['path']}})
elif t == 'image_file':
self.enqueue({'type':'infer_from_files','files': {'image': detected['payload']['path']}})
else:
logger.info("Unknown input type for %s; moving to quarantine", detected['payload'].get('path'))
self.quarantine_file(detected['payload'].get('path'))
def enqueue(self, job: Dict[str,Any]):
try:
self.queue.put_nowait(job)
logger.info("Enqueued job: %s", job.get('type'))
except Exception as e:
logger.exception("Queue full or error enqueuing job: %s", e)
def worker_loop(self):
while not self.stop_event.is_set():
try:
job = self.queue.get(timeout=1)
except Empty:
continue
try:
self.execute_job(job)
except Exception as e:
logger.exception("Job execution failed: %s", e)
finally:
self.queue.task_done()
def execute_job(self, job: Dict[str,Any]):
jtype = job.get('type')
logger.info("Executing job type: %s", jtype)
if jtype == 'train':
manifest = job.get('manifest_path')
self._do_train(manifest)
elif jtype == 'infer_from_json':
sample_json = job.get('sample_json')
self._do_infer_from_json(sample_json)
elif jtype == 'infer_from_files':
files = job.get('files',{})
self._do_infer_from_files(files)
elif jtype == 'calibrate_cs14c':
path = job.get('path')
self._do_calibrate_cs14c(path)
else:
logger.warning("Unknown job type: %s", jtype)
def _ensure_engine(self):
if self.engine is None:
model_dir = self.cfg.get('model_dir','models')
secret = os.environ.get(self.cfg.get('provenance_secret_env','PROVENANCE_SECRET'),'change_this_secret')
self.engine = InferenceEngine(model_dir, self.cfg, secret)
def _do_train(self, manifest_path: str):
logger.info("Starting training from manifest: %s", manifest_path)
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
samples = []
for rec in manifest:
sid = rec['sample_id']
meta = json.load(open(rec['meta_json_path'],'r',encoding='utf-8')) if rec.get('meta_json_path') else {}
def read_opt(p): return open(p,'rb').read() if p and os.path.exists(p) else None
samples.append(MultimodalSample(sample_id=sid, meta=meta,
gcms_bytes=read_opt(rec.get('gcms_path')),
spectrum_bytes=read_opt(rec.get('spectrum_path')),
image_bytes=read_opt(rec.get('image_path')),
afm_bytes=read_opt(rec.get('afm_path')),
cs14c_bytes=read_opt(rec.get('cs14c_path')),
label=rec.get('label')))
res = train_pipeline(samples, self.cfg, self.cfg.get('model_dir','models'))
logger.info("Training finished: %s", res)
self._write_audit({'task':'train','manifest':manifest_path,'result':res,'timestamp':now_iso()})
def _do_infer_from_json(self, sample_json: str):
with open(sample_json, 'r', encoding='utf-8') as f:
rec = json.load(f)
def read_opt(p): return open(p,'rb').read() if p and os.path.exists(p) else None
sample = MultimodalSample(sample_id=rec.get('sample_id', str(uuid.uuid4())), meta=rec.get('meta', {}),
gcms_bytes=read_opt(rec.get('gcms_path')),
spectrum_bytes=read_opt(rec.get('spectrum_path')),
image_bytes=read_opt(rec.get('image_path')),
afm_bytes=read_opt(rec.get('afm_path')),
cs14c_bytes=read_opt(rec.get('cs14c_path')))
self._do_infer_sample(sample)
def _do_infer_from_files(self, files: Dict[str,str]):
meta = {}
sample = MultimodalSample(sample_id=str(uuid.uuid4()), meta=meta,
gcms_bytes=open(files['gcms'],'rb').read() if files.get('gcms') and os.path.exists(files.get('gcms')) else None,
spectrum_bytes=open(files['spectrum'],'rb').read() if files.get('spectrum') and os.path.exists(files.get('spectrum')) else None,
image_bytes=open(files['image'],'rb').read() if files.get('image') and os.path.exists(files.get('image')) else None,
afm_bytes=open(files['afm'],'rb').read() if files.get('afm') and os.path.exists(files.get('afm')) else None)
self._do_infer_sample(sample)
def _do_infer_sample(self, sample: MultimodalSample):
self._ensure_engine()
report = self.engine.infer(sample)
logger.info("Inference report: %s", json.dumps(report, ensure_ascii=False))
self._write_audit({'task':'infer','sample_id':sample.sample_id,'report':report,'timestamp':now_iso()})
conf = float(report.get('confidence',0.0))
if conf >= self.auto_accept:
logger.info("Auto-accepting result for %s (conf=%.3f)", sample.sample_id, conf)
self._publish_report(report)
elif conf < self.human_threshold:
logger.info("Low confidence -> trigger human review for %s (conf=%.3f)", sample.sample_id, conf)
self._trigger_human_review(report, reason="low_confidence")
else:
logger.info("Intermediate confidence -> queue for human review: %s (conf=%.3f)", sample.sample_id, conf)
self._trigger_human_review(report, reason="intermediate_confidence")
def _do_calibrate_cs14c(self, path: str):
logger.info("Processing CS-14C calibration report: %s", path)
try:
recs = parse_cs14c_precise(open(path,'rb').read())
self._write_audit({'task':'calibrate_cs14c','path':path,'records_count':len(recs.get('records',[])),'timestamp':now_iso()})
logger.info("CS-14C calibration records processed: %d", len(recs.get('records',[])))
except Exception as e:
logger.exception("Failed to process CS-14C: %s", e)
self._write_audit({'task':'calibrate_cs14c','path':path,'error':str(e),'timestamp':now_iso()})
def _publish_report(self, report: Dict[str,Any]):
out_dir = os.path.join(self.cfg.get('audit_dir','audit_logs'),'published')
ensure_dir(out_dir)
fname = os.path.join(out_dir, f"{report['sample_id']}_{int(time.time())}.json")
with open(fname, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
logger.info("Published report to %s", fname)
def _trigger_human_review(self, report: Dict[str,Any], reason: str):
review_dir = os.path.join(self.cfg.get('audit_dir','audit_logs'),'human_review')
ensure_dir(review_dir)
fname = os.path.join(review_dir, f"{report['sample_id']}_{int(time.time())}.json")
review_pkg = {'report': report, 'reason': reason, 'timestamp': now_iso()}
with open(fname, 'w', encoding='utf-8') as f:
json.dump(review_pkg, f, ensure_ascii=False, indent=2)
logger.info("Human review package created: %s", fname)
def _write_audit(self, audit: Dict[str,Any]):
audit_dir = self.cfg.get('audit_dir','audit_logs')
ensure_dir(audit_dir)
fname = os.path.join(audit_dir, f"audit_{int(time.time())}_{uuid.uuid4().hex[:8]}.json")
with open(fname, 'w', encoding='utf-8') as f:
json.dump(audit, f, ensure_ascii=False, indent=2)
logger.debug("Audit written: %s", fname)
def quarantine_file(self, path: str):
qdir = os.path.join(self.watch_dir, 'quarantine')
ensure_dir(qdir)
try:
base = os.path.basename(path)
dest = os.path.join(qdir, base)
os.rename(path, dest)
logger.info("Moved unknown file to quarantine: %s", dest)
except Exception:
logger.exception("Failed to quarantine file: %s", path)
def start_workers(self):
for i in range(self.max_workers):
t = Thread(target=self.worker_loop, daemon=True)
t.start()
self.workers.append(t)
logger.info("Started %d worker threads", len(self.workers))
def stop(self):
self.stop_event.set()
logger.info("Stop signal set; waiting for workers to finish")
for t in self.workers:
t.join(timeout=2)
def watch_loop(self):
logger.info("Starting watch loop on %s", self.watch_dir)
seen = set()
while not self.stop_event.is_set():
try:
files = [os.path.join(self.watch_dir, f) for f in os.listdir(self.watch_dir) if not f.startswith('.')]
except Exception:
files = []
for p in files:
if p in seen:
continue
if os.path.isdir(p):
seen.add(p)
continue
try:
detected = self.detect_input(p)
self.route_task(detected)
seen.add(p)
except Exception as e:
logger.exception("Error detecting/routing file %s: %s", p, e)
self.quarantine_file(p)
time.sleep(self.poll_interval)
def run_daemon(self):
logger.info("AutoOrchestrator starting daemon")
self.start_workers()
watch_thread = Thread(target=self.watch_loop, daemon=True)
watch_thread.start()
try:
while not self.stop_event.is_set():
time.sleep(1)
except KeyboardInterrupt:
logger.info("KeyboardInterrupt received; stopping orchestrator")
self.stop()
watch_thread.join(timeout=1)
logger.info("Orchestrator stopped")
# ----------------------------
# FastAPI UI prototype (optional)
# ----------------------------
if FastAPI is not None:
@asynccontextmanager
async def lifespan(app: FastAPI):
model_dir = os.environ.get("MODEL_DIR", DEFAULT_CONFIG['model_dir'])
cfg_path = os.environ.get("CONFIG_PATH", None)
if cfg_path and os.path.exists(cfg_path):
with open(cfg_path, 'r', encoding='utf-8') as f:
cfg = yaml.safe_load(f)
else:
cfg = DEFAULT_CONFIG
secret = os.environ.get(cfg.get('provenance_secret_env', 'PROVENANCE_SECRET'), 'change_this_secret')
try:
app.state.orchestrator = AutoOrchestrator(cfg)
app.state.orchestrator.start_workers()
logger.info("Orchestrator initialized in FastAPI lifespan")
except Exception as e:
logger.exception("Failed to initialize orchestrator: %s", e)
app.state.orchestrator = None
yield
if app.state.orchestrator:
app.state.orchestrator.stop()
app = FastAPI(title="ForensicOilCompleteReform", version=DEFAULT_CONFIG['model_version'], lifespan=lifespan)
@app.post("/upload")
async def upload_sample(sample_meta: str, gcms: Optional[UploadFile] = File(None), spectrum: Optional[UploadFile] = File(None), image: Optional[UploadFile] = File(None), afm: Optional[UploadFile] = File(None), cs14c: Optional[UploadFile] = File(None)):
if app.state.orchestrator is None:
raise HTTPException(status_code=503, detail="Orchestrator not ready")
try:
meta_obj = json.loads(sample_meta)
except Exception:
raise HTTPException(status_code=400, detail="Invalid meta JSON")
sid = meta_obj.get('sample_id', str(uuid.uuid4()))
gcms_b = await gcms.read() if gcms else None
spec_b = await spectrum.read() if spectrum else None
img_b = await image.read() if image else None
afm_b = await afm.read() if afm else None
cs14c_b = await cs14c.read() if cs14c else None
sample = MultimodalSample(sample_id=sid, meta=meta_obj.get('meta',{}), gcms_bytes=gcms_b, spectrum_bytes=spec_b, image_bytes=img_b, afm_bytes=afm_b, cs14c_bytes=cs14c_b)
app.state.orchestrator.enqueue({'type':'infer','sample':sample})
return JSONResponse({"sample_id": sid, "status":"queued"})
@app.get("/", response_class=HTMLResponse)
async def index():
html = """
<html><head><title>Forensic Oil Review</title></head><body>
<h2>Forensic Oil Residue System (Complete Reform)</h2>
<p>Use /upload to POST sample meta and files. This is a prototype UI.</p>
</body></html>
"""
return HTMLResponse(html)
# ----------------------------
# CLI helpers & tests
# ----------------------------
def build_dummy_samples(n=200) -> List[MultimodalSample]:
samples = []
for i in range(n):
label = 1 if i < n//2 else 0
meta = {"sample_id": f"synthetic_{i}", "object_material": "ceramic", "environment": {"temperature_c": 20.0 + np.random.randn(), "relative_humidity": 40.0 + np.random.randn()*5}}
# use placeholders to test robust parsing
samples.append(MultimodalSample(sample_id=meta['sample_id'], meta=meta, gcms_bytes=b"mzml_placeholder", spectrum_bytes=b"spectrum_placeholder", image_bytes=b"tiff_placeholder", afm_bytes=b"afm_placeholder", label=label))
return samples
def generate_deploy_artifacts(out_dir: str):
ensure_dir(out_dir)
dockerfile = f"""FROM python:3.10-slim
WORKDIR /app
COPY {os.path.basename(__file__)} /app/forensic_oil_complete_reform.py
RUN pip install --no-cache-dir fastapi uvicorn numpy pandas scipy scikit-learn xgboost pillow pymzml tifffile pymc3 arviz shap
ENV PROVENANCE_SECRET=change_this_secret
EXPOSE 8000
CMD ["python", "forensic_oil_complete_reform.py", "--mode", "run-api"]
"""
with open(os.path.join(out_dir, 'Dockerfile'), 'w', encoding='utf-8') as f:
f.write(dockerfile)
logger.info("Deployment artifacts generated at %s", out_dir)
# ----------------------------
# Main CLI
# ----------------------------
def main():
import argparse
parser = argparse.ArgumentParser(description="Forensic Oil Complete Reform CLI")
parser.add_argument("--mode", choices=["build-dummy","train","infer-file","auto","run-api","gen-deploy","self-test"], required=False, default="build-dummy")
parser.add_argument("--manifest", type=str, default="train_manifest.json")
parser.add_argument("--model-dir", type=str, default=DEFAULT_CONFIG['model_dir'])
parser.add_argument("--out", type=str, default="out.json")
parser.add_argument("--sample-file", type=str, default=None)
parser.add_argument("--deploy-out", type=str, default="deploy_artifacts")
args = parser.parse_args()
cfg = DEFAULT_CONFIG.copy()
cfg['model_dir'] = args.model_dir
cfg['audit_dir'] = DEFAULT_CONFIG['audit_dir']
cfg['auto']['watch_dir'] = DEFAULT_CONFIG['auto']['watch_dir']
if args.mode == "build-dummy":
samples = build_dummy_samples(200)
ensure_dir(args.model_dir)
res = train_pipeline(samples, cfg, args.model_dir)
print("Dummy training result:", res)
elif args.mode == "train":
if not os.path.exists(args.manifest):
print("Create train_manifest.json listing samples with fields: sample_id, meta_json_path, gcms_path, spectrum_path, image_path, afm_path, cs14c_path (optional), label")
sys.exit(1)
with open(args.manifest, 'r', encoding='utf-8') as f:
manifest_list = json.load(f)
samples = []
for rec in manifest_list:
sid = rec['sample_id']
meta = json.load(open(rec['meta_json_path'], 'r', encoding='utf-8')) if rec.get('meta_json_path') else {}
def read_opt(p):
return open(p, 'rb').read() if p and os.path.exists(p) else None
samples.append(MultimodalSample(sample_id=sid, meta=meta,
gcms_bytes=read_opt(rec.get('gcms_path')),
spectrum_bytes=read_opt(rec.get('spectrum_path')),
image_bytes=read_opt(rec.get('image_path')),
afm_bytes=read_opt(rec.get('afm_path')),
cs14c_bytes=read_opt(rec.get('cs14c_path')),
label=rec.get('label')))
res = train_pipeline(samples, cfg, args.model_dir)
print("Training result:", res)
elif args.mode == "infer-file":
if not args.sample_file or not os.path.exists(args.sample_file):
print("Provide sample JSON file describing paths.")
sys.exit(1)
with open(args.sample_file, 'r', encoding='utf-8') as f:
rec = json.load(f)
def read_opt(p):
return open(p, 'rb').read() if p and os.path.exists(p) else None
sample = MultimodalSample(sample_id=rec.get('sample_id', str(uuid.uuid4())), meta=rec.get('meta', {}),
gcms_bytes=read_opt(rec.get('gcms_path')),
spectrum_bytes=read_opt(rec.get('spectrum_path')),
image_bytes=read_opt(rec.get('image_path')),
afm_bytes=read_opt(rec.get('afm_path')),
cs14c_bytes=read_opt(rec.get('cs14c_path')))
secret = os.environ.get(cfg.get('provenance_secret_env', 'PROVENANCE_SECRET'), 'change_this_secret')
engine = InferenceEngine(args.model_dir, cfg, secret)
report = engine.infer(sample)
with open(args.out, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print("Inference saved to", args.out)
elif args.mode == "auto":
orchestrator = AutoOrchestrator(cfg)
try:
orchestrator.run_daemon()
except Exception as e:
logger.exception("Orchestrator failed: %s", e)
elif args.mode == "run-api":
if FastAPI is None:
print("FastAPI not installed; cannot run API")
sys.exit(1)
import uvicorn
uvicorn.run("forensic_oil_complete_reform:app", host="0.0.0.0", port=8000, reload=False)
elif args.mode == "gen-deploy":
generate_deploy_artifacts(args.deploy_out)
print("Deployment artifacts generated at", args.deploy_out)
elif args.mode == "self-test":
# run a few unit-like checks to ensure robust parsing works
print("Running self-test...")
samples = build_dummy_samples(10)
cfg_local = cfg
ensure_dir(cfg_local['model_dir'])
res = train_pipeline(samples, cfg_local, cfg_local['model_dir'])
print("Self-test train result:", res)
engine = InferenceEngine(cfg_local['model_dir'], cfg_local, os.environ.get(cfg_local.get('provenance_secret_env','PROVENANCE_SECRET'),'change_this_secret'))
report = engine.infer(samples[0])
print("Self-test infer sample:", report['sample_id'], report['verdict'], report['confidence'])
else:
print("Unknown mode:", args.mode)
sys.exit(1)
if __name__ == "__main__":
main()
license: apache-2.0
language:
- en
- zh
pipeline_tag: robotics
#!/usr/bin/env python3
# forensic_oil_complete_reform.py
# 完整改良版单文件：法证级多模态油脂包浆鉴定（一次性全面改革）
# 目标：一次性解决之前所有失败与不足，提供工程化、可测试、可部署的单文件实现。
# 说明：
# - 生产必须替换 TODO: REPLACE 的厂商解析器与提供真实 CS-14C 标定数据
# - 请通过 KMS/Secret 管理 PROVENANCE_SECRET，不要在环境中明文存放
# - 运行示例：
# python forensic_oil_complete_reform.py --mode build-dummy
# python forensic_oil_complete_reform.py --mode train --manifest train_manifest.json --model-dir models
# python forensic_oil_complete_reform.py --mode infer-file --sample-file sample.json --model-dir models
# python forensic_oil_complete_reform.py --mode auto
# uvicorn forensic_oil_complete_reform:app --host 0.0.0.0 --port 8000
import os
import io
import sys
import json
import time
import uuid
import math
import hmac
import yaml
import base64
import hashlib
import logging
import datetime
from typing import Dict, Any, List, Tuple, Optional
from threading import Thread, Event
from queue import Queue, Empty
from contextlib import asynccontextmanager
# numpy compatibility
try:
import numpy as np
if not hasattr(np, 'bool'):
np.bool = np.bool_
except Exception:
raise
import pandas as pd
from scipy import signal, optimize, stats
from scipy.ndimage import gaussian_filter
from sklearn.isotonic import IsotonicRegression
from sklearn.metrics import roc_auc_score, precision_recall_fscore_support
from sklearn.model_selection import train_test_split
import xgboost as xgb
# Optional heavy deps (graceful fallback)
try:
import pymzml
except Exception:
pymzml = None
try:
import tifffile
except Exception:
tifffile = None
try:
import pymc3 as pm
import arviz as az
except Exception:
pm = None
try:
import shap
except Exception:
shap = None
try:
from PIL import Image, UnidentifiedImageError
except Exception:
Image = None
UnidentifiedImageError = Exception
# Web API optional
try:
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel
except Exception:
FastAPI = None
# Logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger("forensic_oil_complete_reform")
# ----------------------------
# Default configuration
# ----------------------------
DEFAULT_CONFIG = {
"model_version": "5.0.0-complete-reform",
"provenance_secret_env": "PROVENANCE_SECRET",
"audit_dir": "audit_logs",
"model_dir": "models",
"incoming_dir": "incoming_samples",