| """Research-only product entrypoints; explicit numeric model vs LLM status.""" |
| from __future__ import annotations |
| from pathlib import Path |
| import numpy as np |
| from .parameters import ParameterPack,ContractError,feature_row,transform,predict,predict_proba,isolation_score,matrix |
|
|
| class NCLPredictor: |
| def __init__(self,pack:ParameterPack): |
| if pack.manifest['product']!='ncl':raise ContractError('NCL pack required') |
| self.pack=pack;self.names=pack.data['feature_cols'];self.models=pack.data['models'] |
| if len(self.names)!=28 or any(m['n_features']!=len(self.names) for m in self.models.values()):raise ContractError('Inconsistent NCL feature space') |
| def run(self,features:dict,*,synthetic:bool)->dict: |
| if synthetic is not True:raise ContractError('This acceptance build only accepts explicitly synthetic fixtures; clinical use is disabled') |
| x=feature_row(features,self.names) |
| outputs={k:predict(v,x)[0].item() for k,v in self.models.items()} |
| return {'schema':'ncl.fitted-research/1','synthetic':True,'research_only':True,'clinical_claim':False,'source_model_sha256':self.pack.manifest['source_files'][0]['sha256'],'parameter_pack_sha256':self.pack.id,'feature_order':self.names,'predictors':outputs,'progression_probabilities':dict(zip(self.models['progression_agent']['classes'],predict_proba(self.models['progression_agent'],x)[0].tolist())),'llm_explanation':None,'llm_status':'NOT_LOADED_BASE_REVISION_AND_RUNTIME_PENDING','external_training_preprocessing':'NOT_ESTABLISHED','scope':'Fitted predictor execution on supplied complete features; not held-out PPMI performance or clinical guidance'} |
|
|
| class MRIPredictor: |
| def __init__(self,pack:ParameterPack): |
| if pack.manifest['product']!='mri':raise ContractError('MRI pack required') |
| self.pack=pack;self.backend=pack.data['backend'] |
| def scores(self,space:str,x:np.ndarray,variant:str='optimized'): |
| if space not in ('baseline','huret') or variant not in ('exact','optimized'):raise ContractError('Unknown space/variant') |
| m=self.pack.data[variant][space];x=matrix(x,len(m['mean'])).astype(np.float64) |
| with np.errstate(over='ignore',invalid='ignore',divide='ignore'):z=(x-m['mean'])/np.maximum(m['scale'],1e-8) |
| if not np.isfinite(z).all():raise ContractError('MRI normalization exceeded finite range') |
| columns=[] |
| for c,p in zip(m['centers'],m['precisions']): |
| d=z-c |
| with np.errstate(over='ignore',invalid='ignore'):s=np.einsum('ij,jk,ik->i',d,p,d,optimize=True) |
| if not np.isfinite(s).all():raise ContractError('MRI quadratic score exceeded finite range') |
| columns.append(s) |
| all_scores=np.column_stack(columns);cluster=np.argmin(all_scores,axis=1);score=all_scores[np.arange(len(x)),cluster] |
| pct=np.searchsorted(m['normal_scores_sorted'],score,side='right')/max(len(m['normal_scores_sorted']),1) |
| return score,pct,cluster,z |
| def run(self,features:dict,*,space:str='huret',variant:str='optimized',synthetic:bool)->dict: |
| if synthetic is not True:raise ContractError('Synthetic acceptance inputs only; no patient endpoint enabled') |
| if space not in ('baseline','huret'):raise ContractError('Unknown feature space') |
| names=self.backend['feature_names_'+space];x=feature_row(features,names);raw,pct,cluster,z=self.scores(space,x,variant) |
| c=int(cluster[0]);deviation=z[0]-self.pack.data[variant][space]['centers'][c];top=np.argsort(np.abs(deviation))[::-1][:8] |
| return {'schema':'mri.fitted-research/1','space':space,'variant':variant,'run_id':self.backend['run_id'],'parameter_pack_sha256':self.pack.id,'raw_reference_distance':float(raw[0]),'anomaly_percentile':float(pct[0]),'anomaly':bool(pct[0]>=self.backend['thresholds'][space]),'nearest_reference_cluster':c,'top_vector_deviations':[{'feature':names[i],'standardized_deviation':float(deviation[i])} for i in top],'research_only':True,'synthetic':True,'clinical_claim':False,'normal_reference':self.backend['normal_reference'],'claim_boundary':self.backend['claim_boundary']} |
|
|
| class RhythmPredictor: |
| def __init__(self,pack:ParameterPack): |
| if pack.manifest['product']!='rhythm':raise ContractError('Rhythm pack required') |
| self.pack=pack;self.model=pack.data;self.names=self.model['common_features'] |
| def run(self,features:dict,*,synthetic:bool,acknowledge_export_version_mismatch:bool=False)->dict: |
| if synthetic is not True:raise ContractError('Synthetic acceptance inputs only') |
| if self.pack.manifest['export_warnings'] and not acknowledge_export_version_mismatch:raise ContractError('Original sklearn 1.9 environment must be reproduced; explicit compatibility-study acknowledgement required') |
| x=feature_row(features,self.names,dtype=np.float32);m=self.model |
| xs=transform(m['scaler'],x);state=predict(m['state_encoder'],xs)[0] |
| guards={} |
| for name in ('fog_guard','balance_guard'): |
| g=m[name];p=predict_proba(g['model'],transform(g['scaler'],x))[0];classes=g['model']['classes'] |
| if 1 not in classes:raise ContractError('Positive guard class missing') |
| guards[name]=float(p[classes.index(1)]) |
| return {'schema':'rhythm.fitted-research/1','synthetic':True,'research_only':True,'clinical_claim':False,'feature_order':self.names,'parameter_pack_sha256':self.pack.id,'latent_state':dict(zip(m['state_names'],state.tolist())),'ood_score':float(isolation_score(m['ood_model'],xs)[0]),'guard_probabilities':guards,'adaptation_enabled':False,'tempo_recommendation':None,'export_version_mismatch':bool(self.pack.manifest['export_warnings']),'scope':'Recovered fitted components in an explicit compatibility study; not online response, device safety or treatment benefit'} |
|
|
