# -*- coding: utf-8 -*- """Golden runs of the Millionaire game: the required state that a refactor must keep. A scenario plays the offline demo content through the real game modal (headless, or live in Kodi) with a fixed random seed and records after every step what the modern card shows: every `WCS.Game.*` window property its XML reads, the focused control and whether the window is open. Live runs also save a frame of every step. record writes scripts/game_harness/golden/..json (+ frames//NN-step.png live) verify replays and compares; frames are compared on the card only, with a small tolerance for the pulsing focus animations """ import json import os import re import time import env HERE = os.path.dirname(os.path.abspath(__file__)) GOLDEN_DIR = os.path.join(HERE, 'golden') XML = os.path.join(env.ADDON_DIR, 'resources', 'skins', 'Default', '1080i', 'games', 'DialogMilionarGameModern.xml') CARD = (100, 60, 1820, 1020) # the card in 1920x1080 skin pixels (the Home clock is outside) SETTLE_S = {'headless': 0.15, 'live': 1.2} # focus threads / animations after a step settled LONG_FEEDBACK = { 'result_announcement': 'Správně, Southampton má na svědomí víc odjezdů, než by čekal i největší vlakový fanoušek, ' 'a ta loď se odsud rozjela s tak velkým očekáváním, že by ji zastavil jen dobrý cestovní pas. ' 'Pro Cherbourg a New York jste měli smůlu, to už byly jen zastávky ve stínu hlavního hrdiny.', 'transition_text': 'Od teď to bude o trochu těžší, takže si pořádně zavažte záchranné vesty a pojďme do středního ' 'kola.'} # (name, demo title, seed, steps); a step: (label, kind, argument, screens to wait for) # kind: act (intent) | answer (right|wrong) | back | call (method, args) | set (attrs) | none SCENARIOS = { 'win-lifelines': ('matrix', 1, [ ('title', 'none', None, {'title'}), ('rules', 'act', 'rules', {'rules'}), ('rules-back', 'back', None, {'title'}), ('intro', 'act', 'start', {'intro'}), ('q1', 'act', 'start', {'question'}), ('a1', 'answer', 'right', {'feedback'}), ('q2', 'act', 'next', {'question'}), ('q2-5050', 'act', 'life5050', {'question'}), ('a2', 'answer', 'right', {'feedback'}), ('q3', 'act', 'next', {'question'}), ('a3', 'answer', 'right', {'feedback'}), ('q4', 'act', 'next', {'question'}), ('q4-tip', 'act', 'lifetip', {'question'}), ('a4', 'answer', 'right', {'feedback'}), ('q5', 'act', 'next', {'question'}), ('a5', 'answer', 'right', {'feedback'}), ('q6', 'act', 'next', {'question'}), ('q6-poll', 'act', 'lifeaudience', {'question'}), ('a6', 'answer', 'right', {'feedback'}), ('q7', 'act', 'next', {'question'}), ('a7', 'answer', 'right', {'feedback'}), ('q8', 'act', 'next', {'question'}), ('a8', 'answer', 'right', {'feedback'}), ('q9', 'act', 'next', {'question'}), ('a9-win', 'answer', 'right', {'win'}), ('results', 'act', 'results', {'results'}), ('again', 'act', 'again', {'intro'}), ]), 'loss-again-exit': ('friends', 2, [ ('title', 'none', None, {'title'}), ('intro', 'act', 'start', {'intro'}), ('q1', 'act', 'start', {'question'}), ('a1', 'answer', 'right', {'feedback'}), ('q2', 'act', 'next', {'question'}), ('a2-wrong', 'answer', 'wrong', {'gameover'}), ('again', 'act', 'again', {'intro'}), ('q1-again', 'act', 'start', {'question'}), ('exit-ask', 'back', None, {'question'}), ('exit-stay', 'act', 'exit_stay', {'question'}), ('exit-ask-2', 'back', None, {'question'}), ('exit-confirm', 'act', 'exit_confirm', {'title'}), ]), 'error-retry-long': ('titanic', 3, [ ('title', 'none', None, {'title'}), ('intro', 'act', 'start', {'intro'}), ('q1', 'act', 'start', {'question'}), ('a1', 'answer', 'right', {'feedback'}), ('long-text', 'set', {'feedback_data': LONG_FEEDBACK}, {'feedback'}), ('error', 'call', ('show_generation_error', ['Testovací chyba moderátora.']), {'error'}), ('retry', 'act', 'retry', {'intro'}), ('q1-retry', 'act', 'start', {'question'}), ('a1-wrong', 'answer', 'wrong', {'gameover'}), ('home', 'act', 'home', None), ]), } def contract(): """Every WCS.Game.* property the modern XML reads (the published state of the card).""" with open(XML, encoding='utf-8') as handle: names = set(re.findall(r'Window\.Property\((WCS\.Game\.[A-Za-z0-9_.]+)\)', handle.read())) # The new launch selector is tested by Story entry tests, separately from the legacy gameplay baseline. names -= {'WCS.Game.Mode', 'WCS.Game.StoryAvailable', 'WCS.Game.Set.story_difficulty'} return sorted(names) def _settled(driver, screens, timeout): def ready(snap): if not snap.get('alive'): return True return snap.get('settled') and (not screens or snap.get('screen') in screens) return driver.wait(ready, timeout) def _record_step(driver, names, label): snap = driver.snapshot() props = driver.command('props', names=names) if snap.get('alive') else {} return {'step': label, 'alive': bool(snap.get('alive')), 'screen': snap.get('screen'), 'game_state': snap.get('game_state'), 'focus': snap.get('focus'), 'props': props} def play(driver, name, frames_dir=None, timeout=60.0): """Run one scenario on an open demo game; returns the recorded steps.""" _title, seed, steps = SCENARIOS[name] names = contract() driver.command('seed', value=seed) out = [] for index, (label, kind, arg, screens) in enumerate(steps, 1): if kind == 'act': driver.click(arg) if driver.kind == 'live' else driver.act(arg) elif kind == 'answer': question = driver.snapshot().get('question') or {} right = question.get('correct', 0) removed = set(driver.snapshot().get('removed') or []) index_ = right if arg == 'right' else next(i for i in range(4) if i != right and i not in removed) intent = 'answer{}'.format(index_) driver.click(intent) if driver.kind == 'live' else driver.act(intent) elif kind == 'back': driver.back() elif kind == 'call': method, args = arg driver.command('call', method=method, args=args) elif kind == 'set': driver.command('set', attrs=arg) driver.command('call', method='update_sidebar_callback', args=[]) # publish like the game does env.real_sleep(0.3 if driver.kind == 'live' else 0.05) if kind == 'back' and screens == {'question'}: driver.wait(lambda s: s.get('exit_prompt') or not s.get('alive'), timeout) elif kind == 'act' and arg == 'exit_stay': driver.wait(lambda s: not s.get('exit_prompt'), timeout) else: _settled(driver, screens, timeout) env.real_sleep(SETTLE_S[driver.kind]) record = _record_step(driver, names, label) if frames_dir and record['alive']: path = os.path.join(frames_dir, '{:02d}-{}.png'.format(index, label)) driver.screenshot(path) shrink(path) record['frame'] = os.path.basename(path) out.append(record) return out def shrink(path): from PIL import Image Image.open(path).convert('RGB').resize(FRAME_SIZE, Image.LANCZOS).save(path, optimize=True) def golden_path(name, kind): return os.path.join(GOLDEN_DIR, '{}.{}.json'.format(name, kind)) def frames_dir(name): return os.path.join(GOLDEN_DIR, 'frames', name) def compare_steps(expected, actual): """Differences between two recorded runs: [(step, what, expected, actual)].""" diffs = [] for index in range(max(len(expected), len(actual))): if index >= len(expected) or index >= len(actual): diffs.append(('#{}'.format(index + 1), 'step count', len(expected), len(actual))) break want, got = expected[index], actual[index] if want['step'] != got['step']: diffs.append((want['step'], 'step', want['step'], got['step'])) continue for key in ('alive', 'screen', 'game_state', 'focus'): if want.get(key) != got.get(key): diffs.append((want['step'], key, want.get(key), got.get(key))) for prop in sorted(set(want['props']) | set(got['props'])): if want['props'].get(prop, '') != got['props'].get(prop, ''): diffs.append((want['step'], prop, want['props'].get(prop, ''), got['props'].get(prop, ''))) return diffs def frame_diff(golden_png, live_png): """(share of card pixels that differ noticeably, max channel difference) between two frames.""" from PIL import Image, ImageChops box = tuple(int(v * FRAME_SIZE[0] / 1920.0) for v in CARD) a = Image.open(golden_png).convert('RGB').resize(FRAME_SIZE, Image.LANCZOS).crop(box) b = Image.open(live_png).convert('RGB').resize(FRAME_SIZE, Image.LANCZOS).crop(box) diff = ImageChops.difference(a, b).convert('L') histogram = diff.histogram() over = sum(histogram[24:]) total = a.size[0] * a.size[1] return over / float(total), max(i for i, n in enumerate(histogram) if n) if any(histogram) else 0 FRAME_SIZE = (1280, 720) # frames are kept and compared at this size (enough for layout, ~1/3 of the bytes) FRAME_TOLERANCE = 0.004 # 0.4 % of the card may differ (pulsing focus rings, text antialiasing) def save(name, kind, steps): os.makedirs(GOLDEN_DIR, exist_ok=True) with open(golden_path(name, kind), 'w', encoding='utf-8') as handle: json.dump({'scenario': name, 'kind': kind, 'recorded': time.strftime('%Y-%m-%d %H:%M:%S'), 'steps': steps}, handle, ensure_ascii=False, indent=1) def load(name, kind): with open(golden_path(name, kind), encoding='utf-8') as handle: return json.load(handle)['steps']