686 lines
33 KiB
Python
686 lines
33 KiB
Python
"""CMD USB terminal: bounded commands, raw capture, and diagnostic bundles."""
|
|
import argparse
|
|
import codecs
|
|
from collections import deque
|
|
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from debug_screen import ScreenWire, ScreenStore, ScreenViewer, ScreenControl
|
|
|
|
DEVICE_COMMANDS = ('help', 'ping', 'status', 'replay', 'sample', 'features',
|
|
'info', 'uptime', 'memory', 'ps', 'bootargs', 'display', 'usb', 'ufs', 'shell',
|
|
'wifi.start','wifi.status','reboot.fastboot','reboot.status',
|
|
'gpu.status','gpu.observe',
|
|
'gpu.hfi.pin','gpu.hfi.status','gpu.hfi.query','gpu.hfi.poll','gpu.hfi.log',
|
|
'screen.start','screen.stop','screen.snapshot',
|
|
'input.status','input.release','input.keepalive')
|
|
TOPICS = {
|
|
'wifi': r'WIFI|PCIE|GIULIA PCI:|GIULIA SMMU:|WCN|MHI|QMI|QMP|PMU|WMI|IO80211',
|
|
'gpu': r'GPU|GMU|ADRENO|SMMU',
|
|
'touch': r'TOUCH|HID|TouchCom',
|
|
'display': r'FB:|framebuffer|display|scanout|backboard',
|
|
'ufs': r'UFS|BLOCK:|ROOT:|HFS|I/O error',
|
|
'usb': r'USB|DEBUG:|TERM:.*usb',
|
|
'process': r'SMP |pid=|launchd|SpringBoard|backboard',
|
|
'panic': r'panic|fault|exception|crash|abort',
|
|
'power': r'power|clock|GDSC|regulator|PMIC|battery',
|
|
'all': r'.',
|
|
}
|
|
HELP = '''Device commands (new kernel needed except help/ping/status/replay/sample):
|
|
info uptime memory ps bootargs display usb ufs features
|
|
help / ping / status / replay / sample (GUI sampler)
|
|
wifi.status / wifi.start inspect/release an opted-in pre-DMA wait
|
|
gpu.status retained GPU preparation/last-observation state
|
|
gpu.observe queue gated read-only power/SMMU observation
|
|
gpu.hfi.pin retain diagnostic DMA buffers until reboot (mode 3)
|
|
gpu.hfi.status inspect the 17 HWSCHED queue headers
|
|
gpu.hfi.query enqueue warmboot query; manual doorbell still required
|
|
gpu.hfi.poll drain bounded HFI reply/debug packets
|
|
gpu.hfi.log read the first 64 words of the GMU log buffer
|
|
reboot.fastboot sync disks and restart into Fastboot (debug kernel)
|
|
reboot.status inspect a pending reboot request
|
|
screen.start live screen with mouse/keyboard controls in browser
|
|
screen.stop stop screen transfer (logs and commands continue)
|
|
screen.snapshot save a real framebuffer screenshot as PNG
|
|
screen.configure N FPS maximum edge 160..2780, target FPS 1..30 (USB limited)
|
|
input.status actual touch/keyboard/button service availability
|
|
input.release release remote touches and held keyboard/buttons
|
|
Host commands:
|
|
:help this help
|
|
logs TOPIC retained session logs, NOT fresh hardware queries
|
|
wifi gpu touch display ufs usb process panic power all
|
|
all queue all bounded device queries
|
|
collect all queries + retained RAM replay + GUI sample;
|
|
saves a diagnostic bundle after 15 seconds
|
|
save save current session metadata and topic logs
|
|
watch COMMAND SECONDS periodically query (minimum 1 second)
|
|
unwatch stop polling
|
|
clear clear screen (keep logs)
|
|
exit / quit / :quit close port
|
|
shell.size COLS ROWS set PTY dimensions before entering shell
|
|
shell real PTY shell (debug kernel + staged daemon/shell)
|
|
Ctrl+C goes to phone; Ctrl+] leaves shell
|
|
reg.map BASE BYTES map explicit peripheral window (no device access)
|
|
reg.maps / reg.unmap N list / release mapped window
|
|
reg.read ADDR WIDTH one actual MMIO read (WIDTH: 8/16/32/64)
|
|
reg.write ADDR WIDTH V one actual MMIO write, no automatic readback
|
|
reg.rmw ADDR WIDTH M V read/modify/write with mask M, bits V (not atomic)
|
|
reg.dump ADDR WIDTH N read at most 64 consecutive registers
|
|
MMIO requires giulia_debug=1 giulia_mmio=1; operator manages clocks/power.
|
|
Up/Down: command history. Ctrl+C exits only outside shell.
|
|
'''
|
|
|
|
|
|
def wire_command(line):
|
|
words = line.split()
|
|
register_arity = {'reg.map': 2, 'reg.maps': 0, 'reg.unmap': 1,
|
|
'reg.read': 2, 'reg.write': 3, 'reg.rmw': 4, 'reg.dump': 3,
|
|
'shell.size': 2}
|
|
if words and words[0] in ('screen.configure', 'input.touch', 'input.key', 'input.button'):
|
|
command = words[0]
|
|
if command == 'input.button':
|
|
valid = len(words) == 3 and words[1] in ('home','back','power','volumeup','volumedown') and words[2] in ('0','1')
|
|
else:
|
|
start = 2 if command == 'input.touch' else 1
|
|
numbers = words[start:]
|
|
valid = len(words) == {'screen.configure':3, 'input.touch':4, 'input.key':4}[command]
|
|
valid = valid and all(re.fullmatch(r'[0-9]{1,5}', n) for n in numbers)
|
|
if valid:
|
|
values = [int(n) for n in numbers]
|
|
if command == 'screen.configure':
|
|
valid = 160 <= values[0] <= 2780 and 1 <= values[1] <= 30
|
|
elif command == 'input.touch':
|
|
valid = words[1] in ('down','move','up','cancel') and all(0 <= n <= 65535 for n in values)
|
|
else:
|
|
valid = 4 <= values[0] <= 231 and values[1] in (0,1) and 0 <= values[2] <= 255
|
|
if not valid:
|
|
raise ValueError('Invalid screen/input command; use :help.')
|
|
elif words and words[0] == 'gpu.dev':
|
|
if len(words)!=7 or any(not re.fullmatch(r'(?:0x[0-9a-f]+|[0-9]+)', x) for x in words[1:]):
|
|
raise ValueError('Invalid GPU development command')
|
|
if any(int(x,16 if x.startswith('0x') else 10)>0xffffffff for x in words[1:]):
|
|
raise ValueError('GPU development argument overflow')
|
|
elif words and words[0] in register_arity:
|
|
if (len(words) != register_arity[words[0]] + 1 or
|
|
any(not re.fullmatch(r'(?:0[xX][0-9a-fA-F]+|[0-9]+)', x) for x in words[1:])):
|
|
raise ValueError('Invalid register syntax; use :help.')
|
|
if any(int(x, 16 if x.lower().startswith('0x') else 10) > (1 << 64)-1 for x in words[1:]):
|
|
raise ValueError('Register argument exceeds 64 bits.')
|
|
values = [int(x,16 if x.lower().startswith('0x') else 10) for x in words[1:]]
|
|
command = words[0]
|
|
if command == 'shell.size':
|
|
if any(not x.isdecimal() for x in words[1:]) or any(not 1 <= x <= 1000 for x in values):
|
|
raise ValueError('Shell size requires decimal columns/rows in 1..1000.')
|
|
elif command in ('reg.read','reg.write','reg.rmw','reg.dump'):
|
|
address,width = values[:2]
|
|
if width not in (8,16,32,64) or address % (width//8):
|
|
raise ValueError('Width must be 8/16/32/64 and address naturally aligned.')
|
|
if command in ('reg.write','reg.rmw') and any(x >= 1 << width for x in values[2:]):
|
|
raise ValueError('Value/mask exceeds selected register width.')
|
|
if command == 'reg.rmw' and values[3] & ~values[2]:
|
|
raise ValueError('RMW bits must be contained in the mask.')
|
|
if command == 'reg.dump' and (not 1 <= values[2] <= 64 or address+(values[2]-1)*(width//8) >= 1 << 64):
|
|
raise ValueError('Dump length must be 1..64 without address overflow.')
|
|
elif command == 'reg.map':
|
|
address,size = values
|
|
if not 0x100000 <= address < 0x80000000 or not 0 < size <= 0x10000 or address+size > 0x80000000:
|
|
raise ValueError('Map must fit the low peripheral aperture, maximum 64 KiB.')
|
|
if address % 4096 or size % 4096:
|
|
raise ValueError('Map must be page aligned; the kernel enforces its actual page size.')
|
|
elif command == 'reg.unmap' and values[0] >= 16:
|
|
raise ValueError('Mapping slot must be 0..15.')
|
|
elif line not in DEVICE_COMMANDS:
|
|
raise ValueError('Unknown diagnostic command; use :help, or enter shell for userspace commands.')
|
|
if len(line) > 63 or any(ord(c)<32 or ord(c)>126 for c in line):
|
|
raise ValueError('Kernel command limit is 63 printable ASCII bytes.')
|
|
return (line + '\n').encode('ascii')
|
|
|
|
|
|
class ShellWire:
|
|
"""Incremental raw-mode markers, including every possible USB split."""
|
|
BEGIN = b'\r\nGIULIA SHELL BEGIN\r\n'
|
|
END = b'\r\nGIULIA SHELL END\r\n'
|
|
|
|
def __init__(self):
|
|
self.active = False
|
|
self.pending = b''
|
|
|
|
def feed(self, data):
|
|
self.pending += data
|
|
events = []
|
|
while self.pending:
|
|
marker = self.END if self.active else self.BEGIN
|
|
at = self.pending.find(marker)
|
|
if at >= 0:
|
|
if at:
|
|
events.append(('raw' if self.active else 'log', self.pending[:at]))
|
|
self.pending = self.pending[at+len(marker):]
|
|
self.active = not self.active
|
|
events.append(('begin' if self.active else 'end', b''))
|
|
continue
|
|
keep = 0
|
|
for length in range(1, min(len(marker), len(self.pending)+1)):
|
|
if self.pending.endswith(marker[:length]):
|
|
keep = length
|
|
ready = self.pending[:-keep] if keep else self.pending
|
|
self.pending = self.pending[-keep:] if keep else b''
|
|
if ready:
|
|
events.append(('raw' if self.active else 'log', ready))
|
|
break
|
|
return events
|
|
|
|
|
|
class WindowsConsole:
|
|
def __init__(self):
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
self.k = ctypes.WinDLL('kernel32', use_last_error=True)
|
|
self.k.GetStdHandle.argtypes = [wintypes.DWORD]
|
|
self.k.GetStdHandle.restype = wintypes.HANDLE
|
|
self.k.GetConsoleMode.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)]
|
|
self.k.SetConsoleMode.argtypes = [wintypes.HANDLE, wintypes.DWORD]
|
|
self.handles = [self.k.GetStdHandle(-10 & 0xffffffff), self.k.GetStdHandle(-11 & 0xffffffff)]
|
|
self.modes = []
|
|
for h in self.handles:
|
|
mode = wintypes.DWORD()
|
|
if not self.k.GetConsoleMode(h, ctypes.byref(mode)):
|
|
raise OSError('Raw shell requires Windows console input/output.')
|
|
self.modes.append(mode.value)
|
|
|
|
def raw(self, enabled):
|
|
if enabled:
|
|
if not self.k.SetConsoleMode(self.handles[0], self.modes[0] & ~7):
|
|
raise OSError('Cannot enable raw keyboard mode.')
|
|
if not self.k.SetConsoleMode(self.handles[1], self.modes[1] | 4):
|
|
self.restore()
|
|
raise OSError('Cannot enable VT terminal output.')
|
|
else:
|
|
self.restore()
|
|
|
|
def restore(self):
|
|
for handle, mode in zip(self.handles, self.modes):
|
|
self.k.SetConsoleMode(handle, mode)
|
|
|
|
|
|
def safe_text(text):
|
|
return ''.join(c for c in text if c in '\n\r\t' or (ord(c) >= 32 and not 127 <= ord(c) <= 159))
|
|
|
|
|
|
def matching_ports(ports, vid=None, pid=None, serial_number=None):
|
|
# Never guess a device from a generic CDC adapter or a partial identity.
|
|
if vid is None or pid is None or not serial_number:
|
|
return []
|
|
return [p.device for p in ports if p.vid == vid and p.pid == pid
|
|
and p.serial_number == serial_number]
|
|
|
|
|
|
class Capture:
|
|
def __init__(self, root):
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
self.root = root / datetime.now().strftime('%Y%m%d-%H%M%S-%f')
|
|
self.root.mkdir()
|
|
self.part = 0
|
|
self.stream = (self.root / 'device-000.log').open('xb')
|
|
self.decoder = codecs.getincrementaldecoder('utf-8')('replace')
|
|
self.lines = deque(maxlen=20000)
|
|
self.pending = ''
|
|
self.events = []
|
|
self.total = 0
|
|
self.screen_wire = ScreenWire()
|
|
self.screen_store = ScreenStore(self.root / 'screens')
|
|
self.screen_control = ScreenControl(self.screen_store)
|
|
self.viewer = None
|
|
self.last_text = b''
|
|
self.screen_notices = deque()
|
|
|
|
def feed(self, data):
|
|
incoming = len(data)
|
|
data, frames = self.screen_wire.feed(data)
|
|
self.last_text = data
|
|
for frame in frames:
|
|
path = self.screen_store.receive(frame)
|
|
if frame[1] or self.screen_store.info['frames'] == 1:
|
|
self.screen_notices.append('Screen PNG: ' + str(path.resolve()))
|
|
if self.screen_wire.last_event == 'stop':
|
|
self.screen_store.set_state('screen transfer stopped; last frame retained')
|
|
if self.stream.tell() + len(data) > 32 * 1024 * 1024:
|
|
self.stream.close()
|
|
self.part += 1
|
|
self.stream = (self.root / f'device-{self.part:03}.log').open('xb')
|
|
self.stream.write(data)
|
|
self.stream.flush()
|
|
self.total += incoming
|
|
text = self.decoder.decode(data)
|
|
chunks = (self.pending + text).split('\n')
|
|
self.lines.extend(chunks[:-1])
|
|
for record in chunks[:-1]:
|
|
self.screen_control.observe(record.rstrip('\r'))
|
|
self.pending = chunks[-1]
|
|
# Bound an unterminated line from a noisy device.
|
|
if len(self.pending) > 16384:
|
|
self.lines.append(self.pending[:16384])
|
|
self.pending = self.pending[-16384:]
|
|
return safe_text(text)
|
|
|
|
def filtered(self, topic):
|
|
if topic not in TOPICS:
|
|
raise ValueError('Topics: ' + ' '.join(TOPICS))
|
|
return [line for line in [*self.lines, self.pending] if re.search(TOPICS[topic], line, re.I)]
|
|
|
|
def save(self, port):
|
|
for topic in TOPICS:
|
|
(self.root / f'{topic}.txt').write_text('\n'.join(self.filtered(topic)), encoding='utf-8')
|
|
(self.root / 'session.json').write_text(json.dumps({
|
|
'port': port, 'bytes': self.total, 'commands': self.events,
|
|
'topic_source': 'last 20000 session log lines; historical, not live snapshots',
|
|
'raw_logs': self.part + 1,
|
|
'screen_binary_separated': True,
|
|
'screen_frames': self.screen_store.info['frames'],
|
|
'screen_decode_errors': self.screen_wire.errors,
|
|
}, indent=2), encoding='utf-8')
|
|
return self.root
|
|
|
|
def close(self):
|
|
self.screen_control.set_connection('disconnected')
|
|
self.stream.close()
|
|
if self.viewer:
|
|
self.viewer.close()
|
|
|
|
def preview(self):
|
|
if self.viewer is None:
|
|
self.viewer = ScreenViewer(self.screen_store, self.screen_control)
|
|
self.viewer.open()
|
|
return self.viewer.url
|
|
|
|
|
|
class Commands:
|
|
def __init__(self, capture, port):
|
|
self.capture, self.port = capture, port
|
|
self.queue = deque()
|
|
self.watch = None
|
|
self.next_watch = 0
|
|
self.next_send = 0
|
|
self.save_at = None
|
|
self.shell_deadline = None
|
|
self.reboot_sent = False
|
|
self.screen_started = False
|
|
self.next_control = 0
|
|
|
|
def accept(self, line):
|
|
words = line.strip().split()
|
|
if not words:
|
|
return ''
|
|
cmd = words[0]
|
|
if cmd in ('exit', 'quit', ':quit'):
|
|
return None
|
|
if cmd == ':help':
|
|
return HELP
|
|
if cmd == 'logs' and len(words) == 2:
|
|
return '[HISTORICAL SESSION LOGS]\n' + ('\n'.join(self.capture.filtered(words[1])) or 'No matching records captured.')
|
|
if cmd == 'save':
|
|
return 'Saved: ' + str(self.capture.save(self.port))
|
|
if cmd == 'unwatch':
|
|
self.watch = None
|
|
return 'Polling stopped.'
|
|
if cmd == 'watch' and len(words) == 3:
|
|
wire_command(words[1])
|
|
if words[1] not in ('ping', 'status', 'features', 'info', 'uptime', 'memory',
|
|
'ps', 'bootargs', 'display', 'usb', 'ufs', 'wifi.status',
|
|
'gpu.status', 'gpu.hfi.status', 'reboot.status', 'input.status'):
|
|
raise ValueError('Only read-only status queries support periodic polling.')
|
|
seconds = float(words[2])
|
|
if not 1 <= seconds <= 86400:
|
|
raise ValueError('Polling interval must be 1..86400 seconds.')
|
|
self.watch = (words[1], seconds)
|
|
self.next_watch = 0
|
|
return 'Polling enabled.'
|
|
if cmd in ('all', 'collect') and len(words) == 1:
|
|
if self.queue or self.save_at is not None:
|
|
raise ValueError('A collection is already pending.')
|
|
self.watch = None
|
|
self.queue.extend(('features','info','uptime','memory','ps','bootargs','display','usb','ufs','gpu.status','status'))
|
|
if cmd == 'collect':
|
|
self.queue.extend(('replay','sample'))
|
|
self.save_at = time.monotonic() + 15
|
|
return 'Queries queued; old kernels will report unsupported commands.'
|
|
if len(self.queue) >= 32:
|
|
raise ValueError('Command queue full; wait for pending queries.')
|
|
wire_command(line)
|
|
if self.reboot_sent:
|
|
raise ValueError('Reboot already sent; wait for Fastboot.')
|
|
if line == 'reboot.fastboot':
|
|
if self.queue or self.save_at is not None or self.shell_deadline is not None:
|
|
raise ValueError('Wait for pending commands before rebooting.')
|
|
self.watch = None
|
|
self.capture.screen_control.set_connection('reboot')
|
|
if line == 'shell':
|
|
if self.queue or self.save_at is not None or self.shell_deadline is not None:
|
|
raise ValueError('Wait for pending commands before entering shell.')
|
|
self.watch = None
|
|
self.capture.screen_control.set_connection('shell')
|
|
self.queue.append(line)
|
|
if line == 'screen.start':
|
|
return 'Live screen viewer: ' + self.capture.preview()
|
|
return ''
|
|
|
|
def tick(self, serial):
|
|
now = time.monotonic()
|
|
if self.watch and now >= self.next_watch and not self.queue:
|
|
self.queue.append(self.watch[0])
|
|
self.next_watch = now + self.watch[1]
|
|
if self.queue and now >= self.next_send:
|
|
command = self.queue.popleft()
|
|
data = wire_command(command)
|
|
if serial.write(data) != len(data):
|
|
raise OSError('Incomplete command write; reconnect before retrying.')
|
|
self.capture.events.append({'time': datetime.now().isoformat(), 'sent': command})
|
|
if command == 'shell':
|
|
self.shell_deadline = now + 5
|
|
self.screen_started = False
|
|
if command == 'screen.start':
|
|
self.screen_started = True
|
|
self.capture.screen_control.request_status()
|
|
if command == 'screen.stop':
|
|
self.screen_started = False
|
|
self.capture.screen_control.handle({'action': 'control', 'enabled': False})
|
|
if command == 'reboot.fastboot':
|
|
self.reboot_sent = True
|
|
self.next_send = now + 0.4
|
|
elif now >= self.next_control:
|
|
command = self.capture.screen_control.poll(now)
|
|
if command:
|
|
data = wire_command(command)
|
|
if serial.write(data) != len(data):
|
|
raise OSError('Incomplete screen control write; reconnect before retrying.')
|
|
# Movement is transient; retaining every pointer/heartbeat would
|
|
# grow session metadata for the entire duration of mirroring.
|
|
if not command.startswith('input.'):
|
|
self.capture.events.append({'time': datetime.now().isoformat(), 'sent': command})
|
|
if command == 'screen.start':
|
|
self.screen_started = True
|
|
elif command == 'screen.stop':
|
|
self.screen_started = False
|
|
self.next_control = now + 0.01
|
|
if self.save_at is not None and now >= self.save_at and not self.queue:
|
|
self.save_at = None
|
|
return 'Saved: ' + str(self.capture.save(self.port))
|
|
return ''
|
|
|
|
|
|
def reboot_acknowledged(capture):
|
|
return any('GIULIA REBOOT: resetting target=fastboot disks=shutdown-complete' in line
|
|
for line in [*capture.lines, capture.pending])
|
|
|
|
|
|
def reboot_failed(capture):
|
|
return any('GIULIA REBOOT: PSCI SYSTEM_RESET returned=' in line or
|
|
'GIULIA REBOOT: shutdown returned=' in line
|
|
for line in [*capture.lines, capture.pending])
|
|
|
|
|
|
def wait_fastboot(serial_number, timeout, runner=subprocess.run,
|
|
now=time.monotonic, sleep=time.sleep):
|
|
deadline = now() + timeout
|
|
while now() < deadline:
|
|
try:
|
|
result = runner(['fastboot', 'devices'], capture_output=True, text=True,
|
|
timeout=min(3, max(.1, deadline-now())), check=False)
|
|
if result.returncode == 0 and any(
|
|
line.split() == [serial_number, 'fastboot'] for line in result.stdout.splitlines()):
|
|
print('Fastboot verified for the requested device.', flush=True)
|
|
return 0
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
sleep(min(.5, max(0, deadline-now())))
|
|
raise TimeoutError('Requested device did not appear in Fastboot before the timeout.')
|
|
|
|
|
|
@contextmanager
|
|
def stop_screen_session(serial, active, control=None):
|
|
try:
|
|
yield
|
|
finally:
|
|
if control:
|
|
# This runs before closing the sole serial connection. The kernel
|
|
# also releases input on DTR loss and its independent lease timeout.
|
|
try:
|
|
if control.available and control.connection == 'diagnostic':
|
|
serial.write(wire_command('input.release'))
|
|
except OSError:
|
|
pass
|
|
control.set_connection('disconnected')
|
|
if active():
|
|
try:
|
|
serial.write(wire_command('screen.stop'))
|
|
except OSError:
|
|
pass # Disconnect also stops the kernel producer.
|
|
|
|
|
|
def run(args, serial_module, ports):
|
|
if args.list:
|
|
for p in ports:
|
|
print(f'{p.device}: {p.description} [{p.hwid}]')
|
|
return 0
|
|
found = matching_ports(ports, getattr(args, 'vid', None),
|
|
getattr(args, 'pid', None), getattr(args, 'usb_serial', None))
|
|
port = args.port or (found[0] if len(found) == 1 else None)
|
|
if not port:
|
|
raise ValueError('No unique CDC port selected. Use --list and --port, or specify --vid, --pid and --usb-serial.')
|
|
if not args.command and (os.name != 'nt' or not sys.stdin.isatty()):
|
|
raise ValueError('Interactive mode requires Windows CMD; use --command for scripted capture.')
|
|
capture = Capture(Path(args.output).resolve())
|
|
console = None
|
|
reboot_disconnected = False
|
|
commands = None
|
|
try:
|
|
# pySerial defaults to DTR/RTS asserted; CDC transmits only while DTR is set.
|
|
with serial_module.Serial(port, 115200, timeout=0, write_timeout=2) as serial, \
|
|
stop_screen_session(serial, lambda: commands is not None and commands.screen_started, capture.screen_control):
|
|
commands = Commands(capture, port)
|
|
print(f'Connected {port}. Logs: {capture.root}\n' + HELP)
|
|
line = ''
|
|
history = []
|
|
history_index = 0
|
|
def show(text):
|
|
if not text:
|
|
return
|
|
text = safe_text(text)
|
|
if not args.command:
|
|
print('\r' + ' ' * (len(line) + 8) + '\r', end='')
|
|
print(text, end='' if text.endswith('\n') else '\n', flush=True)
|
|
if not args.command:
|
|
print('unixnu> ' + line, end='', flush=True)
|
|
if args.command:
|
|
if getattr(args, 'max_size', None) is not None or getattr(args, 'max_fps', None) is not None:
|
|
show(commands.accept(f'screen.configure {args.max_size or 695} {args.max_fps or 1}'))
|
|
show(commands.accept(args.command))
|
|
else:
|
|
import msvcrt
|
|
print('unixnu> ', end='', flush=True)
|
|
end = time.monotonic() + args.seconds
|
|
extended = False
|
|
wire = ShellWire()
|
|
log_decoder = codecs.getincrementaldecoder('utf-8')('replace')
|
|
if not args.command:
|
|
console = WindowsConsole()
|
|
while not args.command or time.monotonic() < end:
|
|
try:
|
|
data = serial.read(min(serial.in_waiting, 16384))
|
|
except (OSError, serial_module.SerialException):
|
|
if not commands.reboot_sent or not reboot_acknowledged(capture) or reboot_failed(capture):
|
|
raise
|
|
reboot_disconnected = True
|
|
show('USB disconnected after the kernel completed shutdown for Fastboot.')
|
|
break
|
|
if data:
|
|
capture.feed(data)
|
|
if commands.reboot_sent and reboot_failed(capture):
|
|
raise OSError('Kernel reset returned without rebooting; inspect the saved log.')
|
|
while capture.screen_notices:
|
|
show(capture.screen_notices.popleft())
|
|
for kind, chunk in wire.feed(capture.last_text):
|
|
if kind == 'log':
|
|
show(safe_text(log_decoder.decode(chunk)))
|
|
elif kind == 'begin':
|
|
commands.shell_deadline = None
|
|
capture.screen_control.set_connection('shell')
|
|
if console:
|
|
console.raw(True)
|
|
print('\r\n[Phone PTY: Ctrl+] returns to debugger]\r', flush=True)
|
|
elif kind == 'raw':
|
|
sys.stdout.buffer.write(chunk)
|
|
sys.stdout.buffer.flush()
|
|
else:
|
|
capture.screen_control.set_connection('diagnostic')
|
|
capture.screen_control.request_status()
|
|
if console:
|
|
console.raw(False)
|
|
print('\r\n[Shell ended]', flush=True)
|
|
if not args.command:
|
|
print('unixnu> ', end='', flush=True)
|
|
if not wire.active:
|
|
show(commands.tick(serial))
|
|
if commands.shell_deadline and time.monotonic() >= commands.shell_deadline:
|
|
commands.shell_deadline = None
|
|
capture.screen_control.set_connection('diagnostic')
|
|
show('Shell did not start. Check giulia_debug=1 and the installed daemon/shell.')
|
|
if not args.command and msvcrt.kbhit():
|
|
key = msvcrt.getwch()
|
|
if wire.active:
|
|
if key in ('\x00', '\xe0'):
|
|
extended = True
|
|
continue
|
|
if extended:
|
|
extended = False
|
|
encoded = {'H':b'\x1b[A','P':b'\x1b[B','M':b'\x1b[C','K':b'\x1b[D',
|
|
'G':b'\x1b[H','O':b'\x1b[F','S':b'\x1b[3~',
|
|
'I':b'\x1b[5~','Q':b'\x1b[6~'}.get(key,b'')
|
|
else:
|
|
encoded = key.encode('utf-8', errors='replace')
|
|
if encoded and serial.write(encoded) != len(encoded):
|
|
raise OSError('Incomplete shell input write.')
|
|
continue
|
|
if commands.shell_deadline:
|
|
# Do not accidentally interpret shell input as a kernel command.
|
|
if key == '\x03':
|
|
serial.write(b'\x1d')
|
|
break
|
|
continue
|
|
if key in ('\x00', '\xe0'):
|
|
extended = True
|
|
continue
|
|
if extended:
|
|
extended = False
|
|
if key in ('H', 'P'):
|
|
history_index = max(0, min(len(history), history_index + (-1 if key == 'H' else 1)))
|
|
previous = len(line)
|
|
line = history[history_index] if history_index < len(history) else ''
|
|
print('\r' + ' ' * (previous + 6) + '\runixnu> ' + line, end='', flush=True)
|
|
continue
|
|
if key == '\x03':
|
|
break
|
|
if key == '\r':
|
|
print()
|
|
command, line = line.strip(), ''
|
|
if command:
|
|
history.append(command)
|
|
history = history[-100:]
|
|
history_index = len(history)
|
|
if command == 'clear':
|
|
os.system('cls')
|
|
show('Screen cleared; logs retained.')
|
|
else:
|
|
try:
|
|
reply = commands.accept(command)
|
|
if reply is None:
|
|
break
|
|
show((reply or 'Queued.') if command else '')
|
|
except ValueError as error:
|
|
show(str(error))
|
|
if not command:
|
|
print('unixnu> ', end='', flush=True)
|
|
elif key == '\b' and line:
|
|
line = line[:-1]
|
|
print('\b \b', end='', flush=True)
|
|
elif 32 <= ord(key) <= 126 and len(line) < 120:
|
|
line += key
|
|
print(key, end='', flush=True)
|
|
time.sleep(0.01)
|
|
if args.command == 'reboot.fastboot' and (
|
|
not reboot_acknowledged(capture) or reboot_failed(capture) or not reboot_disconnected):
|
|
raise TimeoutError('Kernel did not complete the Fastboot reset and disconnect; inspect the saved log.')
|
|
finally:
|
|
if console:
|
|
console.restore()
|
|
try:
|
|
capture.save(port)
|
|
finally:
|
|
capture.close()
|
|
print(f'\nSaved session: {capture.root}')
|
|
return 0
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--port', help='Explicit serial port, e.g. COM21 or /dev/ttyACM0')
|
|
parser.add_argument('--vid', type=lambda value: int(value, 0), help='USB vendor ID, e.g. 0x1209')
|
|
parser.add_argument('--pid', type=lambda value: int(value, 0), help='USB product ID, e.g. 0x0001')
|
|
parser.add_argument('--usb-serial', help='Exact USB serial number for automatic selection')
|
|
parser.add_argument('--list', action='store_true')
|
|
parser.add_argument('--command', help='One command, or collect/all')
|
|
parser.add_argument('--seconds', type=float, default=20)
|
|
parser.add_argument('--fastboot-serial', help='For reboot.fastboot: wait for this exact Fastboot device before success')
|
|
parser.add_argument('--output', default=str(Path(__file__).resolve().parent / '../out/terminal'))
|
|
parser.add_argument('-m', '--max-size', type=int, help='Maximum mirrored edge, 160..2780; keeps panel aspect ratio')
|
|
parser.add_argument('--max-fps', type=int, help='Target capture rate, 1..30; actual rate depends on USB bandwidth')
|
|
args = parser.parse_args()
|
|
identity = (args.vid, args.pid, args.usb_serial)
|
|
if any(value is not None for value in identity):
|
|
if any(value is None for value in identity) or not args.usb_serial:
|
|
parser.error('--vid, --pid and --usb-serial must be supplied together')
|
|
if not 0 <= args.vid <= 65535 or not 0 <= args.pid <= 65535:
|
|
parser.error('USB vendor/product IDs must be 0..65535')
|
|
if not 1 <= args.seconds <= 86400:
|
|
parser.error('--seconds must be 1..86400')
|
|
if args.max_size is not None and not 160 <= args.max_size <= 2780:
|
|
parser.error('--max-size must be 160..2780')
|
|
if args.max_fps is not None and not 1 <= args.max_fps <= 30:
|
|
parser.error('--max-fps must be 1..30')
|
|
if (args.max_size is not None or args.max_fps is not None) and args.command not in ('screen.start','screen.snapshot'):
|
|
parser.error('--max-size/--max-fps require --command screen.start or screen.snapshot')
|
|
if args.command == 'shell':
|
|
parser.error('Use interactive mode for shell; --command cannot control a PTY.')
|
|
if args.fastboot_serial:
|
|
if args.command != 'reboot.fastboot':
|
|
parser.error('--fastboot-serial requires --command reboot.fastboot')
|
|
if not shutil.which('fastboot'):
|
|
parser.error('fastboot must be on PATH to verify the reboot')
|
|
try:
|
|
import serial
|
|
from serial.tools import list_ports
|
|
except ImportError:
|
|
print('Install dependency in CMD: python -m pip install pyserial==3.5', file=sys.stderr)
|
|
return 2
|
|
try:
|
|
result = run(args, serial, list(list_ports.comports()))
|
|
if not result and args.fastboot_serial:
|
|
return wait_fastboot(args.fastboot_serial, args.seconds)
|
|
return result
|
|
except KeyboardInterrupt:
|
|
return 130
|
|
except (OSError, ValueError, serial.SerialException) as error:
|
|
print(f'Error: {error}', file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|