119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
"""Build a captured ``/dev/iec0`` record stream for the canonical PoC session.
|
|
|
|
This reproduces the IEC traffic of the PLAN.md §1 BASIC program::
|
|
|
|
OPEN 1,4
|
|
PRINT#1,"HELLO WORLD"
|
|
PRINT#1,"LINE TWO"
|
|
CLOSE 1
|
|
|
|
as a list of :class:`~iecpoc.device.IecRecord`, so the whole userspace decode +
|
|
trace path can be exercised on a host with no Pi/kernel. Run this module as a
|
|
script to (re)generate ``tests/data/hello_world_session.bin``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from iecpoc import device
|
|
from iecpoc.device import (
|
|
FLAG_ADDRESSED,
|
|
FLAG_EOI,
|
|
KIND_COMMAND,
|
|
KIND_DATA,
|
|
KIND_EVENT,
|
|
IecRecord,
|
|
)
|
|
|
|
MY_ADDRESS = 4
|
|
_PRINTER_CR = 0x0D # PRINT# terminates each record with a carriage return
|
|
|
|
|
|
class _Builder:
|
|
"""Accumulate records with monotonically increasing timestamps."""
|
|
|
|
def __init__(self) -> None:
|
|
self.records: list[IecRecord] = []
|
|
self._ts = 0
|
|
|
|
def _tick(self, step: int = 1_000) -> int:
|
|
self._ts += step
|
|
return self._ts
|
|
|
|
def event(self, code: int, *, addressed: bool = False) -> None:
|
|
flags = FLAG_ADDRESSED if addressed else 0
|
|
self.records.append(IecRecord(KIND_EVENT, code, flags, self._tick()))
|
|
|
|
def command(self, value: int, *, addressed: bool = False) -> None:
|
|
flags = FLAG_ADDRESSED if addressed else 0
|
|
self.records.append(IecRecord(KIND_COMMAND, value, flags, self._tick()))
|
|
|
|
def data(self, value: int, *, eoi: bool = False) -> None:
|
|
flags = FLAG_EOI if eoi else 0
|
|
# data is always addressed-to-us by definition of reaching LISTENER state
|
|
self.records.append(IecRecord(KIND_DATA, value, flags | FLAG_ADDRESSED, self._tick(20)))
|
|
|
|
def data_text(self, text: str) -> None:
|
|
"""Send the bytes of an ASCII/PETSCII-compatible string then a CR (EOI)."""
|
|
for ch in text:
|
|
self.data(ord(ch))
|
|
self.data(_PRINTER_CR, eoi=True)
|
|
|
|
|
|
def build_session() -> list[IecRecord]:
|
|
"""Return the record list for the canonical four-statement BASIC program."""
|
|
b = _Builder()
|
|
|
|
# --- OPEN 1,4 : LISTEN 4, OPEN ch0 ; no filename so no data phase ---------
|
|
b.event(device.EV_IDLE)
|
|
b.event(device.EV_ATN_ASSERTED)
|
|
b.command(0x24, addressed=True) # LISTEN 4
|
|
b.command(0xF0) # OPEN SA=0
|
|
b.event(device.EV_ATN_RELEASED, addressed=True)
|
|
b.event(device.EV_ATN_COMMAND)
|
|
b.command(0x3F) # UNLISTEN
|
|
b.event(device.EV_ATN_RELEASED, addressed=False)
|
|
|
|
# --- PRINT#1,"HELLO WORLD" -----------------------------------------------
|
|
b.event(device.EV_ATN_ASSERTED)
|
|
b.command(0x24, addressed=True) # LISTEN 4
|
|
b.command(0x60) # DATA (reopen) SA=0
|
|
b.event(device.EV_ATN_RELEASED, addressed=True)
|
|
b.data_text("HELLO WORLD")
|
|
b.event(device.EV_ATN_COMMAND)
|
|
b.command(0x3F) # UNLISTEN
|
|
b.event(device.EV_ATN_RELEASED, addressed=False)
|
|
|
|
# --- PRINT#1,"LINE TWO" --------------------------------------------------
|
|
b.event(device.EV_ATN_ASSERTED)
|
|
b.command(0x24, addressed=True) # LISTEN 4
|
|
b.command(0x60) # DATA (reopen) SA=0
|
|
b.event(device.EV_ATN_RELEASED, addressed=True)
|
|
b.data_text("LINE TWO")
|
|
b.event(device.EV_ATN_COMMAND)
|
|
b.command(0x3F) # UNLISTEN
|
|
b.event(device.EV_ATN_RELEASED, addressed=False)
|
|
|
|
# --- CLOSE 1 : LISTEN 4, CLOSE ch0 ---------------------------------------
|
|
b.event(device.EV_ATN_ASSERTED)
|
|
b.command(0x24, addressed=True) # LISTEN 4
|
|
b.command(0xE0) # CLOSE SA=0
|
|
b.event(device.EV_ATN_RELEASED, addressed=False)
|
|
|
|
return b.records
|
|
|
|
|
|
CAPTURE_PATH = os.path.join(os.path.dirname(__file__), "data", "hello_world_session.bin")
|
|
|
|
|
|
def write_capture(path: str = CAPTURE_PATH) -> str:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "wb") as fh:
|
|
device.write_records(fh, build_session())
|
|
return path
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("wrote", write_capture())
|