"""Synthetic operations rehearsal with bursts, absences and overtime clearance."""
from dataclasses import replace
import math,json
import numpy as np
from scipy.stats import t as student_t,beta
from simulation import Job,generate,simulate
from completion_support import *
from support import plot_style,save_figure

POLICIES=[('2 FIFO',2,'FIFO'),('3 FIFO',3,'FIFO'),('2 estimated SPT',2,'SPT'),('3 estimated SPT',3,'SPT')]
CONDITIONS=['reference','five_minute_batches','two_hour_absence','service_plus_40pct','combined_stress']

def workload(seed,condition,inputs):
 service=1.4 if condition in ('service_plus_40pct','combined_stress') else 1.
 jobs=generate(seed,inputs['arrival_segments'],service_mean_s=inputs['mean_service_s']*service,service_cv=inputs['within_class_service_cv'],rework_probability=inputs['rework_probability'],max_visits=inputs['max_visits'])
 if condition in ('five_minute_batches','combined_stress'):
  jobs=[replace(j,arrival_s=math.floor(j.arrival_s/300)*300.) for j in jobs]
 return jobs

def shift_breaks(staff,condition,p):
 breaks=[(worker,start+worker*900,duration) for worker in range(staff) for start,duration in p['breaks_per_worker']]
 if condition in ('two_hour_absence','combined_stress'):breaks.append((0,3600,7200))
 return breaks

def metrics(result):
 jobs=result['jobs'];s=result['summary'];complete=[x for x in jobs if x['state']=='completed'];admitted=[x for x in jobs if x['state']!='rejected'];detailed=[x for x in admitted if x['kind']=='detailed']
 done8=sum(x['finish_s']<=28800 for x in complete);backlog=len(admitted)-done8
 overtime=max([0]+[x['finish_s']-28800 for x in complete])
 if s['unfinished']:overtime=7200. # observed-to-horizon lower bound, flagged separately below
 return dict(offered=s['offered'],admitted=s['admitted'],completed_at_8h=done8,backlog_at_8h=backlog,completed_by_10h=s['completed'],uncleared_at_10h=s['unfinished'],rejected=s['rejected'],rejected_fraction=s['rejected']/s['offered'] if s['offered'] else 0.,observed_clearance_overtime_s=overtime,clearance_censored=bool(s['unfinished']),detailed_wait_p95_s=float(np.quantile([j['waiting_s'] for j in detailed],.95)) if detailed else 0.,max_wip=s['maximum_inside'],balance_error=s['conservation_residual'],wait_area_error_s=s['wait_area_residual_s'],wip_area_error_s=s['wip_area_residual_s'])

def mean_ci(values):
 x=np.asarray(values,dtype=float);mean=float(x.mean());half=float(student_t.ppf(.975,len(x)-1)*x.std(ddof=1)/math.sqrt(len(x)))
 return dict(mean=mean,lower=mean-half,upper=mean+half)

def exact_risk(k,n):
 return dict(lower=0. if k==0 else float(beta.ppf(.025,k,n-k+1)),upper=1. if k==n else float(beta.ppf(.975,k+1,n-k)))

def main():
 p=json.loads((ROOT/'inputs.json').read_text());runs=[];balanced=True;example=[]
 for condition in CONDITIONS:
  for seed in range(8100,8140):
   jobs=workload(seed,condition,p)
   for label,staff,policy in POLICIES:
    result=simulate(jobs,staff=staff,capacity=24,horizon_s=36000,rework_delay_s=p['rework_delay_s'],breaks=shift_breaks(staff,condition,p),policy=policy)
    row=dict(condition=condition,policy=label,seed=seed,**metrics(result))
    row['meets_reference_service_screen']=row['rejected_fraction']<=.01 and row['observed_clearance_overtime_s']<=1800 and not row['clearance_censored'] and row['detailed_wait_p95_s']<=900
    runs.append(row)
    balanced &= row['balance_error']==0 and abs(row['wait_area_error_s'])<1e-6 and abs(row['wip_area_error_s'])<1e-6 and row['max_wip']<=24
    if condition=='combined_stress' and seed==8100 and label=='2 FIFO':example=result['jobs']
 write_csv('results/virtual-operations-runs.csv',runs);write_csv('results/virtual-stress-job-trace.csv',example)
 aggregates=[];risks=[];paired=[]
 for condition in CONDITIONS:
  for label,staff,policy in POLICIES:
   subset=[r for r in runs if r['condition']==condition and r['policy']==label]
   for metric in ['completed_at_8h','backlog_at_8h','rejected','observed_clearance_overtime_s','detailed_wait_p95_s']:
    aggregates.append(dict(condition=condition,policy=label,metric=metric,replications=40,**mean_ci([r[metric] for r in subset])))
   k=sum(not r['meets_reference_service_screen'] for r in subset)
   risks.append(dict(condition=condition,policy=label,shifts_missing_screen=k,replications=40,observed_risk=k/40,**exact_risk(k,40)))
  for metric in ['completed_at_8h','backlog_at_8h','rejected','detailed_wait_p95_s']:
   a=[r for r in runs if r['condition']==condition and r['policy']=='2 FIFO'];b=[r for r in runs if r['condition']==condition and r['policy']=='3 FIFO']
   paired.append(dict(condition=condition,comparison='3 FIFO minus 2 FIFO',metric=metric,**mean_ci([y[metric]-x[metric] for x,y in zip(a,b)])))
 write_csv('results/virtual-operations-means-ci.csv',aggregates);write_csv('results/virtual-service-risk-ci.csv',risks);write_csv('results/virtual-staffing-paired-ci.csv',paired)
 def avg(cond,policy,metric):return float(np.mean([r[metric] for r in runs if r['condition']==cond and r['policy']==policy]))
 offered=sum((b-a)*rate/3600 for a,b,rate in p['arrival_segments']);expected_visits=sum(p['rework_probability']**i for i in range(p['max_visits']));expected_service=p['mean_service_s']*expected_visits
 theory=[dict(staff=staff,expected_arrivals=offered,expected_service_per_job_s=expected_service,available_regular_service_s=staff*(28800-3600),nominal_offered_work_s=offered*expected_service,offered_work_fraction=offered*expected_service/(staff*(28800-3600))) for staff in [1,2,3]]
 write_csv('results/virtual-capacity-budget.csv',theory)
 # Selection uses the initial 40-seed experiment; validation uses disjoint seeds.
 staff_plan={'reference':3,'five_minute_batches':3,'two_hour_absence':4,'service_plus_40pct':4,'combined_stress':5}
 holdout=[];holdout_risk=[]
 for condition in CONDITIONS:
  for seed in range(16100,16200):
   staff=staff_plan[condition];result=simulate(workload(seed,condition,p),staff=staff,capacity=24,horizon_s=36000,rework_delay_s=p['rework_delay_s'],breaks=shift_breaks(staff,condition,p),policy='FIFO')
   r=dict(condition=condition,staff=staff,seed=seed,**metrics(result));r['meets_reference_service_screen']=r['rejected_fraction']<=.01 and r['observed_clearance_overtime_s']<=1800 and not r['clearance_censored'] and r['detailed_wait_p95_s']<=900;holdout.append(r)
  group=[r for r in holdout if r['condition']==condition];k=sum(not r['meets_reference_service_screen'] for r in group);risk=exact_risk(k,100)
  holdout_risk.append(dict(condition=condition,staff=staff_plan[condition],misses=k,shifts=100,**risk,meets_10pct_upper_bound=risk['upper']<=.10))
 write_csv('results/virtual-selected-holdout.csv',holdout);write_csv('results/virtual-selected-risk.csv',holdout_risk)
 write_json('selected-operating-envelope.json',dict(evidence_type='synthetic/theoretical',policy='FIFO',wip_limit=24,staff_by_condition=staff_plan,selection_seeds=[8100,8139],independent_validation_seeds=[16100,16199],total_holdout_runs=len(holdout),service_screen='<=1% refusal; <=30 min clearance; no uncleared work at 10 h; detailed-job p95 waiting <=15 min',acceptance='For each specified condition, exact two-sided 95% upper bound on shift screen-miss probability <=10%',risk_results=holdout_risk,condition_observation='Apply the declared staffing plan before the shift when the corresponding absence or slower-service condition is known. Unexpected conditions are outside the selected envelope.',optimal_staffing_claim=False,physical_validation=False))
 low,high=exact_risk(0,40).values()
 checks=[
 check('replications','Full virtual operations experiment executed',800,len(runs),len(runs)==800,'5 conditions x 4 policies x 40 independent matched seeds'),
 check('balances','All jobs and time-area identities remain accounted for',True,balanced,balanced,'Check every simulation for job count, queue-area, WIP-area and capacity balances'),
 check('backlog','End-of-shift backlog is carried into clearance','8 h completed + backlog = admitted for all runs',all(r['completed_at_8h']+r['backlog_at_8h']==r['admitted'] for r in runs),all(r['completed_at_8h']+r['backlog_at_8h']==r['admitted'] for r in runs),'Inspect each job finish time, not just the final aggregate'),
 check('stress','Combined adverse inputs increase regular-shift backlog','Mean combined-stress backlog > reference for 2 FIFO',dict(reference=avg('reference','2 FIFO','backlog_at_8h'),stress=avg('combined_stress','2 FIFO','backlog_at_8h')),avg('combined_stress','2 FIFO','backlog_at_8h')>avg('reference','2 FIFO','backlog_at_8h'),'Same declared policy; batch arrivals, 40% slower service and a 2 h absence'),
 check('third-worker','Additional staff reduces stress backlog','3 FIFO mean backlog < 2 FIFO',dict(two=avg('combined_stress','2 FIFO','backlog_at_8h'),three=avg('combined_stress','3 FIFO','backlog_at_8h')),avg('combined_stress','3 FIFO','backlog_at_8h')<avg('combined_stress','2 FIFO','backlog_at_8h'),'Matched arrival/service/rework streams, same WIP limit'),
 check('service-screen','Screen identifies adverse operational conditions','At least one shift misses declared screen',sum(not r['meets_reference_service_screen'] for r in runs),any(not r['meets_reference_service_screen'] for r in runs),'Reject >1% arrivals, >30 min clearance, >15 min detailed-job tail wait or unfinished at horizon'),
 check('selected-policy','Selected FIFO staffing plan passes independent holdout risk criterion','Each condition upper 95% risk <= 10%',holdout_risk,all(r['meets_10pct_upper_bound'] for r in holdout_risk),'100 disjoint seeds per condition; 3 regular, 4 for absence or slow service, 5 for combined stress'),
 check('zero-failure-risk','Zero observed failures do not imply zero probability','0 < upper bound < 0.1',high,0<high<.1,'Two-sided exact binomial 95% interval for 0/40 failures'),
 check('capacity-bound','One nominal inspector has insufficient mean work budget','Offered-work fraction > 1 for one, < 1 for two',theory[:2],theory[0]['offered_work_fraction']>1 and theory[1]['offered_work_fraction']<1,'Independent expected workload = arrivals * mean service * sum p^visit; excludes randomness/burst penalty'),
 ]
 display=[]
 for condition in CONDITIONS:
  for label,_,_ in POLICIES:
   risk=next(r for r in risks if r['condition']==condition and r['policy']==label)
   display.append(dict(condition=condition,policy=label,mean_done_8h=round(avg(condition,label,'completed_at_8h'),2),mean_backlog_8h=round(avg(condition,label,'backlog_at_8h'),2),mean_refused=round(avg(condition,label,'rejected'),2),screen_failures=f"{risk['shifts_missing_screen']}/40",screen_risk_95pct=f"{100*risk['lower']:.1f}% to {100*risk['upper']:.1f}%"))
 plt=plot_style();fig,ax=plt.subplots(figsize=(10,4.5));x=np.arange(len(CONDITIONS));width=.36
 for i,label in enumerate(['2 FIFO','3 FIFO']):
  values=[next(r for r in aggregates if r['condition']==c and r['policy']==label and r['metric']=='backlog_at_8h') for c in CONDITIONS];ax.bar(x+(i-.5)*width,[r['mean'] for r in values],width,yerr=[r['upper']-r['mean'] for r in values],capsize=3,ecolor='#e1e6dd',label=label)
 ax.set_xticks(x,[c.replace('_','\n') for c in CONDITIONS]);ax.set(ylabel='Mean unfinished admitted jobs at 8 h',title='End-of-shift backlog; nominal 95% intervals across 40 shifts');ax.legend();ax.grid(axis='y');fig.tight_layout();save_figure(fig,'virtual-operations');plt.close(fig)
 finish('Inspection cell virtual operating trial','Executed 800 synthetic operating shifts and 500 independent holdout shifts with batching, absence, slower service and overtime clearance. Every refused, carried and unfinished job remains counted.',[
 'The original workload is reused with five conditions: reference; five-minute arrival batches; a two-hour worker-0 absence; service times increased 40%; and all three stressors combined. Forty seeds, 8100 through 8139, are generated independently. The four staffing/dispatch policies within a condition receive identical pre-generated jobs.',
 'Batches move each arrival to the beginning of its five-minute interval. This preserves the offered job count and service draws but deliberately changes the temporal pattern. It is a stress input, not an inferred factory distribution.',
 'WIP remains 24 jobs, including work in process, waiting and rework delay. Two or three inspectors use FIFO or shortest estimated service. Class estimates do not see actual sampled durations. Absence starts after any active job finishes, just like the existing deferred-break semantics; it is not an abrupt interrupted-service model.',
 'Arrivals stop after eight hours. Staff continue for at most two hours to clear admitted backlog. The study records completion by eight hours, backlog, refusal, overtime and uncleared work. If work remains at ten hours, clearance is censored at two hours and fails the service screen.',
 'Selected operating policy is FIFO with three inspectors for reference or batched arrivals, four for a known two-hour absence or 40% slower service, and five for the combined case. This is a capacity reserve policy, not a minimum-cost staffing claim. It was evaluated on 100 new seeds per condition, disjoint from policy-selection runs. The acceptance criterion is a two-sided exact 95% upper bound of at most 10% on a shift missing the declared service screen.',
 'The illustrative service screen requires <=1% refused arrivals, <=30 minutes clearance overtime, no uncleared work and <=15 minutes empirical 95th-percentile waiting among detailed jobs. This is an analyst-defined comparison criterion, not a customer service agreement. A policy can raise total completions while making long-job waiting worse.',
 'Mean intervals are Student-t nominal 95% intervals across 40 shifts. Paired differences use the same seed before computing intervals. Exact binomial intervals describe a shift missing the full screen. These intervals capture Monte Carlo variation under the stated inputs, not uncertainty about real demand or service distributions.'
 ],display,checks,[
 'This completes a synthetic operating trial. No actual production shift, time study, staffing decision or throughput gain is claimed.',
 'The experiment includes one deferred absence and a two-hour clearance window. It does not model multi-day carryover, skill differences, inspection errors, physical space or economic cost.',
 'The service screen is deliberately explicit and can fail. Monte Carlo confidence does not certify input realism, and repeated scenario comparisons are not adjusted for multiple testing.'
 ],[('selected-operating-envelope.json','Selected FIFO policy and staffing reserves'),('results/virtual-selected-risk.csv','Independent selected-policy acceptance'),('results/virtual-selected-holdout.csv','500 independent holdout shifts'),('figures/virtual-operations.svg','Mean end-of-shift backlog under the five workload conditions'),('results/virtual-operations-runs.csv','All 800 operating-trial runs'),('results/virtual-operations-means-ci.csv','Scenario means and intervals'),('results/virtual-service-risk-ci.csv','Exact service-screen failure intervals'),('results/virtual-staffing-paired-ci.csv','Matched two-versus-three inspector comparison'),('results/virtual-stress-job-trace.csv','Individual stress-case job outcomes'),('results/virtual-capacity-budget.csv','Independent workload capacity calculation')],[dict(label='Additional virtual operating shifts',value=1300),dict(label='Combined stress backlog, 2 FIFO',value=round(avg('combined_stress','2 FIFO','backlog_at_8h'),2),unit='jobs')])
if __name__=='__main__':main()