Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184"""Completion-aware encounter setup and read-only native inspection."""import collectionsimport contextvarsimport jsonimport mathfrom pathlib import Path
from . import nativefrom .diagnostics import DiagnosticError, rejected
# A persistent owner may supply journaled/deadline-bound calls without taking# another lock. Context-local adapters leave standalone callers unchanged.OWNER_CALL = contextvars.ContextVar('lab_owner_call', default=None)OWNER_STEP = contextvars.ContextVar('lab_owner_step', default=None)
def bounded_action(command, args): adapter = OWNER_STEP.get() return adapter(command, args) if adapter else native.bounded_action(command, args)
def data(command, args=''): adapter = OWNER_CALL.get() reply = adapter(command, args) if adapter else native.call(command, args, timeout=15) failure = rejected(reply) if failure: raise failure return reply
def snapshot(): return data('snapshot')
def profile(reference): r = data('profile', reference) value = r['data'] initialized=value.get('ability_initialized', False) value['combat_attributes_available']=initialized if not initialized: if 'capabilities' in value: for attr in value['capabilities']['attributes']: attr['native_raw'], attr['native_effective'] = attr['raw'], attr['effective'] attr['raw'], attr['effective'] = None, None attr['reason'] = 'ability_manager_not_initialized' for key in ('armor','armor_terms','attack_power','spell_power','level_attribute','defenses'): value['native_'+key]=value.get(key) value[key]=None value['ability_counts'] = dict(collections.Counter(value['abilities'])) value['ability_counts_including_inventory'] = dict(collections.Counter(value['abilities_including_inventory'])) value['semantics'] = { 'ability_counts': 'Native GetAbilities(includeInventoryAbl=false) multiplicities; not a reconstructed contribution sum.', 'inventory_abilities': 'Native GetAbilities(includeInventoryAbl=true); overlaps the first list. Do not add the lists.', 'defenses': 'Final actor typed points/fractions; action-specific effective reductions are recorded separately.', 'template_path': 'Recorded LoadResource.GetPath for owned spawns; unavailable for pre-existing actors.', 'group': 'Native monster classification, not nearby count or number of attackers.', 'held_weapons': 'Native held item names/categories/types. Opaque item IDs are not serialized.', } return r
def profiles(): s = snapshot() lab = data('lab_state') generation = s['generation'] ids = [s['data']['world']['player']['id']] + [a['id'] for a in lab['data']['owned'] if a] values = [profile(f'{generation}|{i}')['data'] for i in dict.fromkeys(ids)] return dict(ok=True, generation=generation, settings=lab['data'], profiles=values)
def require_paused(): s = snapshot() if not s['data']['ready'] or not s['data']['world']['devkit_paused']: raise DiagnosticError('E3002', 'Encounter setup requires ready gameplay paused through call pause.') return s
def teleport(payload, tolerance=.35, max_ticks=120, step=5): native.validate('teleport', payload) before = require_paused() requested = list(map(float, payload.split('|'))) acknowledgement = data('teleport', payload) target = acknowledgement['data'] history = [] elapsed = 0 while True: current = snapshot() if current['generation'] != before['generation']: raise DiagnosticError('E7001', 'Generation changed during teleport; dependent setup cancelled.', {'acknowledgement': acknowledgement}) position = current['data']['world']['player']['position'] distance = math.dist(position, target) history.append(dict(elapsed_ticks=elapsed, position=position, distance_to_target=distance)) if distance <= tolerance or elapsed >= max_ticks: break ticks = min(step, max_ticks-elapsed) advanced=bounded_action('advance', str(ticks)) if not advanced.get('ok',False):raise DiagnosticError('E7001','Setup advancement rejected; elapsed ticks not assumed.',advanced) elapsed += ticks valid = distance <= tolerance return dict(ok=valid, outcome='passed' if valid else 'inconclusive', experimental_validity=valid, reason=None if valid else 'arrival_not_observed_within_tick_budget', generation=before['generation'], requested=requested, navigation_target=target, observed=position, tolerance_m=tolerance, elapsed_ticks=elapsed, history=history, acknowledgement=acknowledgement)
def stability_key(value): # Position and current health may change while actors move/attack. Initialization # is judged by level, pool maxima, armor and native ability multiset only. return json.dumps(dict(level=value['actor']['level'], maximum=value['actor']['max_hp'], health_type=value['health_type'], armor=value['armor'], abilities=value['ability_counts']), sort_keys=True)
def spawn(template, level, count=1, distance=20, max_ticks=180, step=15, stable_samples=3): payload = f'{template}|{count}|{level}|{distance}' native.validate('spawn', payload) before = require_paused() ack = data('spawn', payload) generation = ack['generation'] ids = [v['id'] for v in ack['data'] if v] if generation != before['generation'] or len(ids) != count or any(i < 0 for i in ids): raise DiagnosticError('E7001', 'Spawn did not produce the requested actor identities in the original generation.', ack) history, previous, consecutive, elapsed = [], None, 0, 0 while True: observed = [profile(f'{generation}|{i}')['data'] for i in ids] keys = [stability_key(v) for v in observed] consecutive = consecutive + 1 if keys == previous else 1 previous = keys history.append(dict(elapsed_ticks=elapsed, profiles=observed)) if consecutive >= stable_samples and elapsed >= step*stable_samples or elapsed >= max_ticks: break ticks = min(step, max_ticks-elapsed) advanced=bounded_action('advance', str(ticks)) if not advanced.get('ok',False):raise DiagnosticError('E7001','Setup advancement rejected; elapsed ticks not assumed.',advanced) elapsed += ticks stable = consecutive >= stable_samples and elapsed >= step*stable_samples levels_match = all(v['actor']['level'] == level for v in observed) alive = all(v['actor']['alive'] for v in observed) initialized = all(v.get('ability_initialized', True) and v['health_type'] is not None for v in observed) valid = stable and initialized and levels_match and alive reason = None if valid else 'initialization_not_stable' if not stable else 'combat_profile_unavailable' if not initialized else 'actual_level_mismatch' if not levels_match else 'actor_not_alive' return dict(ok=valid, outcome='passed' if valid else 'unsupported' if stable and not initialized else 'failed' if stable else 'inconclusive', experimental_validity=valid, reason=reason, generation=generation, requested=dict(template=template, count=count, level=level, distance=distance), observed=observed, elapsed_ticks=elapsed, acknowledgement=ack, history=history, stability=dict(heuristic=True, samples=stable_samples, interval_ticks=step, fields=['level', 'max_hp', 'health_type', 'armor', 'ability_counts']))
def events(after=0, generation=None, limit=128): s = snapshot() if generation is None: if after: raise ValueError('--generation is required when --after is nonzero') generation = s['generation'] if generation != s['generation']: raise DiagnosticError('E7001', 'Event cursor belongs to a different world generation.') rows, cursor, gaps = [], after, [] target = None while len(rows) < limit: page = data('event_page', f'{generation}|{cursor}|{min(4,limit-len(rows))}')['data'] if target is None: target = page['latest'] if cursor > page['latest']: raise DiagnosticError('E7001', 'Event cursor is ahead of this generation; trace continuity cannot be established.') if cursor + 1 < page['oldest']: gaps.append(dict(first=cursor+1, last=page['oldest']-1)) cursor = page['oldest']-1 selected = [r for r in page['events'] if r['sequence'] <= target] if not selected: break for row in selected: if row['sequence'] != cursor+1: gaps.append(dict(first=cursor+1, last=row['sequence']-1)) cursor = row['sequence'] rows.append(row) if cursor >= target: break return dict(ok=True, generation=generation, after=after, next_cursor=cursor, target_sequence=target, complete=not gaps and cursor >= target, gaps=gaps, events=rows, more=cursor < target, scope='Events present at the first page; later events belong to the next collection.')
def templates(): return dict(ok=True, **json.loads((native.ROOT/'fixtures/templates.json').read_text()))