"""Sharp lower bounds, distinct reserve premises, and observation decisions.""" from dataclasses import dataclass from fractions import Fraction as F from typing import Protocol from material import nonnegative,attaining_history,History class ReservePremise(Protocol): def effective_parameter(self,B:F,q1:F)->F:... def accepts(self,history:History)->bool:... def initial_split(self,B:F)->F:... def description(self)->str:... @dataclass(frozen=True) class DirectReserve: ceiling:F def __post_init__(self):nonnegative(self.ceiling) def effective_parameter(self,B,q1):return self.ceiling def accepts(self,h):return h.first.final_reserve<=self.ceiling def initial_split(self,B):return F(0) def description(self):return 'Direct bound on reserve immediately before recovery.' @dataclass(frozen=True) class UptakeBound: initial_reserve_ceiling:F cumulative_uptake_ceiling:F def __post_init__(self):nonnegative(self.initial_reserve_ceiling,self.cumulative_uptake_ceiling) def effective_parameter(self,B,q1):return self.initial_reserve_ceiling+self.cumulative_uptake_ceiling def accepts(self,h):return h.first.reserve<=self.initial_reserve_ceiling and h.first.uptake<=self.cumulative_uptake_ceiling def initial_split(self,B):return min(self.initial_reserve_ceiling,B) def description(self):return 'Initial reserve plus gross first-window uptake cap; this sum is NOT a measured pre-wash reserve bound. Actual reserve may additionally contain first-window fresh material.' @dataclass(frozen=True) class UnboundedReserve: def effective_parameter(self,B,q1):return max(F(0),B-q1) def accepts(self,h):return True def initial_split(self,B):return F(0) def description(self):return 'No reserve information. Only reserve-free expressions determine the result.' @dataclass(frozen=True) class MaterialCertificate: initial_total:F mobile_retention:F reserve_retention:F recovery_input:F premise:ReservePremise def __post_init__(self): nonnegative(self.initial_total,self.mobile_retention,self.reserve_retention,self.recovery_input) if not self.mobile_retention<=self.reserve_retention<=1:raise ValueError('Require 0 <= e <= s <= 1.') def evaluate(self,q1,q2): q1,q2=F(q1),F(q2);nonnegative(q1,q2) B,e,s,H=self.initial_total,self.mobile_retention,self.reserve_retention,self.recovery_input J=self.premise.effective_parameter(B,q1) entries=dict(nonnegativity=F(0),first_window=q1-B,total=q1+q2-B-H, reserve_sensitive=e*q1+q2-e*B-(s-e)*J-H,uniform_retention=s*q1+q2-s*B-H) A=max(F(0),q1-B);x=max(F(0),B-q1);carry=e*x+(s-e)*min(J,x);lower=max(entries.values()) if lower!=A+max(F(0),q2-H-carry):raise ArithmeticError('Closed form and affine certificate disagree.') advice=[] if isinstance(self.premise,UnboundedReserve):advice.append('No reserve premise supplied; an independently justified direct reserve or gross-uptake bound may exclude additional histories.') elif e==s or J>=x:advice.append('Local tightening of this reserve parameter has no effect while e=s or J remains above the remaining stock.') elif entries['reserve_sensitive']==lower and lower>0:advice.append(f'The reserve-sensitive expression is active. Tightening its parameter increases this expression at slope {s-e}; follow all expressions through ties.') else:advice.append('Reserve information is not currently the uniquely useful constraint; inspect the active material and observation budgets.') return dict(lower=lower,expressions=entries,active=[k for k,v in entries.items() if v==lower], first_deficit=A,remaining_stock=x,maximum_carryover=carry,effective_reserve_parameter=J,premise=self.premise.description(),sensitivity=advice) def witness(self,q1,q2,initial_reserve=None): q1,q2=F(q1),F(q2);r=self.evaluate(q1,q2) R0=self.premise.initial_split(self.initial_total) if initial_reserve is None else F(initial_reserve) h=attaining_history(q1,q2,self.initial_total,self.mobile_retention,self.reserve_retention,r['effective_reserve_parameter'],self.recovery_input,R0) if not self.premise.accepts(h):raise ValueError('Chosen initial split does not satisfy this premise; use its default attaining split.') if h.fresh!=r['lower']:raise ArithmeticError('Attainment mismatch.') return h def valid_for(self,h): return (h.first.mobile+h.first.reserve<=self.initial_total and self.premise.accepts(h) and h.recovery.mobile_input+h.recovery.reserve_input<=self.recovery_input and h.recovery.mobile_retention<=self.mobile_retention and h.recovery.reserve_retention<=self.reserve_retention) @dataclass(frozen=True) class Observation: first:F second:F first_error:F second_error:F def __post_init__(self):nonnegative(self.first_error,self.second_error) def lower_collections(self): if min(self.first+self.first_error,self.second+self.second_error)<0:return None return max(F(0),self.first-self.first_error),max(F(0),self.second-self.second_error) def report(observation,certificates,threshold,fresh_upper=None): threshold=F(threshold);nonnegative(threshold) if fresh_upper is not None:fresh_upper=F(fresh_upper);nonnegative(fresh_upper) if not certificates:raise ValueError('At least one material certificate required.') q=observation.lower_collections() if q is None:return dict(verdict='incompatible',reason='An observation interval excludes every nonnegative collection.',lower=None) results=[c.evaluate(*q) for c in certificates];lower=max(r['lower'] for r in results) if fresh_upper is not None and lower>fresh_upper:verdict='incompatible' elif lower>=threshold:verdict='at or above' elif fresh_upper is not None and fresh_upper