Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444"""Sequential, evidence-producing experiments. No combat balance implementation."""import hashlibimport jsonimport mathfrom pathlib import Pathimport time
from . import experiments as labfrom . import nativefrom .diagnostics import DiagnosticError
def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def validate(plan): def keys(value, allowed, required=()): if not isinstance(value, dict) or set(value)-set(allowed) or set(required)-set(value): raise ValueError(f'Expected keys {allowed}, required {required}; received {value!r}') keys(plan, ('fixture','trials','repeats','description','stop_after'), ('fixture','trials')) native.validate('load', plan['fixture']) repeats=plan.get('repeats',1) if type(repeats) is not int or not 1<=repeats<=10: raise ValueError('repeats must be integer 1..10') if not isinstance(plan['trials'],list) or not 1<=len(plan['trials'])<=32: raise ValueError('trials must contain 1..32 scenarios') if 'stop_after' in plan and type(plan['stop_after']) is not bool: raise ValueError('stop_after must be boolean') for trial in plan['trials']: keys(trial, ('name','arena','enemy','actions','settings','assertions','policy','declared_variable','sampling'), ('name','arena','enemy','actions')) if 'sampling' in trial: keys(trial['sampling'], ('every_ticks',), ('every_ticks',)) every = trial['sampling']['every_ticks'] if type(every) is not int or not 2 <= every <= 120: raise ValueError('sampling.every_ticks must be integer 2..120') if not isinstance(trial['name'],str) or not trial['name']: raise ValueError('trial name must be nonempty') native.validate('teleport',trial['arena']) enemy=trial['enemy'] keys(enemy,('template','level','count','distance'),('template','level')) native.validate('spawn',f"{enemy['template']}|{enemy.get('count',1)}|{enemy['level']}|{enemy.get('distance',4)}") if not isinstance(trial['actions'],list) or len(trial['actions'])>100: raise ValueError('actions must be a list of at most 100 bounded steps') for action in trial['actions']: keys(action,('input','advance','require_weapon','wait_input_ready','wait_counter_window')) if len(action)!=1:raise ValueError('Each action specifies exactly one of input, advance, require_weapon, wait_input_ready, wait_counter_window') if 'input' in action:native.validate('input',action['input']) elif 'wait_input_ready' in action or 'wait_counter_window' in action: budget=next(iter(action.values())) if type(budget) is not int or not 1<=budget<=600:raise ValueError('Wait budget must be integer 1..600') elif 'require_weapon' in action: if action['require_weapon'] not in ('steel','silver'):raise ValueError('require_weapon must be steel or silver') else: if type(action['advance']) is not int or not 1<=action['advance']<=120: raise ValueError('Runner advance must be integer 1..120; collection occurs between steps') settings=trial.get('settings',{}) keys(settings,('difficulty','upscaling','prototype_disabled')) if 'difficulty' in settings and (type(settings['difficulty']) is not int or not 1<=settings['difficulty']<=4): raise ValueError('difficulty must be integer 1..4') for key in ('upscaling','prototype_disabled'): if key in settings and type(settings[key]) is not bool:raise ValueError(key+' must be boolean') assertions=trial.get('assertions',{}) keys(assertions,('minimum_contacts','minimum_incoming_contacts','health_channel','required_flags','flag_scope','flag_match','no_contacts')) if 'minimum_contacts' in assertions and (type(assertions['minimum_contacts']) is not int or assertions['minimum_contacts']<0): raise ValueError('minimum_contacts must be a nonnegative integer') if 'minimum_incoming_contacts' in assertions and (type(assertions['minimum_incoming_contacts']) is not int or assertions['minimum_incoming_contacts']<0): raise ValueError('minimum_incoming_contacts must be a nonnegative integer') if 'health_channel' in assertions and assertions['health_channel'] not in ('vitality','essence'): raise ValueError('health_channel must be vitality or essence') if 'no_contacts' in assertions and type(assertions['no_contacts']) is not bool:raise ValueError('no_contacts must be boolean') flags=assertions.get('required_flags',[]) if not isinstance(flags,list) or any(x not in ('parried','countered','dodged','critical','armor_zero','parry_stagger','ends_quen_flag') for x in flags): raise ValueError('required_flags must contain supported native flag names') if assertions.get('flag_scope','encounter') not in ('encounter','incoming','outgoing'): raise ValueError('flag_scope must be encounter, incoming or outgoing') if assertions.get('flag_match','independent') not in ('independent','same_contact'): raise ValueError('flag_match must be independent or same_contact') if ('flag_scope' in assertions or 'flag_match' in assertions) and not flags: raise ValueError('flag_scope/flag_match require nonempty required_flags') names=[t['name'] for t in plan['trials']] if len(names)!=len(set(names)):raise ValueError('Trial names must be unique') return plan
def trial_action(trial, command, payload): if 'sampling' in trial: from . import encounters return encounters.sampled_action(command, payload, trial['sampling']['every_ticks']) return native.bounded_action(command, payload)
def configure(settings): value = lab.data('lab_state')['data'] desired={k:settings.get(k,value[k]) for k in ('difficulty','upscaling','prototype_disabled')} if any(value[k]!=v for k,v in desired.items()): lab.data('lab_config',f"{desired['difficulty']}|{int(desired['upscaling'])}|{int(desired['prototype_disabled'])}") observed=lab.data('lab_state')['data'] if any(observed[k]!=v for k,v in desired.items()): raise DiagnosticError('E7001','Requested settings were not observed.',dict(requested=desired,observed=observed)) return observed
def event_reference(event): """IDs are scoped by the containing report's session and trial generation.""" return {key:event.get(key) for key in ('generation','sequence','contact_id','attempt_id','attacker','victim','causer_actor')}
def encounter_direction(event, player_id, enemy_ids): if event.get('attacker')==player_id and event.get('victim') in enemy_ids: return 'outgoing' if event.get('victim')==player_id and event.get('attacker') in enemy_ids: return 'incoming' return None
def assertions_result(spec, events, player_id, enemy_ids): contacts=[e for e in events if e.get('kind')=='contact'] scoped={direction:[e for e in contacts if encounter_direction(e,player_id,enemy_ids)==direction] for direction in ('incoming','outgoing')} scoped['encounter']=[e for e in contacts if encounter_direction(e,player_id,enemy_ids)] outgoing=scoped['outgoing'] checks=[] def check(name, passed, matches, **detail): checks.append(dict(name=name,passed=passed,matched_events=[event_reference(e) for e in matches],**detail)) if 'minimum_contacts' in spec: check('minimum_outgoing_contacts',len(outgoing)>=spec['minimum_contacts'],outgoing,scope='outgoing',observed=len(outgoing),expected=spec['minimum_contacts']) if 'minimum_incoming_contacts' in spec: incoming=scoped['incoming'] check('minimum_incoming_contacts',len(incoming)>=spec['minimum_incoming_contacts'],incoming,scope='incoming',observed=len(incoming),expected=spec['minimum_incoming_contacts']) if spec.get('no_contacts'): check('no_contacts',not scoped['encounter'],scoped['encounter'],scope='encounter',observed=len(scoped['encounter'])) if 'health_channel' in spec: check('outgoing_health_channel',bool(outgoing) and all(e.get('health_type')==spec['health_channel'] for e in outgoing),outgoing,scope='outgoing') scope=spec.get('flag_scope','encounter') mode=spec.get('flag_match','independent') flags=spec.get('required_flags',[]) def has_flag(event,flag): return event.get(flag) is True or (event.get('attack') or {}).get(flag) is True if flags and mode=='same_contact': matches=[e for e in scoped[scope] if all(has_flag(e,flag) for flag in flags)] check('observed_flags_same_contact',bool(matches),matches,scope=scope,flag_match=mode,required_flags=flags) else: for flag in flags: matches=[e for e in scoped[scope] if has_flag(e,flag)] check('observed_'+flag,bool(matches),matches,scope=scope,flag_match=mode) return checks
def interference_events(contacts, player_id, enemy_ids): participants={player_id,*enemy_ids} details=[] def role(actor): if actor==player_id:return 'player' if actor in enemy_ids:return 'owned_enemy' if type(actor) is not int or actor<0:return 'unknown_or_nonactor' return 'outsider' for event in contacts: attacker=event.get('attacker');victim=event.get('victim');causer=event.get('causer_actor') attacker_inside=attacker in participants;victim_inside=victim in participants # Unknown causers may be weapons/effects of a known attacker. Do not infer # an outsider from that absence alone; unknown attackers remain explicit. external_causer=type(causer) is int and causer>=0 and causer not in participants if attacker_inside != victim_inside: direction='external_to_participant' if victim_inside else 'participant_to_external' elif (attacker_inside or victim_inside) and external_causer: direction='external_causer' else: continue details.append(dict(**event_reference(event),direction=direction, attacker_role=role(attacker),victim_role=role(victim),causer_role=role(causer))) return details
def summarize_events(events, player_id, enemy_ids): attempts=[e for e in events if e.get('kind')=='attempt'] contacts=[e for e in events if e.get('kind')=='contact'] contact_attempts={e.get('attempt_id') for e in contacts if e.get('attempt_id')} incoming=[e for e in contacts if e.get('victim')==player_id] enemy_attempts=[e for e in attempts if e['attacker'] in enemy_ids] interference=interference_events(contacts,player_id,enemy_ids) windows=[dict(start_tick=e['tick'],end_tick=e['tick']+30, actor_ids=sorted({a['attacker'] for a in enemy_attempts if e['tick']<=a['tick']<=e['tick']+30})) for e in enemy_attempts] return dict(attempts=len(attempts), contacts=len(contacts), spawn_count=len(enemy_ids), per_actor=[dict(id=i,attempts=sum(e['attacker']==i for e in attempts),contacts=sum(e['attacker']==i for e in contacts)) for i in [player_id]+enemy_ids], attack_overlap_windows=windows, maximum_attackers_in_30_ticks=max((len(w['actor_ids']) for w in windows),default=0), attempts_without_observed_contact=[e['attempt_id'] for e in attempts if e['attempt_id'] not in contact_attempts], incoming_contacts=len(incoming), attacking_enemy_ids=sorted({e['attacker'] for e in attempts if e['attacker'] in enemy_ids}), incoming_health_loss=sum(e['hp_before']-e['hp_after'] for e in incoming), environmental_interference=[e['sequence'] for e in interference], interference_details=interference, interference_semantics='Observed external contacts involving player or owned enemies, including known external actor causers. Unknown attacker/victim endpoints involving a participant are conservative interference. An unavailable causer alone does not establish an outsider. Contact telemetry cannot detect all environmental or AI influences.', unknown_attempt_contacts=[e['sequence'] for e in contacts if not e.get('attempt_id')], pressure_semantics='Distinct attackers beginning DoAttack within inclusive rolling 30-script-tick windows: an overlap proxy, not animation overlap, threat prediction, or an approved difficulty target.')
def geometry(profiles, snapshot): player=profiles['profiles'][0] return dict(player_position=player['actor']['position'],player_heading=player['heading'], stamina=player['stamina'],player_state=snapshot['data']['world']['state'], input_allowed=snapshot['data']['world']['input_allowed'],player_alive=player['actor']['alive'], camera_direction=snapshot['data']['world']['camera_direction'], nearby_count=len(snapshot['data']['world']['nearby']), enemies=[dict(id=p['actor']['id'],position=p['actor']['position'],heading=p['heading'], distance=math.dist(player['actor']['position'],p['actor']['position'])) for p in profiles['profiles'][1:]], interruption_semantics='Sampled player state only; transitions wholly between samples are unavailable.')
def comparisons(plan, results, original): rows=[] for i,left in enumerate(plan['trials']): for j,right in enumerate(plan['trials'][i+1:],i+1): a={**{k:left['enemy'].get(k,{'count':1,'distance':4}.get(k)) for k in ('template','level','count','distance')},**original,**left.get('settings',{})} b={**{k:right['enemy'].get(k,{'count':1,'distance':4}.get(k)) for k in ('template','level','count','distance')},**original,**right.get('settings',{})} changed=[('enemy.'+k if k in ('template','level','count','distance') else k) for k in a if a[k]!=b[k]] declared={left.get('declared_variable'),right.get('declared_variable')}-{None} if not declared:continue if len(changed)!=1 or changed[0] not in declared:continue policy_equal=left['actions']==right['actions'] and left['arena']==right['arena'] lr=[r for r in results if r['name']==left['name']];rr=[r for r in results if r['name']==right['name']] rows.append(dict(left=left['name'],right=right['name'],declared_variable=changed[0], one_declared_input_changed=policy_equal, observed_trials_valid=bool(lr and rr) and all(r['experimental_validity'] for r in lr+rr), repetitions_left=len(lr),repetitions_right=len(rr), deterministic_replay=False, confounders=['Native RNG/AI','Initial load and setup ticks','Observed profiles/geometry may differ; inspect every repeat'], model_validation=False)) return rows
def condensed(report): """Shareable observations without copies of extracted game/mod source.""" out={k:report.get(k) for k in ('version','fixture','session','build','telemetry_version','game_executable','plan','uncontrolled','outcomes','comparisons','settings_restoration','stop','model_validation')} out['trials']=[] for trial in report['trials']: row={k:trial.get(k) for k in ('name','repetition','generation','command_success','experimental_validity','outcome','reason','terminal','settings','policy','declared_variable','assertions','trace_complete','summary','geometry','diagnostic')} spawn=trial.get('spawn',{}) row['setup']={k:spawn.get(k) for k in ('requested','observed','elapsed_ticks','stability','reason')} row['arrival']=trial.get('arrival') row['before']=trial.get('before');row['after']=trial.get('after') row['events']=[] for event in trial['events']: event=dict(event) profile=event.pop('victim_profile_before',None) if profile:event['victim_context']={k:profile.get(k) for k in ('actor','health_type','armor','defenses','immortality_mode','invulnerable','effects')} row['events'].append(event) row['actions']=[] for step in trial['steps']: response=step.get('response',{}) action=dict(command=step['command'],passed=step.get('passed'),weapon_observed=step.get('weapon_observed'),ticks_observed=response.get('ticks_observed'),terminal=response.get('terminal')) if 'sampling' in response: action['sampling']={k:v for k,v in response['sampling'].items() if k!='samples'} action['sampling']['samples_in_full_result']=True row['actions'].append(action) row['wait_sampling']=[{k:v for k,v in sample.items() if k!='samples'} for sample in trial.get('wait_samples',[])] out['trials'].append(row) return out
def verify_evidence(path, current_sources=False): path=Path(path).resolve() manifest=json.loads(path.read_text()) failures=[] for entry in manifest['files']: target=(path.parent/entry['path']).resolve() if not target.is_relative_to(path.parent): failures.append(dict(path=entry['path'],reason='path_outside_evidence'));continue if not target.is_file():failures.append(dict(path=entry['path'],reason='missing')) elif sha(target)!=entry['sha256']:failures.append(dict(path=entry['path'],reason='hash_mismatch')) if current_sources and entry.get('source_path'): source=Path(entry['source_path']) if not source.is_file() or sha(source)!=entry['sha256']: failures.append(dict(path=str(source),reason='current_source_differs')) return dict(ok=not failures,manifest=str(path),checked=len(manifest['files']),failures=failures, scope='Content integrity, not proof of experiment semantics or combat-model satisfaction.')
def run(plan_path, stop_after=None): plan=validate(json.loads(Path(plan_path).read_text())) native.state() from . import lifecycle installation=lifecycle.verify() lab.require_paused() folder=native.legacy.ARTIFACTS/('experiment-'+str(time.time_ns()));folder.mkdir(parents=True) manifest=[] def save(name,value): path=folder/name;path.parent.mkdir(parents=True,exist_ok=True) native.legacy.atomic_json(path,value) return str(path) def copy(name,source): path=folder/name;path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(Path(source).read_bytes()) manifest.append(dict(path=name,sha256=sha(path),source_path=str(Path(source).resolve()))) save('plan.json',plan) launchers=[native.ROOT/name for name in ('w3dev','cli.py','native.py') if (native.ROOT/name).is_file()] for p in launchers+sorted((native.ROOT/'witcher_devkit').glob('*.py'))+sorted((native.ROOT/'scripts').glob('*.ws.in')): copy('inputs/devkit/'+p.relative_to(native.ROOT).as_posix(),p) for p in sorted((native.legacy.manage.GAME/'mods').rglob('*')): if p.is_file():copy('inputs/mods/'+str(p.relative_to(native.legacy.manage.GAME/'mods')),p) for name in ('deployment.json','native/deployment.json'): copy('inputs/'+name,native.STATE_ROOT/name) copy('inputs/templates.json',native.ROOT/'fixtures/templates.json') fixtures=[p for p in (native.legacy.manage.DOCS/'gamesaves').iterdir() if p.suffix.lower()=='.sav' and p.stem.lower()==plan['fixture'].lower()] if len(fixtures)!=1:raise DiagnosticError('E7001','Exact fixture file could not be resolved uniquely for hashing.') fixture=dict(name=plan['fixture'],path=str(fixtures[0]),sha256=sha(fixtures[0])) report=dict(version=2,ok=False,fixture=fixture,session=native.state()['session'],build=native.state()['build'], telemetry_version=2,plan=plan,trials=[],artifacts=str(folder), installation_verification=installation,game_executable=json.loads((native.STATE_ROOT/'native/deployment.json').read_text())['executable'], uncontrolled=['Native random draws','Actor AI decisions and pathing','Observed script ticks do not imply deterministic rendered frames', 'Save-load readiness polling may allow a small variable number of initial unpaused ticks'], model_validation=False) # A ready loaded player is required so the original runtime controls are observable. original={k:v for k,v in lab.data('lab_state')['data'].items() if k!='owned'} save('recovery.json',dict(status='running',original_settings=original,fixture=fixture,session=report['session'])) failure=None try: for repetition in range(plan.get('repeats',1)): for index, trial in enumerate(plan['trials']): result=dict(name=trial['name'],repetition=repetition,command_success=False, experimental_validity=False,outcome='inconclusive',steps=[],events=[],policy=trial.get('policy'), declared_variable=trial.get('declared_variable')) report['trials'].append(result) try: loaded=native.load(plan['fixture']) if not loaded['ok']:raise DiagnosticError('E7001','Fixture load failed.',loaded) lab.data('pause') result['generation']=loaded['generation'] result['settings']=configure({**original,**trial.get('settings',{})}) result['arrival']=lab.teleport(trial['arena']) if not result['arrival']['ok']: result.update(outcome='failed',reason='arrival_precondition_failed');continue enemy=trial['enemy'] result['spawn']=lab.spawn(enemy['template'],enemy['level'],enemy.get('count',1),enemy.get('distance',4)) if not result['spawn']['ok']: result.update(outcome=result['spawn']['outcome'],reason=result['spawn']['reason']);continue result['before']=lab.profiles() result['before_snapshot']=lab.snapshot() generation=result['generation'];cursor=lab.events(generation=generation)['target_sequence'] complete=True for action in trial['actions']: started=time.monotonic() if 'wait_input_ready' in action or 'wait_counter_window' in action: elapsed=0;budget=next(iter(action.values())) counter='wait_counter_window' in action target_ref=f"{generation}|{result['spawn']['observed'][0]['actor']['id']}" start_window=(lab.profile(target_ref)['data'].get('initialization') or {}).get('counter_window_start_seconds') if counter else None if counter and start_window is None:raise DiagnosticError('E7001','Counter-window observation unavailable in this build.') while True: ready=lab.snapshot()['data']['world'] observed_window=lab.profile(target_ref)['data']['initialization']['counter_window_start_seconds'] if counter else None condition=observed_window>start_window if counter else ready['input_allowed'] if condition or not ready['player']['alive'] or elapsed>=budget:break ticks=min(5,budget-elapsed) response=trial_action(trial,'advance',str(ticks)) if 'sampling' in response: result.setdefault('wait_samples',[]).append(response['sampling']) if 'sampling' in response: complete = complete and response['sampling']['complete'] if not response['ok']:raise DiagnosticError('E7001','Readiness advance rejected.',response) collected=lab.events(cursor,generation) result['events'].extend(collected['events']);cursor=collected['next_cursor'] complete=complete and collected['complete'];elapsed+=ticks result['steps'].append(dict(command={'advance':ticks},declared_wait=action,response=response,events=collected,profiles=lab.profiles(),snapshot=lab.snapshot())) result['steps'].append(dict(command=action,elapsed_ticks=elapsed,passed=condition,start_window=start_window,observed_window=observed_window,profiles=lab.profiles(),snapshot=lab.snapshot())) if not condition: result.update(reason='wait_precondition_failed',outcome='failed');break continue if 'require_weapon' in action: observed=lab.snapshot()['data']['world']['weapon'] expected={'steel':1,'silver':2}[action['require_weapon']] result['steps'].append(dict(command=action,weapon_observed=observed,passed=observed==expected,profiles=lab.profiles(),snapshot=lab.snapshot())) if observed!=expected: result.update(reason='weapon_precondition_failed',outcome='failed') break continue command='input' if 'input' in action else 'advance' args=action.get('input',str(action.get('advance'))) response=trial_action(trial,command,args) if 'sampling' in response: complete = complete and response['sampling']['complete'] if not response['ok']: result['rejected_action']=dict(command=action,response=response) raise DiagnosticError('E7001','Bounded action rejected.',response) collected=lab.events(cursor,generation) result['events'].extend(collected['events']);cursor=collected['next_cursor'] complete=complete and collected['complete'] result['steps'].append(dict(command=action,response=response,events=collected, wall_seconds=time.monotonic()-started,profiles=lab.profiles(),snapshot=lab.snapshot())) if not response['after']['data']['world']['player']['alive']: result['terminal']='player_death';break owned_profiles=result['steps'][-1]['profiles']['profiles'][1:] if owned_profiles and all(not p['actor']['alive'] for p in owned_profiles): result['terminal']='owned_enemies_dead';break if result.get('reason') in ('weapon_precondition_failed','wait_precondition_failed'):continue result['after']=lab.profiles() result['geometry']=[geometry(result['before'],result['before_snapshot'])]+[geometry(x['profiles'],x['snapshot']) for x in result['steps']] player_id=result['before']['profiles'][0]['actor']['id'] enemy_ids=[p['actor']['id'] for p in result['spawn']['observed']] result['summary']=summarize_events(result['events'],player_id,enemy_ids) result['assertions']=assertions_result(trial.get('assertions',{}),result['events'],player_id,enemy_ids) result['trace_complete']=complete interference=bool(result['summary']['environmental_interference']) result.update(command_success=True,experimental_validity=complete and not interference) result['outcome']='inconclusive' if not result['experimental_validity'] or not result['assertions'] else 'passed' if all(a['passed'] for a in result['assertions']) else 'failed' result['reason']='trace_incomplete' if not complete else 'environmental_interference' if interference else 'no_assertions_declared' if not result['assertions'] else None except (OSError,RuntimeError,ValueError) as error: result.update(reason=str(error),outcome='inconclusive') if isinstance(error,DiagnosticError):result['diagnostic']=error.as_dict() if hasattr(error,'as_dict') else dict(code=error.code,message=str(error),evidence=error.evidence) if native.state().get('uncertain'):raise finally: save(f'trials/{repetition}-{index}.json',result) save('result.json',report) report['ok']=True except (OSError,RuntimeError,ValueError) as error: failure=str(error);report['error']=failure finally: try: reset=native.load(plan['fixture']) if not reset.get('ok'):raise RuntimeError('Final fixture reset failed') lab.data('pause') restored=configure(original) report['settings_restoration']=dict(ok=True,observed=restored) except (OSError,RuntimeError,ValueError) as error: report['settings_restoration']=dict(ok=False,error=str(error));report['ok']=False if plan.get('stop_after',True) if stop_after is None else stop_after: try: report['stop']=native.stop() if not report['stop'].get('save_files_unchanged'):report['ok']=False except (OSError,RuntimeError,ValueError) as error: report['stop']=dict(ok=False,error=str(error));report['ok']=False report['comparisons']=comparisons(plan,report['trials'],original) report['outcomes']=dict(__import__('collections').Counter(t['outcome'] for t in report['trials'])) save('result.json',report) save('condensed.json',condensed(report)) save('recovery.json',dict(status='completed' if report['ok'] else 'needs_inspection',original_settings=original, fixture=fixture,restoration=report.get('settings_restoration'),stop=report.get('stop'))) inputs={e['path'] for e in manifest} for p in sorted(folder.rglob('*')): if p.is_file() and str(p.relative_to(folder)) not in inputs: manifest.append(dict(path=str(p.relative_to(folder)),sha256=sha(p))) save('manifest.json',dict(version=1,files=manifest)) return dict(ok=report['ok'],outcomes=report['outcomes'],result=str(folder/'result.json'),manifest=str(folder/'manifest.json'), condensed=str(folder/'condensed.json'),command_success=report['ok'],all_trials_passed=all(t['outcome']=='passed' for t in report['trials']), model_validation=False,stop=report.get('stop'),error=failure)