"""Bounded GSC1 stream decoder, PNG writer, and localhost live screen viewer.""" import binascii from collections import deque from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import math from pathlib import Path import re import secrets import struct import threading import time import webbrowser import zlib MAGIC = b'\0GSC1\r\n\x1a' HEADER = struct.Struct('<8sIIHHIIIIII') MAX_WIDTH, MAX_HEIGHT = 1264, 2780 MAX_BYTES = MAX_WIDTH * MAX_HEIGHT * 2 MAX_CHUNK = 1024 SHELL_BEGIN = b'\r\nGIULIA SHELL BEGIN\r\n' SHELL_END = b'\r\nGIULIA SHELL END\r\n' class ScreenWire: """Remove screen packets before UTF-8/PTY parsing, preserving log bytes. Headers and payloads have separate CRCs. A malformed header enters bounded resynchronization: bytes are discarded until another verified screen header, rather than treating possible pixel bytes as terminal output. """ def __init__(self): self.pending = bytearray() self.frame = None self.identity = None self.errors = 0 self.quarantine = False self.shell = False self.stops = 0 self.last_event = None def feed(self, data): self.last_event = None self.pending.extend(data) logs = bytearray() frames = [] while self.pending: if self.shell: end = self.pending.find(SHELL_END) if end >= 0: logs.extend(self.pending[:end + len(SHELL_END)]) del self.pending[:end + len(SHELL_END)] self.shell = False continue keep = 0 for n in range(1, min(len(SHELL_END), len(self.pending) + 1)): if self.pending[-n:] == SHELL_END[:n]: keep = n ready = len(self.pending) - keep logs.extend(self.pending[:ready]) del self.pending[:ready] break at = self.pending.find(MAGIC) shell_at = self.pending.find(SHELL_BEGIN) if not self.quarantine else -1 if shell_at >= 0 and (at < 0 or shell_at < at): logs.extend(self.pending[:shell_at + len(SHELL_BEGIN)]) del self.pending[:shell_at + len(SHELL_BEGIN)] self.shell = True continue if at < 0: keep = 0 for marker in (MAGIC, SHELL_BEGIN) if not self.quarantine else (MAGIC,): for n in range(1, min(len(marker), len(self.pending) + 1)): if self.pending[-n:] == marker[:n]: keep = max(keep, n) ready = len(self.pending) - keep if not self.quarantine: logs.extend(self.pending[:ready]) del self.pending[:ready] break if at: if not self.quarantine: logs.extend(self.pending[:at]) del self.pending[:at] if len(self.pending) < HEADER.size: break _, frame_id, flags, width, height, total, offset, size, crc, frame_crc, header_crc = HEADER.unpack_from(self.pending) control = flags == 2 and not any((width, height, total, offset, size, crc, frame_crc)) valid = (frame_id and (control or (flags in (0, 1) and 1 <= width <= MAX_WIDTH and 1 <= height <= MAX_HEIGHT and total == width * height * 2 and total <= MAX_BYTES and 1 <= size <= MAX_CHUNK and offset <= total and size <= total - offset)) and binascii.crc32(self.pending[8:40]) == header_crc) if not valid: self.errors += 1 self.frame = self.identity = None self.quarantine = True del self.pending[0] continue if len(self.pending) < HEADER.size + size: break payload = bytes(self.pending[HEADER.size:HEADER.size + size]) del self.pending[:HEADER.size + size] self.quarantine = False if control: self.frame = self.identity = None self.stops += 1 self.last_event = 'stop' continue if binascii.crc32(payload) != crc: self.errors += 1 self.frame = self.identity = None self.quarantine = True continue identity = (frame_id, flags, width, height, total, frame_crc) if offset == 0: self.identity, self.frame = identity, bytearray() if self.identity != identity or self.frame is None or offset != len(self.frame): self.errors += 1 self.frame = self.identity = None continue self.frame.extend(payload) if len(self.frame) == total: if binascii.crc32(self.frame) == frame_crc: frames.append((frame_id, flags, width, height, bytes(self.frame))) self.last_event = 'frame' else: self.errors += 1 self.frame = self.identity = None return bytes(logs), frames def rgb565_png(width, height, pixels): if not 1 <= width <= MAX_WIDTH or not 1 <= height <= MAX_HEIGHT or len(pixels) != width * height * 2: raise ValueError('Invalid screen dimensions/pixel length') scanlines = bytearray() for y in range(height): scanlines.append(0) row = bytearray(width * 3) for x in range(width): n = 2 * (y * width + x) pixel = pixels[n] | pixels[n + 1] << 8 r, g, b = (pixel >> 11) & 31, (pixel >> 5) & 63, pixel & 31 row[x * 3] = (r << 3) | (r >> 2) row[x * 3 + 1] = (g << 2) | (g >> 4) row[x * 3 + 2] = (b << 3) | (b >> 2) scanlines.extend(row) def chunk(name, data): return struct.pack('>I', len(data)) + name + data + struct.pack('>I', binascii.crc32(name + data)) return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', width, height, 8, 2, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(scanlines, 3)) + chunk(b'IEND', b'')) class ScreenStore: def __init__(self, directory): self.directory = Path(directory) self.lock = threading.Lock() self.png = None self.info = {'frames': 0, 'state': 'waiting for a real framebuffer frame'} def update(self, **fields): with self.lock: self.info.update(fields) def _publish_info_locked(self): self.directory.mkdir(exist_ok=True) temporary = self.directory / 'screen.json.tmp' temporary.write_text(json.dumps(self.info, indent=2), encoding='utf-8') temporary.replace(self.directory / 'screen.json') def set_state(self, state): with self.lock: self.info['state'] = state self._publish_info_locked() def receive(self, frame): frame_id, snapshot, width, height, pixels = frame png = rgb565_png(width, height, pixels) self.directory.mkdir(exist_ok=True) latest = self.directory / 'latest.png' temporary = self.directory / 'latest.tmp' temporary.write_bytes(png) temporary.replace(latest) if snapshot: (self.directory / f'screen-{frame_id:08x}.png').write_bytes(png) with self.lock: self.png = png self.info.update({'frames': self.info['frames'] + 1, 'frame_id': frame_id, 'width': width, 'height': height, 'snapshot': bool(snapshot), 'received_utc': datetime.now(timezone.utc).isoformat(), 'state': 'snapshot received' if snapshot else 'live frame received', 'latest_png': str(latest.resolve()), 'source': 'actual panel framebuffer, aspect-preserving point sampling, RGB565'}) self._publish_info_locked() return latest def _integer(value, low, high, label): if type(value) is not int or not low <= value <= high: raise ValueError(f'{label} must be an integer in {low}..{high}.') return value def keyboard_usage(code): """KeyboardEvent.code to USB HID keyboard page; layout stays on the phone.""" if isinstance(code, str) and re.fullmatch(r'Key[A-Z]', code): return ord(code[-1]) - ord('A') + 4 if isinstance(code, str) and re.fullmatch(r'Digit[0-9]', code): return 39 if code[-1] == '0' else 29 + int(code[-1]) if isinstance(code, str) and re.fullmatch(r'F(?:[1-9]|1[0-2])', code): return 57 + int(code[1:]) special = {'Enter': 40, 'Escape': 41, 'Backspace': 42, 'Tab': 43, 'Space': 44, 'Minus': 45, 'Equal': 46, 'BracketLeft': 47, 'BracketRight': 48, 'Backslash': 49, 'Semicolon': 51, 'Quote': 52, 'Backquote': 53, 'Comma': 54, 'Period': 55, 'Slash': 56, 'CapsLock': 57, 'PrintScreen': 70, 'ScrollLock': 71, 'Pause': 72, 'Insert': 73, 'Home': 74, 'PageUp': 75, 'Delete': 76, 'End': 77, 'PageDown': 78, 'ArrowRight': 79, 'ArrowLeft': 80, 'ArrowDown': 81, 'ArrowUp': 82, 'NumLock': 83, 'NumpadDivide': 84, 'NumpadMultiply': 85, 'NumpadSubtract': 86, 'NumpadAdd': 87, 'NumpadEnter': 88, 'Numpad0': 98, 'NumpadDecimal': 99, 'ContextMenu': 101, 'ControlLeft': 224, 'ShiftLeft': 225, 'AltLeft': 226, 'MetaLeft': 227, 'ControlRight': 228, 'ShiftRight': 229, 'AltRight': 230, 'MetaRight': 231} if isinstance(code, str) and re.fullmatch(r'Numpad[1-9]', code): return 88 + int(code[-1]) try: return special[code] except (KeyError, TypeError): raise ValueError('Unsupported physical keyboard key.') from None class ScreenControl: """Bounded UI mailbox. Only the terminal loop writes to the serial port. Input availability comes from the running kernel, never from a click. A UI heartbeat and the independent device lease release held input after a crash. """ LIMIT = 128 LEASE = 2.5 def __init__(self, store): self.store = store self.lock = threading.Lock() self.queue = deque() self.connection = 'diagnostic' self.available = False self.capabilities = {'touch': False, 'keyboard': False, 'buttons': False} self.enabled = False self.last_client = 0 self.next_keepalive = 0 self.touch_active = False self.settings = {'max_size': 695, 'max_fps': 1, 'control': False} self._publish('Waiting for the device input capability report') def _publish(self, message=None): fields = dict(connection=self.connection, input_available=self.available, input_capabilities=dict(self.capabilities), settings=dict(self.settings)) if message is not None: fields['input_status'] = message self.store.update(**fields) def _release(self): self.queue = deque(c for c in self.queue if not c.startswith('input.')) if self.available and self.connection == 'diagnostic': self.queue.appendleft('input.release') self.touch_active = False def set_connection(self, connection): with self.lock: self.queue.clear() self.enabled = self.settings['control'] = False self.touch_active = False self.connection = connection self._publish('Control paused: ' + connection) def shutdown(self): self.set_connection('disconnected') def observe(self, line): with self.lock: if line.startswith('GIULIA INPUT:'): values = dict(re.findall(r'(available|touch|keyboard|buttons)=([01])\b', line)) if 'available' in values: self.available = values['available'] == '1' self.capabilities = {k: values.get(k) == '1' for k in self.capabilities} if not self.available: self._release() self.enabled = self.settings['control'] = False self._publish(line) elif line.startswith('GIULIA SCREEN:'): size = re.search(r'max_size=(\d+)', line) fps = re.search(r'max_fps=(\d+)', line) if size and fps: self.settings.update(max_size=int(size[1]), max_fps=int(fps[1])) self._publish() if 'unavailable' in line or 'requires ' in line or 'invalid' in line: self.store.set_state(line) def request_status(self): with self.lock: if self.connection == 'diagnostic' and 'input.status' not in self.queue: if len(self.queue) < self.LIMIT: self.queue.append('input.status') def handle(self, payload): if not isinstance(payload, dict) or not isinstance(payload.get('action'), str): raise ValueError('Expected a screen control action.') action = payload['action'] with self.lock: if self.connection != 'diagnostic': raise ValueError('Screen controls are paused while ' + self.connection + '.') if action == 'control': enabled = payload.get('enabled') if type(enabled) is not bool: raise ValueError('enabled must be a boolean.') if enabled and not self.available: raise ValueError('Device input service has not reported ready; use a new GUI kernel.') if not enabled: self._release() self.enabled = self.settings['control'] = enabled self.last_client = time.monotonic() self._publish('Desktop control enabled' if enabled else 'Desktop control disabled') return {'queued': True, 'enabled': enabled} if action == 'release': self._release() return {'queued': True} if action == 'heartbeat': if self.enabled: self.last_client = time.monotonic() return {'queued': False} if action in ('start', 'stop', 'snapshot'): commands = ['screen.' + action] if action == 'start': commands.append('input.status') if action == 'stop': self._release() self.enabled = self.settings['control'] = False self._publish('Desktop control disabled') elif action == 'configure': size = _integer(payload.get('max_size'), 160, 2780, 'Maximum size') fps = _integer(payload.get('max_fps'), 1, 30, 'Maximum FPS') commands = [f'screen.configure {size} {fps}'] # Drop obsolete pending changes, preserving input event order. self.queue = deque(c for c in self.queue if not c.startswith('screen.configure ')) else: if not self.enabled or not self.available: raise ValueError('Enable desktop control after the device input service is ready.') commands = self._input_commands(action, payload) coalesced = action == 'touch' and payload.get('phase') == 'move' and self.queue and self.queue[-1].startswith('input.touch move ') if len(self.queue) + (0 if coalesced else len(commands)) > self.LIMIT: self._release() self.enabled = self.settings['control'] = False self._publish('Input queue full; held input released. Enable control to retry.') raise ValueError('Screen input queue full; control paused and held input released.') if coalesced: self.queue[-1] = commands[0] else: self.queue.extend(commands) if action == 'touch': self.touch_active = payload['phase'] in ('down', 'move') if action in ('touch', 'key', 'scroll'): self.last_client = time.monotonic() return {'queued': True, 'commands': len(commands)} def _input_commands(self, action, payload): if action in ('touch', 'scroll'): if not self.capabilities['touch']: raise ValueError('The device touch input service is unavailable.') x = _integer(payload.get('x'), 0, 65535, 'Touch X') y = _integer(payload.get('y'), 0, 65535, 'Touch Y') if action == 'touch': phase = payload.get('phase') if phase not in ('down', 'move', 'up', 'cancel'): raise ValueError('Unknown touch phase.') if phase == 'down' and self.touch_active: raise ValueError('A pointer gesture is already active.') if phase == 'move' and not self.touch_active: raise ValueError('A pointer gesture must start with down.') return [f'input.touch {phase} {x} {y}'] if self.touch_active: raise ValueError('Finish the pointer drag before scrolling.') dx, dy = payload.get('delta_x', 0), payload.get('delta_y', 0) if any(type(n) not in (int, float) or not math.isfinite(n) or abs(n) > 10000 for n in (dx, dy)): raise ValueError('Scroll deltas must be finite and within 10000.') # A wheel scroll is a bounded single-finger swipe. Device input is # serialized at 100Hz rather than sleeping inside an HTTP handler. end_x = max(0, min(65535, x - round(dx * 80))) end_y = max(0, min(65535, y - round(dy * 80))) return [f'input.touch down {x} {y}', *[ f'input.touch move {round(x + (end_x-x)*i/6)} {round(y + (end_y-y)*i/6)}' for i in range(1, 7)], f'input.touch up {end_x} {end_y}'] if action == 'key': pressed = payload.get('pressed') if type(pressed) is not bool: raise ValueError('pressed must be a boolean.') key = payload.get('key') if key is not None: if key not in ('home', 'back', 'power', 'volumeup', 'volumedown'): raise ValueError('Unknown device button.') if not self.capabilities['keyboard' if key == 'back' else 'buttons']: raise ValueError('The device button input service is unavailable.') return [f'input.button {key} {int(pressed)}'] if not self.capabilities['keyboard']: raise ValueError('The device keyboard service is unavailable.') usage = keyboard_usage(payload.get('code')) modifiers = _integer(payload.get('modifiers', 0), 0, 255, 'Keyboard modifiers') return [f'input.key {usage} {int(pressed)} {modifiers}'] raise ValueError('Unknown screen control action.') def poll(self, now=None): now = time.monotonic() if now is None else now with self.lock: if self.connection != 'diagnostic': return None if self.enabled and now - self.last_client >= self.LEASE: self._release() self.enabled = self.settings['control'] = False self._publish('Viewer heartbeat expired; held input released') if self.queue: return self.queue.popleft() if self.enabled and now >= self.next_keepalive: self.next_keepalive = now + 1 return 'input.keepalive' return None VIEWER_HTML_PATH = Path(__file__).with_name('debug_screen_viewer.html') class ScreenViewer: def __init__(self, store, control=None): self.store = store self.control = control or ScreenControl(store) self.token = secrets.token_urlsafe(32) viewer = self class Handler(BaseHTTPRequestHandler): def _trusted_host(handler): return handler.headers.get('Host') == f'127.0.0.1:{handler.server.server_port}' def _reply(handler, status, payload, mime='application/json'): handler.send_response(status) handler.send_header('Content-Type', mime) handler.send_header('Content-Length', str(len(payload))) handler.send_header('Cache-Control', 'no-store') handler.send_header('X-Content-Type-Options', 'nosniff') handler.send_header('Referrer-Policy', 'no-referrer') handler.send_header('Content-Security-Policy', "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; frame-ancestors 'none'; base-uri 'none'") handler.end_headers() handler.wfile.write(payload) def do_GET(handler): if not handler._trusted_host(): handler.send_error(403) return path = handler.path.split('?', 1)[0] with store.lock: if path == '/': payload = VIEWER_HTML_PATH.read_bytes().replace(b'__VIEWER_TOKEN__', viewer.token.encode('ascii')) mime = 'text/html; charset=utf-8' elif path == '/status': payload, mime = json.dumps(store.info).encode(), 'application/json' elif path == '/frame.png' and store.png is not None: payload, mime = store.png, 'image/png' else: handler.send_error(404) return handler._reply(200, payload, mime) def do_POST(handler): if handler.path != '/control': handler.send_error(404) return origin = handler.headers.get('Origin') expected = f'http://127.0.0.1:{handler.server.server_port}' token = handler.headers.get('X-Screen-Token', '') if not handler._trusted_host() or (origin is not None and origin != expected) or not secrets.compare_digest(token, viewer.token): handler.send_error(403) return try: length = int(handler.headers.get('Content-Length', '0')) if not 0 < length <= 4096 or handler.headers.get('Transfer-Encoding'): raise ValueError('Screen control payload must be 1..4096 bytes.') if handler.headers.get_content_type() != 'application/json': raise ValueError('Expected application/json.') payload = json.loads(handler.rfile.read(length)) result = viewer.control.handle(payload) except (ValueError, UnicodeError, RecursionError) as error: handler._reply(400, json.dumps({'error': str(error)}).encode()) return handler._reply(202, json.dumps(result).encode()) def log_message(handler, *_args): pass def setup(handler): super().setup() handler.connection.settimeout(3) self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.url = f'http://127.0.0.1:{self.server.server_port}/' def open(self): webbrowser.open(self.url) def close(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=2)