Jeff Luke
41ef12f954
- Add a four-second arcade-style boot sequence followed by the Character / Nickname introduction. - Separate cabinet power-on from RFID credit and player Start. - Add operator Power on control and cabinet power-state reporting. - Prevent repeated power commands from restarting boot or gameplay. - Ignore RFID scans received before the cabinet is ready. - Add power-state, RFID, controller, and GPIO feedback regression tests. - Expand the master walkthrough with export, packaging, file transfer, fresh-Pi setup, RFID compilation, and end-to-end testing instructions. - Correct tester hardware documentation: BCM23/J20 drives an LED; physical buttons connect through a separate I2C1 input board. - Document operator-triggered power-on as the current test workflow and retain the temporary 2.4 configuration compatibility notes. - Record successful physical testing of power-on, RFID credit, and player Start. - Add an Odin integration README describing the proposed Rust/NATS bridge, shared messages, Flutter GM interface, autopilot sequencing, hardware ownership, and Starfish SSH-management role. - Define migration milestones and information needed from the framework owner; link the integration guide from the root README. Validation: 69 controller tests, 10 GPIO tests, Pac-Man power/RFID checks, managed-cabinet lifecycle checks, and visual screen inspection.
115 lines
6.5 KiB
Python
115 lines
6.5 KiB
Python
"""Read-only feedback inventory; output writes remain in pi_gpio_bridge.Outputs."""
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections import deque
|
|
from pathlib import Path
|
|
|
|
SIGNALS = dict(zip(['J10','J11','J12','J13','J14','J15','J17','J18','J19','J20','J21','J22'], [5,6,13,19,26,21,16,12,24,23,18,25]))
|
|
RESERVED = {17,27,22,20,8,9,10,11}
|
|
|
|
def load_devices(path, output_pins):
|
|
data = json.loads(Path(path).read_text(encoding='utf-8-sig'))
|
|
if not isinstance(data, list) or len(data)>32: raise ValueError('Device config must be a list of up to 32 devices')
|
|
ids, ports, inputs, readers = set(), set(), set(), False
|
|
for d in data:
|
|
if not isinstance(d,dict) or not re.fullmatch(r'[A-Za-z0-9_-]{1,64}',d.get('id','')): raise ValueError('Invalid device ID')
|
|
if d['id'] in ids or d.get('port') in ports: raise ValueError('Duplicate device ID or port')
|
|
ids.add(d['id']); ports.add(d.get('port'))
|
|
if not isinstance(d.get('name',''),str) or len(d.get('name',''))>100: raise ValueError('Invalid name')
|
|
kind, port = d.get('kind'), d.get('port')
|
|
if kind=='rfid':
|
|
if port not in ['J'+str(i) for i in range(1,9)]: raise ValueError('RFID requires J1-J8')
|
|
readers=True
|
|
elif kind in ('input','output'):
|
|
if port not in SIGNALS: raise ValueError('Digital devices require a signal connector')
|
|
d['pin']=SIGNALS[port]
|
|
if kind=='input':
|
|
if type(d.get('pull_up')) is not bool: raise ValueError('Inputs require explicit pull_up true/false')
|
|
inputs.add(d['pin'])
|
|
elif d['pin'] not in output_pins: raise ValueError('Output connector not enabled in output pin allowlist')
|
|
else: raise ValueError('Supported kinds: input, output, rfid')
|
|
if inputs & set(output_pins): raise ValueError('A GPIO cannot be both an input and an output')
|
|
if readers and RESERVED & (set(output_pins)|inputs): raise ValueError('RFID selector/reset/SPI pins cannot be configured as general GPIO')
|
|
return data
|
|
|
|
class Feedback:
|
|
def __init__(self, definitions, input_factory, outputs, reader_binary=None, clock=time.monotonic):
|
|
self.definitions, self.outputs, self.clock = definitions, outputs, clock
|
|
self.lock=threading.RLock(); self.stop=threading.Event(); self.values={}; self.events=deque(maxlen=64)
|
|
self.boot=uuid.uuid4().hex; self.sequence=0; self.inputs=[]; self.worker=None; self.reader=None
|
|
try:
|
|
for d in definitions:
|
|
self.values[d['id']] = dict(id=d['id'],name=d.get('name',d['id']),port=d['port'],kind=d['kind'],value=None,error='Awaiting first sample',stamp=clock())
|
|
if d['kind']=='input': self.inputs.append((d,input_factory(d['pin'],pull_up=d['pull_up'],bounce_time=.05)))
|
|
readers=[d for d in definitions if d['kind']=='rfid']
|
|
if readers:
|
|
if not reader_binary: raise ValueError('RFID devices require --rfid-helper executable')
|
|
self.reader=subprocess.Popen([str(Path(reader_binary).resolve()),','.join(str(int(d['port'][1:])-1) for d in readers)],stdout=subprocess.PIPE,stderr=None,text=True,bufsize=1)
|
|
self.worker=threading.Thread(target=self._readers,args=(readers,),daemon=True); self.worker.start()
|
|
except Exception:
|
|
self.close(); raise
|
|
def update(self, ident, value, error=''):
|
|
with self.lock:
|
|
old=self.values[ident]
|
|
changed=old['value']!=value or old['error']!=error
|
|
old.update(value=value,error=error,stamp=self.clock())
|
|
if changed:
|
|
self.sequence+=1
|
|
self.events.append(dict(id=self.boot+':'+str(self.sequence),device=ident,value=value,error=error,kind=old['kind'],stamp=self.clock()))
|
|
def _readers(self, readers):
|
|
slots={int(d['port'][1:])-1:d['id'] for d in readers}
|
|
try:
|
|
for line in self.reader.stdout:
|
|
if self.stop.is_set(): break
|
|
if len(line)>2048: continue
|
|
value=json.loads(line)
|
|
ident=slots.get(value.get('slot'))
|
|
if ident:
|
|
self.update(ident,str(value.get('uid',''))[:32],str(value.get('error',''))[:120])
|
|
except Exception as exc:
|
|
for ident in slots.values(): self.update(ident,None,'Reader helper error: '+str(exc)[:80])
|
|
finally:
|
|
if not self.stop.is_set():
|
|
for ident in slots.values(): self.update(ident,None,'Reader helper stopped')
|
|
def tick(self):
|
|
for d,device in self.inputs:
|
|
try: self.update(d['id'],bool(device.value))
|
|
except Exception as exc: self.update(d['id'],None,str(exc)[:120])
|
|
with self.outputs.lock:
|
|
for d in self.definitions:
|
|
if d['kind']!='output': continue
|
|
entry=self.outputs.pins.get(d['pin'])
|
|
try: self.update(d['id'],bool(entry['device'].value) if entry else None,'' if entry else 'Output not initialized')
|
|
except Exception as exc: self.update(d['id'],None,str(exc)[:120])
|
|
def snapshot(self):
|
|
with self.lock:
|
|
now=self.clock()
|
|
return dict(devices=[{k:v for k,v in d.items() if k!='stamp'}|{'age_ms':int(max(0,now-d['stamp'])*1000)} for d in self.values.values()],events=[{k:v for k,v in e.items() if k!='stamp'}|{'age_ms':int(max(0,now-e['stamp'])*1000)} for e in self.events])
|
|
def close(self):
|
|
self.stop.set()
|
|
if self.reader:
|
|
self.reader.terminate()
|
|
try:self.reader.wait(timeout=2)
|
|
except subprocess.TimeoutExpired:self.reader.kill();self.reader.wait()
|
|
if self.worker:self.worker.join(timeout=2)
|
|
self.reader.stdout.close()
|
|
for _,device in self.inputs: device.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
parser=argparse.ArgumentParser(description='Extract saved PCB feedback inventory from the exported room catalog')
|
|
parser.add_argument('--catalog',required=True)
|
|
parser.add_argument('--cabinet',required=True)
|
|
parser.add_argument('--output',required=True)
|
|
args=parser.parse_args()
|
|
catalog=json.loads(Path(args.catalog).read_text(encoding='utf-8-sig'))
|
|
devices=catalog['cabinets'][args.cabinet].get('devices',[])
|
|
with Path(args.output).open('x',encoding='utf-8') as file:
|
|
json.dump(devices,file,indent=2)
|
|
print('Wrote',len(devices),'device definitions. Review input polarity and output allowlist before running the bridge.')
|