Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421"""Self-describing terminal interface; imports no third-party packages."""import argparseimport difflibimport fcntlimport jsonimport mathimport reimport shutilimport subprocessimport sys
from . import catalogfrom . import diagnostics as diagfrom . import nativefrom . import debug_cli
debug_cli.register()
class Parser(argparse.ArgumentParser): def error(self, message): invalid = re.search(r"invalid choice: '([^']+)'", message) if invalid: matches = difflib.get_close_matches(invalid[1], catalog.CATALOG, n=1) if matches: message += '\nDid you mean ' + matches[0] + '?' raise diag.DiagnosticError('E1001', message + '\n' + self.format_usage() + 'Read ' + self.prog + ' --help for arguments and examples.')
def parser(): p = Parser(prog='w3dev', formatter_class=argparse.RawDescriptionHelpFormatter, description='Control and inspect the installed Witcher 3 game through its native bridge.', epilog='Start: guide quickstart\nDiscover: commands --json | help input | call spawn --help\n' 'Diagnose: doctor | explain E3003\n' 'Operations emit JSON; exit 0=success, 1=failure, 2=usage error.\n' 'One agent owns the live session. Never blindly replay a timed-out action.') subs = p.add_subparsers(dest='operation', required=True) parsers = {} for name, spec in catalog.CATALOG.items(): if name.startswith(('call ', 'debug ')): continue parsers[name] = subs.add_parser(name, help=spec['summary'], description=catalog.render(spec), formatter_class=argparse.RawDescriptionHelpFormatter) debug_cli.parser(parsers['debug']) parsers['migrate-state'].add_argument('--from', dest='legacy', required=True, help='Existing closed-game checkout; never a second installation.') parsers['migrate-state'].add_argument('--apply', action='store_true', help='Move verified state and publish global configuration. Default is read-only planning.') update = parsers['reconcile-game-update'].add_mutually_exclusive_group() update.add_argument('--apply', action='store_true') update.add_argument('--rollback', metavar='ARTIFACT') a = parsers['call'] calls = a.add_subparsers(dest='command', required=True) for name in sorted(native.COMMANDS): spec = catalog.CATALOG['call ' + name] a = calls.add_parser(name, help=spec['summary'], description=catalog.render(spec), formatter_class=argparse.RawDescriptionHelpFormatter) a.add_argument('args', nargs='?', default='', help=spec['arguments']) for operation in ('ui-tree', 'ui-find', 'ui-dispatch'): a = parsers[operation] a.add_argument('--root', default='', help='Deepest menu by default, or popup:NAME / hud:NAME.') if operation != 'ui-tree': a.add_argument('field', choices=('label', 'text', 'name', 'class', 'path')) a.add_argument('value', help='Exact case-sensitive selector value.') a.add_argument('--timeout', type=float, default=10, help='Seconds per selector wait, >0 and <=60 (default 10).') if operation == 'ui-dispatch': a.add_argument('--expect-label', help='Wait for this unique label after dispatch; never repeats input.') parsers['perf-launch'].add_argument('--proton', required=True, help='Active Steam Proton directory; temporary app-scoped user_settings.py extension.') for op in ('perf-start', 'perf-window'): a = parsers[op] a.add_argument('--seconds', type=float, required=True) a.add_argument('--arm', required=True) a.add_argument('--threshold-ms', type=float, action='append', required=True) a.add_argument('--budget-ms', type=float) parsers['perf-window'].add_argument('--warmup', type=float, default=3) parsers['perf-window'].add_argument('--pause-seconds', type=float, default=0) parsers['perf-mark'].add_argument('name') parsers['perf-mark'].add_argument('--phase', choices=('begin','end','event'), default='event') parsers['perf-mark'].add_argument('--kind', default='custom') parsers['perf-compare'].add_argument('plan') parsers['perf-verify'].add_argument('report') parsers['load'].add_argument('filename', help='Exact filename from call saves.') parsers['input'].add_argument('args', help='Quoted move_x|move_y|look_x|look_y|button|ticks; use -- before negative payload.') parsers['advance'].add_argument('ticks', type=int, help='Script ticks, 1..600.') parsers['evaluate'].add_argument('path', help='UTF-8 WitcherScript function body returning JSON, <=16KiB.') parsers['evaluate'].add_argument('--scoped', action='store_true', help='Restore previous source and compile it after the probe; emit recovery receipt.') parsers['evaluation-recover'].add_argument('path') parsers['experiment'].add_argument('plan') parsers['experiment'].add_argument('--keep-open', action='store_true') parsers['verify-evidence'].add_argument('manifest') parsers['verify-evidence'].add_argument('--current-sources', action='store_true') parsers['profile'].add_argument('reference') parsers['asset-inspect'].add_argument('path', help='Exact depot path, forward or backward slashes; reads installed bundles only.') parsers['combat-state'].add_argument('references', nargs='*', help='generation|actor_id; default Geralt plus owned spawns (maximum 24 selected).') parsers['encounter'].add_argument('reference', help='generation|entity_id from encounter discovery/parent link, a separate registry from actor IDs.') a = parsers['encounters'] a.add_argument('--radius', type=float, default=100) a.add_argument('--tag', default='') a.add_argument('--limit', type=int, default=16) a.add_argument('--generation', type=int) a.add_argument('--revision', type=int) a.add_argument('--start', type=int, default=0) a = parsers['sample'] a.add_argument('--ticks', type=int, default=60) a.add_argument('--input', help='Full input payload; final ticks must equal --ticks. Otherwise advance without input.') a.add_argument('--every', type=int, default=2) a.add_argument('--actor', action='append', default=[], help='generation|actor_id, repeatable; Geralt always included.') parsers['ability'].add_argument('name') a = parsers['spawn'] a.add_argument('template'); a.add_argument('--level', required=True, type=int) a.add_argument('--count', type=int, default=1); a.add_argument('--distance', type=float, default=20) a.add_argument('--max-ticks', type=int, default=180); a.add_argument('--step', type=int, default=15) a.add_argument('--stable-samples', type=int, default=3) a = parsers['teleport']; a.add_argument('position'); a.add_argument('--tolerance', type=float, default=.35) a.add_argument('--max-ticks', type=int, default=120); a.add_argument('--step', type=int, default=5) a = parsers['events']; a.add_argument('--after', type=int, default=0) a.add_argument('--generation', type=int); a.add_argument('--limit', type=int, default=128) for name, description in ( ('commands', 'List all command contracts; --json includes full specifications.'), ('help', 'Read a command contract without accessing the game.'), ('guide', 'Read an embedded workflow: quickstart, combat, ui, recovery, protocol.'), ('explain', 'Explain a stable diagnostic code and its recovery steps.'), ): a = subs.add_parser(name, help=description, description=description) a.add_argument('--json', action='store_true', help='Emit machine-readable documentation.') if name == 'help': a.add_argument('topic', nargs='+', help='Command words, e.g. call spawn.') elif name == 'guide': a.add_argument('topic', nargs='?', choices=sorted(catalog.GUIDES), default='quickstart') elif name == 'explain': a.add_argument('code', help='Stable code, e.g. E3003.') return p
def lookup(mapping, key, what): if key not in mapping: matches = difflib.get_close_matches(key, mapping, n=3) hint = ' Did you mean ' + ', '.join(matches) + '?' if matches else '' raise diag.DiagnosticError('E1001', f'Unknown {what}: {key}.{hint} Available: ' + ', '.join(mapping)) return mapping[key]
def documentation(a): if a.operation == 'commands': value = dict(ok=True, contract_version=catalog.CONTRACT_VERSION, discovery=['commands --json', 'help <command> --json', 'guide <topic> --json', 'explain CODE --json'], guides=list(catalog.GUIDES), diagnostics=list(diag.ERRORS), commands=list(catalog.CATALOG.values())) human = '\n'.join(f"{name:18} {spec['summary']}" for name, spec in catalog.CATALOG.items()) human += '\n\nRead: help <command> | call <command> --help | guide quickstart | explain CODE' elif a.operation == 'help': spec = lookup(catalog.CATALOG, ' '.join(a.topic), 'command') value = dict(ok=True, contract_version=catalog.CONTRACT_VERSION, command=spec) human = catalog.render(spec) elif a.operation == 'guide': value = dict(ok=True, topic=a.topic, text=catalog.GUIDES[a.topic]) human = value['text'] else: summary, recovery = lookup(diag.ERRORS, a.code, 'diagnostic code') value = dict(ok=True, code=a.code, summary=summary, recovery=recovery) human = a.code + ': ' + summary + '\n\n' + '\n'.join(recovery) print(json.dumps(value) if a.json else human) return 0
def doctor(): """Read-only local inspection, usable without receipts, session or network.""" from . import lifecycle checks = []
def check(name, ok, detail, code='E2001'): item = dict(name=name, ok=bool(ok), detail=str(detail)) if not ok: item['code'] = code checks.append(item)
check('game_directory', native.legacy.manage.GAME.is_dir(), native.legacy.manage.GAME) check('documents_directory', native.legacy.manage.DOCS.is_dir(), native.legacy.manage.DOCS) check('source_root', True, native.ROOT) check('configuration', True, native.legacy.manage.CONFIG) check('installation_state', True, native.STATE_ROOT) for tool in ('steam', 'xdotool', 'spectacle'): path = shutil.which(tool) check(tool, path is not None, path or 'Executable not found in PATH.') try: result = lifecycle.verify() check('installed_dependencies', True, f"{result['verified_files']} files verified; build {result['build']}") except (OSError, ValueError, RuntimeError, KeyError) as error: check('installed_dependencies', False, error, diag.classify(error)) session = {'recorded': native.STATE.exists(), 'live': False} if native.STATE.exists(): try: s = json.loads(native.STATE.read_text()) session.update(live=native.legacy.process_identity(s['process']['pid']) == s['process'], uncertain=s.get('uncertain'), session=s['session']) except (OSError, ValueError, KeyError, TypeError) as error: session['error'] = str(error) return dict(ok=all(c['ok'] for c in checks), checks=checks, session=session, scope='Local files/tools/process identity only; no engine requests or writes.', next='guide quickstart' if all(c['ok'] for c in checks) else 'guide recovery')
def preflight(a): """Validate arguments before opening a lock, snapshot or game connection.""" try: if a.operation == 'debug': if a.debug_command not in ('open','attach','status','recover','operation','cancel','evidence-verify','export','experiment'): debug_cli.operation(a) elif a.operation == 'call': native.validate(a.command, a.args) elif a.operation in ('input', 'load', 'advance'): value = a.args if a.operation == 'input' else a.filename if a.operation == 'load' else str(a.ticks) native.validate(a.operation, value) elif a.operation == 'spawn': native.validate('spawn', f'{a.template}|{a.count}|{a.level}|{a.distance}') elif a.operation == 'teleport': native.validate('teleport', a.position) if not math.isfinite(a.tolerance) or not 0 < a.tolerance <= 5: raise ValueError('--tolerance must be finite metres >0 and <=5') elif a.operation == 'profile': native.validate('profile', a.reference) elif a.operation == 'asset-inspect': path=a.path.replace('\\','/') if not re.fullmatch(r'[A-Za-z0-9_./-]{1,512}',path) or path.startswith('/') or '..' in path.split('/'): raise ValueError('Expected a relative depot path with no parent traversal') elif a.operation == 'encounter': native.validate('encounter', a.reference) elif a.operation in ('combat-state', 'sample'): refs = a.references if a.operation == 'combat-state' else a.actor if len(refs) > 24: raise ValueError('At most 24 selected actors plus Geralt') for ref in refs: native.validate('actor', ref) if refs and len({r.split('|')[0] for r in refs}) != 1: raise ValueError('All references must use the same world generation') if a.operation == 'sample': native.validate('sample_arm', f'0|owned|{a.every}|{a.ticks}') if a.input: native.validate('input', a.input) if int(a.input.split('|')[-1]) != a.ticks: raise ValueError('Input payload ticks must equal --ticks') elif a.operation == 'encounters': native.validate('encounter_scan', f'{a.radius}|{a.tag}') if (a.generation is None) != (a.revision is None): raise ValueError('--generation and --revision must be supplied together for a cached page') if a.start and a.revision is None: raise ValueError('--start requires --generation and --revision from the first page') native.validate('encounter_page', f'{a.generation if a.generation is not None else 0}|{a.revision if a.revision is not None else 1}|{a.start}|{a.limit}') elif a.operation == 'ability': native.validate('ability', a.name) elif a.operation == 'events': if not 0 <= a.after <= 2147483646 or not 1 <= a.limit <= 128 or (a.after and a.generation is None): raise ValueError('--after must be nonnegative, --limit 1..128, --generation required with nonzero --after') if a.generation is not None and not 0 <= a.generation <= 2147483646: raise ValueError('--generation must be an unsigned 32-bit script integer') elif a.operation.startswith('ui-'): native.validate('ui_scan', a.root) if hasattr(a, 'timeout') and (not math.isfinite(a.timeout) or not 0 < a.timeout <= 60): raise ValueError('--timeout must be a finite number >0 and <=60 seconds') if a.operation == 'experiment': from . import runner runner.validate(json.loads(__import__('pathlib').Path(a.plan).read_text())) if a.operation in ('spawn', 'teleport'): if not 1 <= a.max_ticks <= 600 or not 1 <= a.step <= 120: raise ValueError('--max-ticks must be 1..600 and --step 1..120') if a.operation == 'spawn' and not 2 <= a.stable_samples <= 10: raise ValueError('--stable-samples must be 2..10') except ValueError as error: name = command_name(a) example = catalog.CATALOG[name]['examples'][0] raise diag.DiagnosticError('E1001', str(error) + '\nExample syntax: ' + catalog.PREFIX + example + '\nFull contract: ' + catalog.PREFIX + 'help ' + name) from error
def command_name(a): return a.operation + (' ' + a.command if a.operation == 'call' else ' ' + a.debug_command if a.operation == 'debug' else '')
def dispatch(a): if a.operation.startswith('perf-'): from . import performance as perf if a.operation == 'perf-status': return perf.capability() if a.operation == 'perf-launch': return perf.launch(a.proton) if a.operation == 'perf-restore': return perf.restore() if a.operation == 'perf-start': return perf.start(a.seconds,a.arm,a.threshold_ms,a.budget_ms) if a.operation == 'perf-stop': return perf.stop() if a.operation == 'perf-window': return perf.window(a.seconds,a.warmup,a.arm,a.threshold_ms,a.budget_ms,a.pause_seconds) if a.operation == 'perf-compare': return perf.compare(a.plan) if a.operation == 'perf-verify': return perf.verify_report(a.report) if a.operation == 'perf-mark': if not perf.ACTIVE.exists() or perf.read(perf.ACTIVE).get('finished'):raise RuntimeError('No active performance capture') if len(a.name)>128 or len(a.kind)>64:raise ValueError('Marker name/kind exceed 128/64 characters') perf.marker('window_'+a.phase if a.phase!='event' else 'annotation',dict(name=a.name,kind=a.kind)) return dict(ok=True,marker=a.name,phase=a.phase) if a.operation == 'asset-inspect': from . import asset_inspection return asset_inspection.inspect(a.path) if a.operation in ('combat-state', 'encounters', 'encounter', 'sample'): from . import encounters if a.operation == 'combat-state': return encounters.combat_state(a.references) if a.operation == 'encounter': return encounters.lab.data('encounter', a.reference) if a.operation == 'encounters': return encounters.discover(a.radius, a.tag, a.limit, a.generation, a.revision, a.start) return encounters.sampled_action('input' if a.input else 'advance', a.input or str(a.ticks), a.every, a.actor) if a.operation == 'experiment': from . import runner return runner.run(a.plan, stop_after=not a.keep_open) if a.operation == 'verify-evidence': from . import runner return runner.verify_evidence(a.manifest,a.current_sources) if a.operation == 'reset-game-settings': from . import game_settings return game_settings.reset() if a.operation.startswith('background-'): from . import background return getattr(background, a.operation.removeprefix('background-'))() if a.operation.startswith('workspace-'): from . import workspace return getattr(workspace, a.operation.removeprefix('workspace-'))() if a.operation == 'attach': return native.attach() if a.operation == 'call': return native.call(a.command, a.args) if a.operation == 'load': return native.load(a.filename) if a.operation == 'input': return native.bounded_action('input', a.args) if a.operation == 'advance': return native.bounded_action('advance', str(a.ticks)) if a.operation == 'evaluate': if a.scoped: from . import evaluation return evaluation.scoped(a.path) return native.evaluate(a.path) if a.operation == 'evaluation-recover': from . import evaluation return evaluation.recover(a.path) if a.operation in ('profiles', 'profile', 'ability', 'spawn', 'teleport', 'events', 'templates'): from . import experiments as lab if a.operation == 'profile':return lab.profile(a.reference) if a.operation == 'ability':return lab.data('ability', a.name) if a.operation == 'spawn':return lab.spawn(a.template, a.level, a.count, a.distance, a.max_ticks, a.step, a.stable_samples) if a.operation == 'teleport':return lab.teleport(a.position, a.tolerance, a.max_ticks, a.step) if a.operation == 'events':return lab.events(a.after, a.generation, a.limit) return getattr(lab, a.operation)() if a.operation == 'wait-ready': return native.wait_ready() if a.operation == 'status': return {'ok': True, 'historical': True, 'session': json.loads(native.STATE.read_text())} if a.operation == 'recover': s = native.state() s['uncertain'] = None native.legacy.atomic_json(native.STATE, s) return {'ok': True, 'replayed': False} if a.operation == 'launch': from . import game_settings return game_settings.launch() if a.operation == 'deploy': return native.legacy.deploy() if a.operation == 'stage': from . import stage return stage.stage() if a.operation in ('verify', 'install-native', 'install-ui', 'uninstall'): from . import lifecycle return getattr(lifecycle, a.operation.replace('-', '_'))() if a.operation == 'reconcile-game-update': from . import compatibility return compatibility.rollback(a.rollback) if a.rollback else compatibility.reconcile(a.apply) if a.operation in ('ui-tree', 'ui-find', 'ui-dispatch'): from . import ui_driver if a.operation == 'ui-tree': return ui_driver.tree(a.root) if a.operation == 'ui-find': return ui_driver.find(a.field, a.value, a.root, a.timeout) return ui_driver.dispatch(a.field, a.value, a.root, a.timeout, a.expect_label) return getattr(native, a.operation)()
def context(): try: s = json.loads(native.STATE.read_text()) return dict(historical=True, **{k: s.get(k) for k in ('session', 'request', 'generation', 'uncertain')}) except (OSError, ValueError, TypeError, AttributeError): return {}
def main(argv=None): argv = sys.argv[1:] if argv is None else argv name = ' '.join(argv[:2]) if argv[:1] == ['call'] else argv[0] if argv else 'commands' try: a = parser().parse_args(argv) name = command_name(a) if a.operation in ('commands', 'help', 'guide', 'explain'): return documentation(a) if a.operation == 'migrate-state': from .installation_state import migrate result=migrate(a.legacy, apply=a.apply) print(json.dumps(result)) return 0 preflight(a) if a.operation not in ('debug', 'doctor', 'verify', 'templates', 'verify-evidence', 'perf-verify'): debug_cli.debugger.reject_legacy() if a.operation == 'debug': result = debug_cli.dispatch(a) elif a.operation == 'doctor': result = doctor() elif a.operation in ('verify','background-status','workspace-status','templates','verify-evidence','perf-status','perf-verify'): result = dispatch(a) else: native.legacy.manage.state_directory() with (native.STATE_ROOT / '.lock').open('a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) result = dispatch(a) failure = diag.rejected(result) if failure: return diag.emit(failure, name, context()) print(json.dumps(result)) return 0 except (OSError, ValueError, RuntimeError, KeyError, subprocess.TimeoutExpired) as error: return diag.emit(error, name, context())
if __name__ == '__main__': sys.exit(main())