first try

This commit is contained in:
Christian Werner 2026-06-18 22:25:42 +02:00
parent e5e51754b2
commit e4cab679b5
27 changed files with 3185 additions and 48 deletions

21
.gitignore vendored
View File

@ -1 +1,20 @@
.idea
.idea
# Python
.venv/
__pycache__/
*.py[cod]
*.egg-info/
.pytest_cache/
# Kernel module build artifacts
kernel/*.o
kernel/*.ko
kernel/*.mod
kernel/*.mod.c
kernel/*.mod.o
kernel/*.symvers
kernel/*.order
kernel/.*.cmd
kernel/.tmp_versions/
kernel/dts/*.dtbo

View File

@ -0,0 +1,81 @@
# comodore-iec-emu — IEC listener PoC
A proof-of-concept Commodore **IEC listener** device on a Raspberry Pi Zero 2 W.
It behaves like a printer (default device address **4**): it never talks back, it
only **listens**, and it debug-prints in real time everything the C64 sends.
The timing-critical IEC handshake (ATN ack, per-bit sampling, EOI) runs in a
**Linux kernel module**; a **Python** userspace program decodes and pretty-prints.
See [`_plans/poc-listener-printer-PLAN.md`](_plans/poc-listener-printer-PLAN.md)
for the full design and [`_research/`](_research/) for the grounding research.
## Layout
```
kernel/ # the kernel module (real-time half) — builds on the Pi
iec_listener.c ISR, IRQ-off bit loop, listener state machine, /dev/iec0
iec_listener.h shared struct iec_record + ioctls (mirrors device.py)
iec_lines.h logical line layer, direct BCM register access (hot path)
iec_timing.h handshake timing constants
Makefile out-of-tree build + load/unload/overlay targets
dts/ device-tree overlay (pin reservation / pulls)
iecpoc/ # Python userspace (everything else) — host-testable
device.py /dev/iec0 record stream (wire format, replay)
decode.py command-byte + event decoder
petscii.py PETSCII -> display glyphs
log.py the annotated trace formatter (PLAN §8)
main.py CLI (--address, --raw, --logfile, --replay)
tests/ # host-runnable, no Pi/kernel needed
docs/ # wiring.md, kernel-notes.md
```
## Userspace (Phase 0 — works on any host)
```bash
python3 -m venv .venv && .venv/bin/pip install -e '.[dev]'
.venv/bin/pytest # run the tests
.venv/bin/python -m tests.fixtures # (re)generate the capture
.venv/bin/python -m iecpoc.main --replay tests/data/hello_world_session.bin
```
On the Pi, point it at the live device instead:
```bash
.venv/bin/python -m iecpoc.main --address 4 # reads /dev/iec0
```
Example trace (the `OPEN 1,4 / PRINT# / CLOSE` session, abbreviated):
```
[IDLE] waiting for ATN
[ATN] asserted -> DATA low (ack)
[CMD] $24 LISTEN 4 (addressed: ME)
[CMD] $60 DATA SA=0
[ATN] released -> LISTENER
[DATA] $48 'H'
...
[DATA] $0D <CR> <EOI>
[ATN] released -> not addressed -> IDLE
```
## Kernel module (on the Pi)
```bash
sudo apt install raspberrypi-kernel-headers
cd kernel && make
sudo insmod iec_listener.ko address=4 # creates /dev/iec0
sudo rmmod iec_listener # releases DATA on unload
```
See [`docs/wiring.md`](docs/wiring.md) for the level-shifter circuit and the
pre-C64 bring-up checklist, and [`docs/kernel-notes.md`](docs/kernel-notes.md)
for the resolved kernel real-time decisions.
## Status
- **Phase 0 (userspace + tests):** complete, runs on any host.
- **Phase 1 (kernel command phase):** skeleton present; needs Pi bring-up.
- **Phase 2 (data-phase reception):** the real experiment; tune timing on real
hardware.
See PLAN.md §9 for the phased milestones.

View File

@ -62,14 +62,11 @@ From the research, a pure listener has a radically reduced surface:
|-------|-------------------------|--------------|
| ATN | C64 drives; we react | **input** |
| CLK | Talker (C64) drives during data; we sample | **input** |
| DATA | Listener drives the handshake (ack ATN, ready, byte-ack, EOI-ack); talker drives the bits we sample | **bidirectional** (open-drain) |
| DATA | Listener drives (ack ATN, ready, byte-ack, EOI-ack) | **output** (open-collector) |
| RESET | C64 drives; we react | **input** |
| SRQ | unused on stock C64 | not connected |
So we **actively drive exactly one** bus line (DATA, open-drain) and sense the
rest — but DATA doubles as an *input* during bit sampling (we release it to Hi-Z,
then read the talker's bits), which is why it is a single bidirectional pin (§3).
No turnaround, no
So we drive **exactly one** bus line (DATA) and read three. No turnaround, no
EOI generation, no 60 µs talker timing. The only timing the device generates is
the 1 ms ATN ack and the per-byte ack — both forgiving. The tight constraint is
*sampling* the C64's ~20 µs bit clock while receiving (see §6).
@ -78,50 +75,34 @@ the 1 ms ATN ack and the per-byte ack — both forgiving. The tight constraint i
## 3. Hardware
### 3.1 Interface circuit (single bidirectional DATA pin, sd2iec-style)
### 3.1 Interface circuit (per the research, simplified for one output)
Design decision: DATA is a **single bidirectional pin** doing **open-drain
emulation**, mirroring the sd2iec one-line scheme rather than a separate
drive-pin + sense-pin pair. Because the Pi is **not 5 V-tolerant**, the line goes
through a **bidirectional MOSFET level shifter** (BSS138 + pull-ups — the NXP
AN10441 / common "logic level converter" topology). sd2iec needs no shifter only
because its AVR runs at 5 V; the 3.3 V Pi does.
- **DATA (single pin, bidirectional):** GPIO 18 ↔ level shifter ↔ IEC DATA. The Pi
pin is never a push-pull driver of the bus; it emulates open-drain:
- **Assert** (pull bus low): GPIO 18 = **output LOW** → shifter pulls bus to 0 V.
- **Release / read**: GPIO 18 = **input (Hi-Z)** → the bus pull-ups define the
level and GPIO 18 reads the bus. **Non-inverting:** bus low (asserted) ⇒ Pi
reads LOW. (Releasing to Hi-Z is what lets us sample the talker's bits in §6.)
- Direction is flipped in the hot path via the BCM `GPFSEL` register — a single
register write, ~tens of ns, negligible against the 20 µs bit window.
- **ATN, CLK, RESET (sense, input only):** route through the **same kind of
shifter** (remaining channels of a 4-channel BSS138 board). Pi side always input.
**Non-inverting:** bus low (asserted) ⇒ Pi reads LOW. *(A resistor divider also
works for these input-only lines, but reusing the shifter gives a uniform, fully
in-spec 3.3 V swing and sidesteps the marginal divider voltages flagged in the
kernel research.)*
- **Recommended part:** one 4-channel bidirectional level-shifter board carries all
four lines (DATA bidirectional; ATN/CLK/RESET input-only). No 7406, no transistor,
no hand-matched dividers — low side to Pi 3.3 V, high side to bus 5 V.
- 100 nF bypass cap on the shifter VCC pins. Common GND between C64 IEC pin 2 and Pi.
- **DATA (drive):** open-collector driver. Two equivalent options:
- **7406/74LS06** inverting OC buffer: Pi GPIO → 7406 input → bus. Pi HIGH ⇒ bus
pulled low (asserted). *(Software must invert.)*
- **NPN transistor** (2N3904/BC547): Pi GPIO → 1 kΩ → base; emitter → GND;
collector → DATA line. Pi HIGH ⇒ transistor on ⇒ bus low. Same inversion.
- Use whichever is on hand; the 7406 also gives spare drivers for later (CLK,
talker mode). Recommend the **7406** to match the eventual full device.
- **ATN, CLK, RESET (sense):** resistor divider 5 V→~3.1 V into each Pi input
(e.g. 3.3 kΩ top / 2.2 kΩ to GND). **No inversion** in sensing: bus low (0 V,
asserted) ⇒ Pi reads LOW.
- 100 nF bypass cap across the 7406 VCCGND. Common GND between C64 IEC pin 2 and Pi.
- **Do not connect bus 5 V to any Pi pin directly — Pi GPIO is not 5 V tolerant.**
### 3.2 Pin map (Pi Zero 2 W, 40-pin header)
| IEC signal | Pi GPIO (BCM) | Header pin | Direction | Logic note |
|------------|---------------|------------|-----------|------------|
| ATN | GPIO 2 | 3 | input (via shifter) | bus low = Pi low = asserted |
| CLK | GPIO 17 | 11 | input (via shifter) | bus low = Pi low = asserted |
| DATA | GPIO 18 | 12 | **bidirectional (via shifter)** | open-drain: assert = output LOW, release/read = input. bus low = Pi low = asserted |
| RESET | GPIO 3 | 5 | input (via shifter) | bus low = Pi low = asserted |
| GND | — | 6 (or any) | — | tie to IEC pin 2 and shifter GND |
| ATN | GPIO 2 | 3 | input (divider) | bus low = Pi low = asserted |
| CLK | GPIO 17 | 11 | input (divider) | bus low = Pi low = asserted |
| DATA (in) | GPIO 18 | 12 | output → 7406 → bus | Pi HIGH = bus asserted (inverted) |
| RESET | GPIO 3 | 5 | input (divider) | bus low = Pi low = asserted |
| GND | — | 6 (or any) | — | tie to IEC pin 2 |
> Convention used in code: a helper layer (`iec_lines.h`) converts electrical
> reads/writes into **logical** `asserted=True / released=False`. With the
> non-inverting level shifter the mapping is direct — assert ⇒ DATA GPIO output
> LOW, release ⇒ DATA GPIO input (Hi-Z), and a Pi LOW read on any line means
> asserted. No 7406 inversion to juggle.
> Convention used in code: a helper layer converts electrical reads/writes into
> **logical** `asserted=True / released=False` so the rest of the code reasons in
> the protocol's true/false sense and never juggles the inversions inline.
---
@ -224,11 +205,9 @@ never juggles inversions inline (mirrors the original userspace plan, now in C):
- `atn_asserted()` — true when ATN GPIO reads low
- `clk_asserted()` — true when CLK GPIO reads low
- `data_in()` — read DATA line (the talker drives it during bit transfer; only
valid after `data_release()` has set the pin to Hi-Z input)
- `data_assert()` / `data_release()` — assert DATA (set GPIO 18 **output LOW**) /
release it (set GPIO 18 **input / Hi-Z**, letting the bus float high). Open-drain
emulation through the level shifter; **non-inverting** (no 7406 to invert)
- `data_in()` — read DATA line (the talker drives it during bit transfer)
- `data_assert()` / `data_release()` — drive DATA true / float it; encapsulates
the 7406 inversion (assert ⇒ Pi GPIO HIGH)
- `reset_asserted()` — true when RESET GPIO reads low
### 5.2 Userspace record stream (`device.py`)
@ -437,7 +416,7 @@ Requirements:
| Even in-kernel, busy-poll mid-byte gets preempted/IRQ'd and garbles a bit | Medium | Disable local IRQs/preemption for the duration of a byte; if still marginal, isolcpus + steer other IRQs away (raspbiec pattern); last resort `PREEMPT_RT`. Quantify in Phase 0.5. |
| Keeping IRQs off too long for a byte harms system stability | Medium | Bound the off-window to one byte (~few hundred µs); re-enable between bytes. A design constraint to validate in Phase 0.5. |
| Kernel module build/ABI churn against Raspberry Pi OS kernel | Medium | Build against installed kernel headers; pin kernel version; document in `kernel-notes.md`. |
| Level-shifter wiring / DATA direction-flip bug | Medium | `iec_lines.h` logical layer isolates assert/release/read; module `selftest` ioctl drives DATA (output LOW), releases to Hi-Z and reads it back, and reads ATN/CLK with the C64 off. Confirm Hi-Z release truly floats so the talker's bits read through. |
| 7406 inversion / divider miswire | Medium | `iec_lines.h` logical layer isolates it; module `selftest` ioctl toggles DATA and reads ATN/CLK with the C64 off. |
| C64 has JiffyDOS | Low (printer rarely uses it) | Out of scope; document. Standard protocol still works for a printer device. |
| ATN response missed | Low | Hardware IRQ pulls DATA in the ISR; 1 ms budget is ample for kernel interrupt latency. |

File diff suppressed because it is too large Load Diff

125
docs/kernel-notes.md Normal file
View File

@ -0,0 +1,125 @@
# Kernel module notes
Phase-0.5 decisions, resolved in
`_research/pi-kernel-module-rt-gpio-2026-06-18.md` (read that for the full
evidence and source-level analysis). This file is the short, actionable summary
the code in `kernel/` is built on.
## Decisions baked into the code
| Question | Decision | Where |
|----------|----------|-------|
| CLK: interrupt vs. busy-poll | ATN falling-edge hardirq enters the state machine; bytes are busy-polled with `local_irq_save()` held for one byte (ninepin pattern) | `atn_isr`, `receive_byte` |
| IRQ discipline | `local_irq_save()` around each byte only (~200 µs normal, ≤ 8 ms abort cap), **not** the whole ATN phase | `receive_byte` |
| GPIO access (hot path) | Direct BCM register access via `ioremap` (~3040× faster than gpiod) | `iec_lines.h` |
| GPIO access (init/exit) | gpiod descriptor API (`gpio_to_desc`, `gpiod_to_irq`) | `iec_init`/`iec_exit` |
| Kernel↔userspace | Character device `/dev/iec0` + `kfifo` + wait queue (IEC ≤ 1000 B/s; relayfs not justified) | `iec_read`, `emit_record` |
| `udelay` vs. poll | Poll-with-timeout for CLK transitions; `udelay` only for fixed delays (EOI ack 80 µs, EOI detect 250 µs) | `iec_timing.h`, `wait_clk` |
| isolcpus / nohz_full | **Not** in the Phase-1 baseline. Add `isolcpus=3 nohz_full=3 rcu_nocbs=3 irqaffinity=0-2` only if Phase-2 bit-error rate > 1% | (boot cmdline) |
| PREEMPT_RT | Stock kernel sufficient; RT is a last resort | — |
| Module signing | Not required on stock RPi OS Bookworm | — |
| Peripheral base | `0x3F000000` (BCM2710A1 / Pi Zero 2 W; **not** ninepin's `0x20000000`) | `iec_lines.h` |
## Level-shifter / DATA-sensing note (deviation from the research doc)
The research doc (§4.3) analysed a **7406 inverting** drive + separate resistor
divider sense, and flagged a "DATA sensing gap" (a 7406 output pin can't read the
bus back). The **PLAN** instead specifies a **non-inverting bidirectional level
shifter** (BSS138, sd2iec-style single DATA pin, §3.1). That choice:
- makes the logic **non-inverting**: bus low (asserted) ⇒ Pi reads LOW;
- **closes the sensing gap** — when the DATA pin is switched to *input* (Hi-Z),
reading it back through the bidirectional shifter returns the real bus state
the C64 is driving. This is exactly what `receive_byte` relies on for bit
sampling.
So `iec_lines.h` is non-inverting; there is no 7406 inversion to track. Confirm
with the `IEC_IOC_SELFTEST` ioctl (drive low → read low; release → read high)
before connecting the C64 (PLAN.md §10 risk row).
## Build & deploy (on the Pi)
```bash
sudo apt install raspberrypi-kernel-headers
cd kernel
make # iec_listener.ko
make overlay # dts/iec-overlay.dtbo (optional pin reservation)
sudo insmod iec_listener.ko address=4
ls -l /dev/iec0
# ... talk to the C64 ...
sudo rmmod iec_listener # releases DATA on the way out
```
Optional pin-reservation overlay (Bookworm paths — note the `/boot/firmware/`
prefix; the legacy `/boot/` paths no longer apply on 64-bit Bookworm):
```bash
sudo cp dts/iec-overlay.dtbo /boot/firmware/overlays/
echo "dtoverlay=iec-overlay" | sudo tee -a /boot/firmware/config.txt
sudo reboot
# after reboot, verify it loaded:
dtoverlay -l
```
**Pin the kernel *before* the first `apt full-upgrade`** — any kernel bump
breaks the module via a vermagic mismatch (`Invalid module format`), so hold the
kernel packages up front rather than after the fact:
```bash
sudo apt-mark hold raspberrypi-kernel raspberrypi-kernel-headers raspberrypi-bootloader
# record the pinned version here once known:
# Pinned: raspberrypi-kernel <VERSION> (<DATE>)
```
## Building off the Pi (emulated arm64 Docker)
You can compile the module on a non-Pi (x86) host with `kernel/build-in-docker.sh`
(or `make docker-build`). It runs an **emulated arm64** Raspberry Pi OS container,
installs the raspberrypi kernel headers via apt, and builds natively so the
module's *vermagic* matches the Pi.
```bash
cd kernel
./build-in-docker.sh # -> iec_listener.ko (arm64) in this dir
./build-in-docker.sh clean
```
Configurable via env vars (kernel version is configurable as requested):
| Var | Default | Purpose |
|-----|---------|---------|
| `HEADERS_PKG` | `linux-headers-rpi-v8` | headers package; use `-v7`/`-v6` for 32-bit, or `raspberrypi-kernel-headers` |
| `KERNEL_VERSION` | *(latest)* | exact version pin, e.g. `1:6.6.51-1+rpt3` |
| `IMAGE` | `iec-kbuild` | builder image tag |
```bash
KERNEL_VERSION=1:6.6.51-1+rpt3 ./build-in-docker.sh
```
**vermagic caveat:** the raspberrypi apt archive normally serves only the
*latest* kernel in its pool, so pinning `KERNEL_VERSION` to an old release may
not be downloadable. The reliable strategy is to keep the Pi current
(`sudo apt full-upgrade`) and build with the default (latest) — then the Pi and
the container agree. If you must target an older/specific kernel, copy the Pi's
`/lib/modules/$(uname -r)/build` tree into the container instead of using apt.
`uname -r` is **not** used inside the container (it reports the host kernel under
emulation); the entrypoint derives `KDIR` from the installed headers under
`/lib/modules/`.
The host needs qemu binfmt for arm64; the script registers it once via
`tonistiigi/binfmt --install arm64` (a one-time privileged container).
## Starting timing constants
See `kernel/iec_timing.h`. Tune in Phase 2 against a real C64 / logic analyser.
The likely first knobs: `IEC_EOI_DETECT_US` (EOI false positives/negatives) and
`IEC_CLK_TIMEOUT_US` (frame errors under load).
## Open items to verify on hardware
- GPIO IRQ latency on BCM2710A1 under representative load (target ATN ack ≪ 1 ms).
- Direct-register read latency inside the IRQ-off loop on the A53 (budget vs. the
20 µs C64 bit window).
- Whether `isolcpus` is needed once Phase 2 runs under WiFi/SD load.
- `IEC_GPIO_DATA` direction-flip latency (`GPFSEL` write) — should be tens of ns.

55
docs/wiring.md Normal file
View File

@ -0,0 +1,55 @@
# Wiring — Pi Zero 2 W ↔ Commodore IEC bus
PoC listener (printer) interface. Full rationale: PLAN.md §3 and
`_research/commodore-iec-serial-bus-2026-06-18.md` §1.
> ⚠️ The Pi GPIO is **3.3 V and NOT 5 V tolerant**. The IEC bus idles at 5 V.
> A level shifter is **mandatory** — never wire a bus line straight to a Pi pin.
## Topology
A single **4-channel bidirectional MOSFET level shifter** (BSS138 + pull-ups,
the "logic level converter" board) carries all four lines:
- low side → Pi 3.3 V rail
- high side → IEC bus 5 V rail
- **DATA** is the one bidirectional, open-drain-emulated line.
- **ATN / CLK / RESET** are input-only (Pi always reads them).
- 100 nF bypass cap across the shifter VCC pins.
- **Common ground** between Pi GND and IEC pin 2.
- SRQ (IEC pin 1) left unconnected.
The shifter is **non-inverting**: bus low (asserted) ⇒ Pi reads LOW.
## Pin map
| IEC signal | IEC DIN pin | Pi GPIO (BCM) | Header pin | Direction |
|------------|-------------|---------------|------------|-----------|
| ATN | 3 | GPIO 2 | 3 | input (via shifter) |
| CLK | 4 | GPIO 17 | 11 | input (via shifter) |
| DATA | 5 | GPIO 18 | 12 | **bidirectional** (via shifter) |
| RESET | 6 | GPIO 3 | 5 | input (via shifter) |
| GND | 2 | GND | 6 (or any) | — |
| SRQ | 1 | — | — | not connected |
## DATA open-drain emulation
GPIO 18 is never a push-pull driver of the bus. The module flips its direction:
- **assert** (pull bus low): GPIO 18 = output **LOW** → shifter pulls bus to 0 V.
- **release / read**: GPIO 18 = **input (Hi-Z)** → bus pull-ups float it to 5 V,
and GPIO 18 reads the level the *talker* (C64) is driving — this is how bits
are sampled during receive.
Logical mapping (handled in `kernel/iec_lines.h`): asserted ⇔ Pi reads LOW.
## Bring-up checklist (before connecting the C64)
1. Load the module: `sudo insmod kernel/iec_listener.ko`.
2. Run the wiring selftest (drives DATA low, releases, reads back; reads sense
lines) via the `IEC_IOC_SELFTEST` ioctl — expect:
- DATA reads low while driven, high after release (Hi-Z floats up);
- ATN / CLK / RESET all released (high) with the C64 powered off.
3. Meter/LED-check each line for the expected idle-high.
4. Verify common ground continuity Pi↔IEC pin 2.
5. Only then connect the C64 and try `OPEN 1,4 : CLOSE 1`.

8
iecpoc/__init__.py Normal file
View File

@ -0,0 +1,8 @@
"""Userspace half of the Commodore IEC listener PoC.
The timing-critical IEC handshake runs in the kernel module under ``kernel/``;
this package decodes the tagged byte/record stream it produces and renders the
debug trace. See ``_plans/poc-listener-printer-PLAN.md``.
"""
__version__ = "0.1.0"

103
iecpoc/decode.py Normal file
View File

@ -0,0 +1,103 @@
"""Symbolic decode of IEC command bytes and state events for the trace.
The kernel module decodes *just enough* addressing to know whether it is the
addressed listener (PLAN.md §7). The rich, human-readable decode -- mnemonics,
primary/secondary address split, "is this for us?" -- lives here in userspace
where it costs nothing.
Command byte structure (research §2.4)::
$20+PA LISTEN pa $3F UNLISTEN
$40+PA TALK pa $5F UNTALK
$60+SA SECOND/DATA sa (reopen channel, after LISTEN/TALK)
$E0+SA CLOSE sa
$F0+SA OPEN sa (filename bytes follow in the data phase)
"""
from __future__ import annotations
from dataclasses import dataclass
from . import device
from .petscii import to_glyph
# command groups
LISTEN = "LISTEN"
UNLISTEN = "UNLISTEN"
TALK = "TALK"
UNTALK = "UNTALK"
SECOND = "DATA" # $60+SA, "reopen"/data channel
CLOSE = "CLOSE"
OPEN = "OPEN"
UNKNOWN = "?"
@dataclass(frozen=True)
class Command:
"""A decoded command byte."""
value: int
mnemonic: str
primary: int | None = None # primary address for LISTEN/TALK
secondary: int | None = None # secondary address for SECOND/CLOSE/OPEN
def targets(self, my_address: int) -> bool:
"""True if this command addresses the device at ``my_address``."""
return self.primary is not None and self.primary == my_address
def describe(self) -> str:
"""Short mnemonic + operand, e.g. ``LISTEN 4`` or ``OPEN SA=0``."""
if self.primary is not None:
return f"{self.mnemonic} {self.primary}"
if self.secondary is not None:
# pad mnemonic so SA-style commands line up in the trace
return f"{self.mnemonic:<5} SA={self.secondary}"
return self.mnemonic
def decode_command(value: int) -> Command:
"""Decode a single ATN-phase command byte."""
if not 0 <= value <= 0xFF:
raise ValueError(f"byte out of range: {value!r}")
if value == 0x3F:
return Command(value, UNLISTEN)
if value == 0x5F:
return Command(value, UNTALK)
if 0x20 <= value <= 0x3E:
return Command(value, LISTEN, primary=value - 0x20)
if 0x40 <= value <= 0x5E:
return Command(value, TALK, primary=value - 0x40)
if 0x60 <= value <= 0x6F:
return Command(value, SECOND, secondary=value - 0x60)
if 0xE0 <= value <= 0xEF:
return Command(value, CLOSE, secondary=value - 0xE0)
if 0xF0 <= value <= 0xFF:
return Command(value, OPEN, secondary=value - 0xF0)
return Command(value, UNKNOWN)
# Human-readable text for each state event (device.EV_*).
_EVENT_TEXT = {
device.EV_IDLE: "waiting for ATN",
device.EV_ATN_ASSERTED: "asserted -> DATA low (ack)",
device.EV_ATN_COMMAND: "asserted -> command phase",
device.EV_RESET: "RESET -> release bus, IDLE",
}
def describe_event(rec: device.IecRecord) -> str:
"""Human-readable text for an event record."""
if rec.value == device.EV_ATN_RELEASED:
if rec.addressed:
return "released -> LISTENER"
return "released -> not addressed -> IDLE"
return _EVENT_TEXT.get(rec.value, f"event ${rec.value:02X}")
def describe_data(value: int) -> str:
"""Render a data byte as the trace shows it: ``$48 'H'`` / ``$0D <CR>``."""
glyph = to_glyph(value)
if glyph.printable:
return f"${value:02X} '{glyph.text}'"
return f"${value:02X} {glyph.text}"

143
iecpoc/device.py Normal file
View File

@ -0,0 +1,143 @@
"""Userspace view of the ``/dev/iec0`` record stream.
The kernel module (``kernel/iec_listener.c``) pushes fixed-size ``struct
iec_record`` entries through a character device. This module mirrors that wire
format so the whole userspace side can be exercised on a host by replaying a
captured record file -- no Pi or kernel required (PLAN.md Phase 0).
Wire format (must match ``struct iec_record`` in ``kernel/iec_listener.h``)::
struct iec_record {
__u8 kind; // KIND_*
__u8 value; // the byte, or an EV_* code when kind == KIND_EVENT
__u8 flags; // FLAG_* bitmask
__u8 _pad;
__u64 ts_ns; // kernel monotonic timestamp (ktime_get_ns)
} __packed; // 12 bytes, little-endian on the Pi (ARM LE)
"""
from __future__ import annotations
import struct
from dataclasses import dataclass
from typing import BinaryIO, Iterable, Iterator
# --- record kinds -----------------------------------------------------------
KIND_COMMAND = 0 # a command byte received during the ATN phase
KIND_DATA = 1 # a data byte received in the data phase
KIND_EVENT = 2 # a state-machine event; ``value`` is an EV_* code
# --- record flags (bitmask) -------------------------------------------------
FLAG_EOI = 0x01 # this byte was the last in the stream (EOI signalled)
FLAG_ADDRESSED = 0x02 # this transfer / transition concerns *our* address
# --- event codes (value field when kind == KIND_EVENT) ----------------------
EV_IDLE = 0 # entered IDLE, waiting for ATN
EV_ATN_ASSERTED = 1 # ATN went low from IDLE -> we pulled DATA low (ack)
EV_ATN_COMMAND = 2 # ATN re-asserted mid-stream -> back to command phase
EV_ATN_RELEASED = 3 # ATN released; FLAG_ADDRESSED set => we become listener
EV_RESET = 4 # RESET asserted -> bus released, state cleared
# struct layout: kind, value, flags, 1 pad byte, u64 timestamp (little-endian)
RECORD_FMT = "<BBBxQ"
RECORD_SIZE = struct.calcsize(RECORD_FMT)
assert RECORD_SIZE == 12, RECORD_SIZE
@dataclass(frozen=True)
class IecRecord:
"""One decoded ``struct iec_record``."""
kind: int
value: int
flags: int
ts_ns: int
# -- convenience predicates ---------------------------------------------
@property
def is_command(self) -> bool:
return self.kind == KIND_COMMAND
@property
def is_data(self) -> bool:
return self.kind == KIND_DATA
@property
def is_event(self) -> bool:
return self.kind == KIND_EVENT
@property
def eoi(self) -> bool:
return bool(self.flags & FLAG_EOI)
@property
def addressed(self) -> bool:
return bool(self.flags & FLAG_ADDRESSED)
# -- (de)serialisation ---------------------------------------------------
@classmethod
def from_bytes(cls, raw: bytes) -> "IecRecord":
if len(raw) != RECORD_SIZE:
raise ValueError(f"expected {RECORD_SIZE} bytes, got {len(raw)}")
kind, value, flags, ts_ns = struct.unpack(RECORD_FMT, raw)
return cls(kind=kind, value=value, flags=flags, ts_ns=ts_ns)
def to_bytes(self) -> bytes:
return struct.pack(RECORD_FMT, self.kind, self.value, self.flags, self.ts_ns)
def iter_records(stream: BinaryIO) -> Iterator[IecRecord]:
"""Yield :class:`IecRecord` objects from a binary stream.
Reads exactly ``RECORD_SIZE`` bytes at a time, which matches the kernel
char device returning one record per ``read()``. Stops cleanly at EOF; a
trailing partial record raises ``ValueError`` (a corrupt capture).
"""
while True:
raw = stream.read(RECORD_SIZE)
if not raw:
return
if len(raw) < RECORD_SIZE:
raise ValueError(
f"truncated record: {len(raw)} of {RECORD_SIZE} bytes at EOF"
)
yield IecRecord.from_bytes(raw)
def write_records(stream: BinaryIO, records: Iterable[IecRecord]) -> None:
"""Serialise records back to a binary stream (used to build test captures)."""
for rec in records:
stream.write(rec.to_bytes())
class IecDevice:
"""Open and iterate the live kernel character device (or a replay file).
On the Pi this wraps ``/dev/iec0``. On a host pass a captured file to
``path`` to replay it through the exact same code path.
"""
def __init__(self, path: str = "/dev/iec0"):
self.path = path
self._fh: BinaryIO | None = None
def open(self) -> "IecDevice":
# buffering=0 so each read() maps to one kernel record on the real device.
self._fh = open(self.path, "rb", buffering=0)
return self
def __enter__(self) -> "IecDevice":
return self.open()
def __exit__(self, *exc) -> None:
self.close()
def close(self) -> None:
if self._fh is not None:
self._fh.close()
self._fh = None
def __iter__(self) -> Iterator[IecRecord]:
if self._fh is None:
raise RuntimeError("device not open; call open() or use as a context manager")
return iter_records(self._fh)

84
iecpoc/log.py Normal file
View File

@ -0,0 +1,84 @@
"""Render the annotated debug trace (PLAN.md §8).
Consumes :class:`~iecpoc.device.IecRecord` objects and emits one human-readable
line per record. This is the primary PoC deliverable on the userspace side, and
it is fully decoupled from any timing: it just formats whatever the kernel
pushed up. ``--raw`` switches to a bare hex dump of the byte stream.
"""
from __future__ import annotations
from typing import Iterable, Iterator
from . import decode, device
# bracketed state tag, padded so the content columns line up
_TAG_WIDTH = 8
def _line(tag: str, content: str) -> str:
return f"{f'[{tag}]':<{_TAG_WIDTH}} {content}"
class TraceFormatter:
"""Format an IEC record stream into the annotated trace.
Parameters
----------
my_address:
Our primary device address; used to annotate which commands target us.
raw:
When True, emit only a bare hex dump of command + data bytes and drop
the symbolic annotations and state events.
"""
def __init__(self, my_address: int = 4, raw: bool = False):
self.my_address = my_address
self.raw = raw
def format(self, rec: device.IecRecord) -> str | None:
"""Format one record. Returns ``None`` for records to be skipped."""
if self.raw:
# raw mode: just the byte value of command/data records
if rec.is_command or rec.is_data:
return f"{rec.value:02X}"
return None
if rec.is_event:
return self._format_event(rec)
if rec.is_command:
return self._format_command(rec)
if rec.is_data:
return self._format_data(rec)
return _line("?", f"unknown kind {rec.kind}")
def format_stream(self, records: Iterable[device.IecRecord]) -> Iterator[str]:
"""Format a stream, dropping skipped records."""
for rec in records:
line = self.format(rec)
if line is not None:
yield line
# -- per-kind formatters -------------------------------------------------
def _format_event(self, rec: device.IecRecord) -> str:
text = decode.describe_event(rec)
if rec.value == device.EV_IDLE:
tag = "IDLE"
elif rec.value == device.EV_RESET:
tag = "RESET"
else:
tag = "ATN"
return _line(tag, text)
def _format_command(self, rec: device.IecRecord) -> str:
cmd = decode.decode_command(rec.value)
content = f"${rec.value:02X} {cmd.describe()}"
if cmd.targets(self.my_address):
content = f"{content:<22} (addressed: ME)"
return _line("CMD", content)
def _format_data(self, rec: device.IecRecord) -> str:
content = decode.describe_data(rec.value)
if rec.eoi:
content = f"{content} <EOI>"
return _line("DATA", content)

91
iecpoc/main.py Normal file
View File

@ -0,0 +1,91 @@
"""CLI entry point for the IEC listener PoC userspace (PLAN.md §5).
Reads the tagged record stream from the kernel char device (``/dev/iec0``) -- or
from a captured file via ``--replay`` for host testing -- and prints the
annotated debug trace. The kernel module owns all timing; this process has none.
"""
from __future__ import annotations
import argparse
import os
import sys
from typing import Iterable, TextIO
from . import device
from .log import TraceFormatter
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="iecpoc",
description="Commodore IEC listener PoC -- decode and print what the C64 sends.",
)
p.add_argument(
"-a", "--address", type=int, default=4,
help="our primary device address (default: 4, a printer)",
)
p.add_argument(
"-d", "--device", default="/dev/iec0",
help="kernel character device to read (default: /dev/iec0)",
)
p.add_argument(
"--replay", metavar="FILE",
help="replay a captured record file instead of opening the device",
)
p.add_argument(
"--raw", action="store_true",
help="dump a bare hex stream instead of the annotated trace",
)
p.add_argument(
"--logfile", metavar="FILE",
help="also append the trace to this file",
)
return p
def run(records: Iterable[device.IecRecord], fmt: TraceFormatter,
out: TextIO, logfile: TextIO | None = None) -> None:
"""Format ``records`` through ``fmt`` and write lines to ``out`` (+ logfile)."""
try:
for line in fmt.format_stream(records):
print(line, file=out, flush=True)
if logfile is not None:
print(line, file=logfile, flush=True)
except BrokenPipeError:
# downstream (e.g. `head`) closed the pipe; redirect stdout to devnull so
# the interpreter's shutdown flush doesn't re-raise, then exit quietly.
devnull = os.open(os.devnull, os.O_WRONLY)
os.dup2(devnull, sys.stdout.fileno())
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
fmt = TraceFormatter(my_address=args.address, raw=args.raw)
logfile = open(args.logfile, "a") if args.logfile else None
try:
if args.replay:
with open(args.replay, "rb") as fh:
run(device.iter_records(fh), fmt, sys.stdout, logfile)
else:
try:
with device.IecDevice(args.device) as dev:
run(dev, fmt, sys.stdout, logfile)
except FileNotFoundError:
print(
f"error: {args.device} not found -- is the kernel module loaded? "
f"(use --replay FILE to replay a capture on a host)",
file=sys.stderr,
)
return 1
except KeyboardInterrupt:
return 0
finally:
if logfile is not None:
logfile.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())

95
iecpoc/petscii.py Normal file
View File

@ -0,0 +1,95 @@
"""PETSCII -> display-string conversion for the debug trace.
Data bytes the C64 sends to a printer are PETSCII, not ASCII. For the upper-
case/graphics character set (the C64 power-on default) the printable
sub-ranges line up with ASCII for letters, digits and most punctuation, with a
handful of Commodore-specific glyphs (``£ ``). Control codes (RETURN, RVS,
colour changes, cursor moves) are rendered with short symbolic names so the
trace stays readable.
This is deliberately a *display* mapping for logging, not a font renderer. The
goal (PLAN.md §8) is: printable PETSCII -> its glyph; control code -> ``<NAME>``;
anything else (graphics characters) -> ``.``.
"""
from __future__ import annotations
from dataclasses import dataclass
# Symbolic names for the C64 control codes that actually turn up in a printer
# byte stream. Anything not in here renders as ``<$xx>``.
_CONTROL_NAMES = {
0x03: "STOP",
0x05: "WHT",
0x08: "SHIFT-DIS",
0x09: "SHIFT-EN",
0x0A: "LF",
0x0D: "CR",
0x0E: "LOWER",
0x11: "DOWN",
0x12: "RVS-ON",
0x13: "HOME",
0x14: "DEL",
0x1C: "RED",
0x1D: "RIGHT",
0x1E: "GRN",
0x1F: "BLU",
0x81: "ORG",
0x8D: "SHIFT-CR",
0x90: "BLK",
0x91: "UP",
0x92: "RVS-OFF",
0x93: "CLR",
0x94: "INST",
0x95: "BRN",
0x99: "GRY",
0x9D: "LEFT",
0x9E: "YEL",
0x9F: "CYN",
}
# Printable glyphs that differ from plain ASCII in the upper-case/graphics set.
_GLYPH_OVERRIDES = {
0x5C: "£", # £
0x5E: "", # ↑
0x5F: "", # ←
0xA0: " ", # shifted space
}
CAT_PRINTABLE = "printable"
CAT_CONTROL = "control"
CAT_GRAPHIC = "graphic"
@dataclass(frozen=True)
class Glyph:
"""The rendered form of one PETSCII byte."""
text: str # what to show in the trace
category: str # CAT_PRINTABLE | CAT_CONTROL | CAT_GRAPHIC
@property
def printable(self) -> bool:
return self.category == CAT_PRINTABLE
def to_glyph(byte: int) -> Glyph:
"""Map one PETSCII byte to a :class:`Glyph` for the debug trace."""
if not 0 <= byte <= 0xFF:
raise ValueError(f"byte out of range: {byte!r}")
if byte in _GLYPH_OVERRIDES:
return Glyph(_GLYPH_OVERRIDES[byte], CAT_PRINTABLE)
# Space through '_' (minus the overrides above) matches ASCII directly.
if 0x20 <= byte <= 0x5F:
return Glyph(chr(byte), CAT_PRINTABLE)
# Control ranges: $00-$1F and $80-$9F.
if byte <= 0x1F or 0x80 <= byte <= 0x9F:
name = _CONTROL_NAMES.get(byte, f"${byte:02X}")
return Glyph(f"<{name}>", CAT_CONTROL)
# Everything else ($60-$7F, $A0-$FF) is graphics / shifted glyphs in the
# upper-case set -- not meaningful as printer text, render as '.'.
return Glyph(".", CAT_GRAPHIC)

48
kernel/Dockerfile Normal file
View File

@ -0,0 +1,48 @@
# syntax=docker/dockerfile:1
#
# Build the IEC listener kernel module for Raspberry Pi OS (arm64) inside an
# emulated arm64 container. The module is compiled natively against the same
# raspberrypi kernel headers the Pi runs, which keeps the module "vermagic" in
# sync so `insmod` accepts it on the Pi.
#
# Requires qemu-binfmt on the host (the build-in-docker.sh wrapper sets this up):
# docker run --privileged --rm tonistiigi/binfmt --install arm64
#
# Kernel version is configurable via build args:
# --build-arg HEADERS_PKG=linux-headers-rpi-v8 # 64-bit Pi Zero 2 W (default)
# --build-arg KERNEL_VERSION=1:6.6.51-1+rpt3 # optional exact pin (see notes)
#
# Build context is this kernel/ directory.
FROM --platform=linux/arm64 debian:bookworm
# The raspberrypi kernel/headers live in the raspberrypi.com archive, not Debian.
RUN apt-get update \
&& apt-get install -y --no-install-recommends ca-certificates curl gnupg \
&& curl -fsSL https://archive.raspberrypi.com/debian/raspberrypi.gpg.key \
| gpg --dearmor -o /usr/share/keyrings/raspberrypi-archive-keyring.gpg \
&& echo "deb [signed-by=/usr/share/keyrings/raspberrypi-archive-keyring.gpg] http://archive.raspberrypi.com/debian/ bookworm main" \
> /etc/apt/sources.list.d/raspi.list \
&& rm -rf /var/lib/apt/lists/*
# Toolchain + module build dependencies.
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
build-essential bc bison flex libssl-dev libelf-dev kmod make \
&& rm -rf /var/lib/apt/lists/*
# Kernel headers. Override HEADERS_PKG / KERNEL_VERSION to target a specific Pi
# kernel. Default targets the current 64-bit Raspberry Pi OS Bookworm kernel.
ARG HEADERS_PKG=linux-headers-rpi-v8
ARG KERNEL_VERSION=
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
"${HEADERS_PKG}${KERNEL_VERSION:+=${KERNEL_VERSION}}" \
&& rm -rf /var/lib/apt/lists/*
COPY docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh
RUN chmod +x /usr/local/bin/docker-entrypoint.sh
WORKDIR /build
ENTRYPOINT ["/usr/local/bin/docker-entrypoint.sh"]
# default: build the module (override with e.g. `clean`)
CMD []

52
kernel/Makefile Normal file
View File

@ -0,0 +1,52 @@
# Out-of-tree build for the IEC listener kernel module.
# Build ON the Pi (or cross-compile) against the running kernel's headers:
# sudo apt install raspberrypi-kernel-headers
# make # builds iec_listener.ko
# sudo make load # insmod + show /dev/iec0
# sudo make unload
#
# See docs/kernel-notes.md for headers/pinning details.
MODULE_NAME := iec_listener
obj-m += $(MODULE_NAME).o
# iec_listener.o is built from iec_listener.c; the other headers are includes.
KDIR ?= /lib/modules/$(shell uname -r)/build
PWD := $(shell pwd)
# overlay
DTS := dts/iec-overlay.dts
DTBO := dts/iec-overlay.dtbo
.PHONY: all clean load unload overlay install docker-build docker-clean
all:
$(MAKE) -C $(KDIR) M=$(PWD) modules
# Cross-build in an emulated arm64 container (for compiling off the Pi).
# Override e.g. KERNEL_VERSION=... HEADERS_PKG=... ; see build-in-docker.sh.
docker-build:
./build-in-docker.sh
docker-clean:
./build-in-docker.sh clean
clean:
$(MAKE) -C $(KDIR) M=$(PWD) clean
rm -f $(DTBO)
# Compile the device-tree overlay (pin reservation / pulls).
overlay: $(DTBO)
$(DTBO): $(DTS)
dtc -@ -I dts -O dtb -o $@ $<
load: all
sudo insmod $(MODULE_NAME).ko $(if $(ADDRESS),address=$(ADDRESS),)
@echo "loaded; device:" && ls -l /dev/iec0
unload:
sudo rmmod $(MODULE_NAME)
install: all
$(MAKE) -C $(KDIR) M=$(PWD) modules_install
sudo depmod -a

48
kernel/build-in-docker.sh Executable file
View File

@ -0,0 +1,48 @@
#!/usr/bin/env bash
#
# Build the IEC listener kernel module in an emulated arm64 container, so it can
# be compiled on a non-Pi (x86) host. The resulting iec_listener.ko is written
# back into this kernel/ directory on the host.
#
# Usage:
# ./build-in-docker.sh # build the module
# ./build-in-docker.sh clean # clean build artifacts
#
# Configurable via environment variables:
# HEADERS_PKG headers package (default: linux-headers-rpi-v8, 64-bit Zero 2 W)
# e.g. linux-headers-rpi-v7 / -v6 for 32-bit, raspberrypi-kernel-headers
# KERNEL_VERSION exact version to pin, e.g. 1:6.6.51-1+rpt3 (default: latest in repo)
# IMAGE builder image tag (default: iec-kbuild)
#
# NOTE: the raspberrypi apt archive generally serves only the *latest* kernel in
# its pool, so pinning KERNEL_VERSION to an old release may fail to download. The
# reliable match strategy is to keep the Pi current (`sudo apt full-upgrade`) and
# build with the default (latest). See docs/kernel-notes.md.
set -euo pipefail
cd "$(dirname "$0")"
IMAGE="${IMAGE:-iec-kbuild}"
HEADERS_PKG="${HEADERS_PKG:-linux-headers-rpi-v8}"
KERNEL_VERSION="${KERNEL_VERSION:-}"
# 1. Make sure the host can run arm64 binaries under qemu (one-time, privileged).
if [ ! -e /proc/sys/fs/binfmt_misc/qemu-aarch64 ]; then
echo "==> registering qemu arm64 binfmt (one-time, needs a privileged container)"
docker run --privileged --rm tonistiigi/binfmt --install arm64
fi
# 2. Build the builder image (installs toolchain + matching kernel headers).
echo "==> building image '$IMAGE' (HEADERS_PKG=$HEADERS_PKG KERNEL_VERSION=${KERNEL_VERSION:-latest})"
docker build --platform linux/arm64 \
--build-arg HEADERS_PKG="$HEADERS_PKG" \
--build-arg KERNEL_VERSION="$KERNEL_VERSION" \
-t "$IMAGE" .
# 3. Compile; mount this dir so the .ko lands on the host, owned by the caller.
echo "==> compiling module"
docker run --rm --platform linux/arm64 \
-u "$(id -u):$(id -g)" -e HOME=/tmp \
-v "$PWD:/build" "$IMAGE" "$@"
echo "==> done"
ls -l iec_listener.ko 2>/dev/null || echo "(no iec_listener.ko produced — see output above)"

17
kernel/docker-entrypoint.sh Executable file
View File

@ -0,0 +1,17 @@
#!/bin/sh
# Resolve the installed Raspberry Pi kernel version and build against it.
#
# `uname -r` inside a container returns the HOST kernel (the container shares the
# host kernel), so we must derive KDIR from the headers actually installed in the
# image rather than from uname. Pick the newest installed module tree.
set -eu
KVER="$(ls -1 /lib/modules 2>/dev/null | sort -V | tail -n 1 || true)"
if [ -z "$KVER" ] || [ ! -d "/lib/modules/$KVER/build" ]; then
echo "error: no kernel headers found under /lib/modules/*/build" >&2
echo " (the HEADERS_PKG build-arg may be wrong for this image)" >&2
exit 1
fi
echo "==> building against kernel headers: $KVER"
exec make KDIR="/lib/modules/$KVER/build" "$@"

View File

@ -0,0 +1,39 @@
/*
* iec-overlay.dts - pin reservation + pull configuration for the IEC listener.
*
* GPIO 2 = ATN (input, pull-up: bus idle = 5 V = high)
* GPIO 17 = CLK (input, pull-up)
* GPIO 3 = RESET (input, pull-up)
* GPIO 18 = DATA (bidirectional; starts as input/Hi-Z, the module flips it)
*
* The module drives GPIO via direct registers, so this overlay's main job is to
* reserve the pins and set sane pull defaults. Build & install:
* dtc -@ -I dts -O dtb -o iec-overlay.dtbo iec-overlay.dts
* sudo cp iec-overlay.dtbo /boot/firmware/overlays/ # Bookworm path
* echo "dtoverlay=iec-overlay" | sudo tee -a /boot/firmware/config.txt
* # reboot, then: dtoverlay -l
*
* See _research/pi-kernel-module-rt-gpio-2026-06-18.md §4.1.
*/
/dts-v1/;
/plugin/;
/ {
compatible = "brcm,bcm2837";
fragment@0 {
target = <&gpio>;
__overlay__ {
iec_input_pins: iec_input_pins {
brcm,pins = <2 17 3>; /* ATN, CLK, RESET */
brcm,function = <0>; /* 0 = input */
brcm,pull = <2>; /* 2 = pull-up */
};
iec_data_pin: iec_data_pin {
brcm,pins = <18>; /* DATA */
brcm,function = <0>; /* start as input (Hi-Z) */
brcm,pull = <0>; /* 0 = none (shifter drives) */
};
};
};
};

106
kernel/iec_lines.h Normal file
View File

@ -0,0 +1,106 @@
/* SPDX-License-Identifier: GPL-2.0 */
/*
* iec_lines.h - logical IEC line layer (PLAN.md §5.1).
*
* The handshake code reasons in protocol true/false terms (asserted / released)
* and never juggles electrical inversions inline. With the PLAN's non-inverting
* bidirectional level shifter (sd2iec-style single DATA pin, §3.1):
*
* bus asserted (0 V) <=> Pi GPIO reads LOW
* bus released (5 V) <=> Pi GPIO reads HIGH
*
* DATA is open-drain emulated on a single bidirectional pin:
* assert : drive GPIO18 OUTPUT LOW -> shifter pulls bus to 0 V
* release : set GPIO18 INPUT (Hi-Z) -> bus pull-ups float it to 5 V;
* the talker's bits can now be read back through the shifter.
*
* The hot path uses direct BCM register access (ioremap'd), which the kernel
* research found to be ~30-40x faster than the gpiod descriptor API and is what
* ninepin/raspbiec use inside the bit loop.
*
* NOTE: this differs from the 7406 *inverting* scheme analysed in
* _research/pi-kernel-module-rt-gpio-2026-06-18.md §4.3. The PLAN deliberately
* chose the non-inverting bidirectional shifter, which also closes that doc's
* "DATA sensing gap" (reading back through the shifter while the pin is an input
* IS valid). See docs/kernel-notes.md.
*/
#ifndef _IEC_LINES_H
#define _IEC_LINES_H
#include <linux/io.h>
/* --- GPIO assignment (BCM numbering, PLAN.md §3.2) ----------------------- */
#define IEC_GPIO_ATN 2 /* header pin 3 - input */
#define IEC_GPIO_RESET 3 /* header pin 5 - input */
#define IEC_GPIO_CLK 17 /* header pin 11 - input */
#define IEC_GPIO_DATA 18 /* header pin 12 - bidirectional (open-drain) */
/* --- BCM2710A1 / BCM2837 (Pi Zero 2 W) peripheral map ------------------- */
/* Same peripheral base as RPi 3. RPi 1 = 0x20000000, RPi 4 = 0xFE000000. */
#define IEC_BCM_PERI_BASE 0x3F000000UL
#define IEC_GPIO_BASE (IEC_BCM_PERI_BASE + 0x200000UL)
#define IEC_GPIO_LEN 0x1000
/* Register word offsets (multiply by 4 for byte offset) */
#define GPFSEL0 0 /* function select base (3 bits/pin, 10 pins/reg) */
#define GPSET0 7 /* output set (write 1 to set pin high) */
#define GPCLR0 10 /* output clear (write 1 to set pin low) */
#define GPLEV0 13 /* pin level (read) */
/* ioremap'd base of the GPIO block; set in module init. */
extern u32 __iomem *iec_gpio;
/* --- low-level direct register helpers ---------------------------------- */
static inline int iec_gpio_read(unsigned int pin)
{
return (ioread32(iec_gpio + GPLEV0) >> (pin & 31)) & 1;
}
static inline void iec_gpio_set(unsigned int pin) /* drive high */
{
iowrite32(1u << (pin & 31), iec_gpio + GPSET0);
}
static inline void iec_gpio_clr(unsigned int pin) /* drive low */
{
iowrite32(1u << (pin & 31), iec_gpio + GPCLR0);
}
/* Set the 3-bit function field for a pin: 0 = input, 1 = output. */
static inline void iec_gpio_fsel(unsigned int pin, unsigned int fn)
{
u32 __iomem *reg = iec_gpio + GPFSEL0 + (pin / 10);
unsigned int shift = (pin % 10) * 3;
u32 v = ioread32(reg);
v &= ~(0x7u << shift);
v |= (fn & 0x7u) << shift;
iowrite32(v, reg);
}
#define IEC_FSEL_INPUT 0
#define IEC_FSEL_OUTPUT 1
/* --- logical (protocol-level) line operations --------------------------- */
/* sense lines: asserted == Pi reads LOW (non-inverting shifter) */
static inline bool iec_atn_asserted(void) { return iec_gpio_read(IEC_GPIO_ATN) == 0; }
static inline bool iec_clk_asserted(void) { return iec_gpio_read(IEC_GPIO_CLK) == 0; }
static inline bool iec_reset_asserted(void) { return iec_gpio_read(IEC_GPIO_RESET) == 0; }
/* DATA, open-drain emulation on the single bidirectional pin */
static inline void iec_data_assert(void) /* pull bus low */
{
iec_gpio_clr(IEC_GPIO_DATA); /* preload output latch LOW */
iec_gpio_fsel(IEC_GPIO_DATA, IEC_FSEL_OUTPUT);
}
static inline void iec_data_release(void) /* float bus high (Hi-Z) */
{
iec_gpio_fsel(IEC_GPIO_DATA, IEC_FSEL_INPUT);
}
/* read the DATA bus state (valid only after iec_data_release()) */
static inline bool iec_data_asserted(void) { return iec_gpio_read(IEC_GPIO_DATA) == 0; }
#endif /* _IEC_LINES_H */

483
kernel/iec_listener.c Normal file
View File

@ -0,0 +1,483 @@
// SPDX-License-Identifier: GPL-2.0
/*
* iec_listener.c - Commodore IEC *listener* device for Raspberry Pi Zero 2 W.
*
* The timing-critical IEC handshake (ATN ack, per-bit sampling, per-byte and
* EOI acknowledge) runs here in the kernel, above the normal scheduler. Every
* received byte and state transition is pushed as a `struct iec_record` up to
* userspace through the /dev/iec0 character device; all decoding, PETSCII and
* pretty-printing happen in the Python userspace (iecpoc/).
*
* Architecture (validated in _research/pi-kernel-module-rt-gpio-2026-06-18.md):
* - ATN falling-edge hardirq pulls DATA low immediately (<= 1 ms presence ack)
* and wakes a worker kthread.
* - The worker runs the listener state machine; each byte is received by
* receive_byte(), which busy-polls CLK with local_irq_save() held for the
* ~200 us duration of one byte (ninepin pattern).
* - Hot-path GPIO uses direct BCM register access (iec_lines.h); init/exit use
* the gpiod descriptor API.
*
* Scope: listener only (printer, default address 4). No talker / turnaround.
* See _plans/poc-listener-printer-PLAN.md §6, §7.
*/
#include <linux/cdev.h>
#include <linux/delay.h>
#include <linux/fs.h>
#include <linux/gpio/consumer.h>
#include <linux/interrupt.h>
#include <linux/kfifo.h>
#include <linux/kthread.h>
#include <linux/ktime.h>
#include <linux/module.h>
#include <linux/poll.h>
#include <linux/uaccess.h>
#include <linux/wait.h>
#include "iec_listener.h"
#include "iec_lines.h"
#include "iec_timing.h"
MODULE_LICENSE("GPL");
MODULE_AUTHOR("Christian Werner");
MODULE_DESCRIPTION("Commodore IEC listener (printer) device for Pi Zero 2 W");
MODULE_VERSION("0.1");
/* --- module parameters --------------------------------------------------- */
static int address = 4; /* our primary device address (printer) */
module_param(address, int, 0644);
MODULE_PARM_DESC(address, "primary IEC device address (default 4)");
/* --- globals ------------------------------------------------------------- */
u32 __iomem *iec_gpio; /* defined extern in iec_lines.h; ioremap'd in init */
static struct gpio_desc *gd_atn, *gd_clk, *gd_reset, *gd_data;
static int irq_atn;
#define IEC_FIFO_SIZE 256 /* records; power of two */
static DECLARE_KFIFO(iec_fifo, struct iec_record, IEC_FIFO_SIZE);
static DECLARE_WAIT_QUEUE_HEAD(iec_read_wq);
static DEFINE_SPINLOCK(iec_fifo_lock);
static dev_t iec_devno;
static struct cdev iec_cdev;
static struct class *iec_class;
/* worker thread + wakeup flag set by the ATN ISR */
static struct task_struct *iec_worker_task;
static atomic_t iec_atn_pending = ATOMIC_INIT(0);
static DECLARE_WAIT_QUEUE_HEAD(iec_work_wq);
/* listener addressing state (decoded in-kernel, just enough to participate) */
static bool addressed_listener;
/* receive_byte() outcomes */
enum iec_rx {
IEC_RX_OK = 0, /* byte received */
IEC_RX_ATN, /* ATN asserted mid-byte -> abort */
IEC_RX_RESET, /* RESET asserted -> abort */
IEC_RX_TIMEOUT, /* CLK did not transition in time */
};
/* --- record emission ----------------------------------------------------- */
static void emit_record(u8 kind, u8 value, u8 flags)
{
struct iec_record rec = {
.kind = kind,
.value = value,
.flags = flags,
.ts_ns = ktime_get_ns(),
};
unsigned long lflags;
spin_lock_irqsave(&iec_fifo_lock, lflags);
kfifo_put(&iec_fifo, rec);
spin_unlock_irqrestore(&iec_fifo_lock, lflags);
wake_up_interruptible(&iec_read_wq);
}
/* --- busy-poll helper ---------------------------------------------------- */
/*
* Spin until CLK reaches `want_asserted`, or timeout/ATN/RESET. Caller holds
* local IRQs disabled. Uses a microsecond budget rather than a fixed udelay so
* a slightly-fast C64 is handled correctly (research §1.3).
*/
static enum iec_rx wait_clk(bool want_asserted, unsigned int timeout_us)
{
unsigned int waited = 0;
while (iec_clk_asserted() != want_asserted) {
if (iec_atn_asserted())
return IEC_RX_ATN;
if (iec_reset_asserted())
return IEC_RX_RESET;
if (waited++ >= timeout_us)
return IEC_RX_TIMEOUT;
udelay(1);
}
return IEC_RX_OK;
}
/* --- receive one byte (PLAN.md §6) --------------------------------------- */
/*
* Entry: we hold DATA asserted (low). Returns the byte in *out and EOI in *eoi.
* `data_phase` enables EOI detection (false during the ATN command phase).
*/
static enum iec_rx receive_byte(u8 *out, bool *eoi, bool data_phase)
{
unsigned long flags;
enum iec_rx rc;
u8 value = 0;
int i;
*eoi = false;
/* 1. READY-FOR-DATA: wait for talker to release CLK, then release DATA */
rc = wait_clk(false /* released */, IEC_CLK_TIMEOUT_US);
if (rc != IEC_RX_OK)
return rc;
iec_data_release();
local_irq_save(flags); /* IRQs off for the duration of the byte */
/* 2. EOI DETECT (data phase only): no CLK assertion within Tye => EOI */
if (data_phase) {
unsigned int waited = 0;
while (!iec_clk_asserted()) {
if (iec_atn_asserted()) { rc = IEC_RX_ATN; goto out; }
if (iec_reset_asserted()) { rc = IEC_RX_RESET; goto out; }
if (waited++ >= IEC_EOI_DETECT_US) {
/* ack EOI: pull DATA low for Tei, then release */
iec_data_assert();
udelay(IEC_EOI_ACK_HOLD_US);
iec_data_release();
*eoi = true;
break;
}
udelay(1);
}
}
/* 3. RECEIVE 8 BITS, LSB first */
for (i = 0; i < 8; i++) {
rc = wait_clk(true /* asserted: data invalid/setup */, IEC_CLK_TIMEOUT_US);
if (rc != IEC_RX_OK)
goto out;
rc = wait_clk(false /* released: data valid -> sample */, IEC_CLK_TIMEOUT_US);
if (rc != IEC_RX_OK)
goto out;
/* released(high) = bit 1, asserted(low) = bit 0 */
if (!iec_data_asserted())
value |= (1u << i);
}
/* 4. BYTE ACKNOWLEDGE: pull DATA low (Tf), hold for the next RFD */
iec_data_assert();
*out = value;
rc = IEC_RX_OK;
out:
local_irq_restore(flags);
return rc;
}
/* --- addressing decode (in-kernel subset, PLAN.md §7) -------------------- */
static void decode_addressing(u8 b)
{
if (b == 0x3F) { /* UNLISTEN */
addressed_listener = false;
} else if (b >= 0x20 && b <= 0x3E) { /* LISTEN n */
addressed_listener = ((b - 0x20) == address);
}
/* TALK/UNTALK/secondary: logged in userspace; listener ignores for now */
}
/* --- state machine (PLAN.md §7) ------------------------------------------ */
/*
* Runs in the worker kthread after the ATN ISR has pulled DATA and woken us.
* Returns when the bus is back to IDLE (ATN released & not addressed, or RESET).
*/
static void run_state_machine(void)
{
emit_record(IEC_KIND_EVENT, IEC_EV_ATN_ASSERTED, 0);
for (;;) {
u8 b, flags;
bool eoi;
enum iec_rx rc;
if (iec_reset_asserted())
goto reset;
if (iec_atn_asserted()) {
/* RECEIVE_COMMAND: ATN held low */
rc = receive_byte(&b, &eoi, false);
if (rc == IEC_RX_RESET)
goto reset;
if (rc != IEC_RX_OK)
continue; /* ATN change / timeout: re-evaluate */
decode_addressing(b);
flags = addressed_listener ? IEC_FLAG_ADDRESSED : 0;
emit_record(IEC_KIND_COMMAND, b, flags);
continue;
}
/* ATN released */
if (!addressed_listener) {
emit_record(IEC_KIND_EVENT, IEC_EV_ATN_RELEASED, 0);
iec_data_release();
return; /* not our business -> IDLE */
}
/* LISTENER: data phase */
emit_record(IEC_KIND_EVENT, IEC_EV_ATN_RELEASED, IEC_FLAG_ADDRESSED);
for (;;) {
if (iec_reset_asserted())
goto reset;
if (iec_atn_asserted()) {
/* C64 interrupts with a new command */
emit_record(IEC_KIND_EVENT, IEC_EV_ATN_COMMAND, 0);
break;
}
rc = receive_byte(&b, &eoi, true);
if (rc == IEC_RX_RESET)
goto reset;
if (rc == IEC_RX_ATN) {
emit_record(IEC_KIND_EVENT, IEC_EV_ATN_COMMAND, 0);
break;
}
if (rc != IEC_RX_OK)
continue;
emit_record(IEC_KIND_DATA, b,
IEC_FLAG_ADDRESSED | (eoi ? IEC_FLAG_EOI : 0));
}
}
reset:
addressed_listener = false;
iec_data_release();
emit_record(IEC_KIND_EVENT, IEC_EV_RESET, 0);
}
static int iec_worker(void *unused)
{
emit_record(IEC_KIND_EVENT, IEC_EV_IDLE, 0);
while (!kthread_should_stop()) {
wait_event_interruptible(iec_work_wq,
atomic_read(&iec_atn_pending) || kthread_should_stop());
if (kthread_should_stop())
break;
atomic_set(&iec_atn_pending, 0);
run_state_machine();
emit_record(IEC_KIND_EVENT, IEC_EV_IDLE, 0);
}
return 0;
}
/* --- ATN interrupt: meet the <= 1 ms presence deadline ------------------- */
static irqreturn_t atn_isr(int irq, void *dev)
{
if (iec_atn_asserted()) {
iec_data_assert(); /* pull DATA low immediately (ack) */
atomic_set(&iec_atn_pending, 1);
wake_up_interruptible(&iec_work_wq);
}
return IRQ_HANDLED;
}
/* --- character device ---------------------------------------------------- */
static ssize_t iec_read(struct file *f, char __user *buf, size_t count, loff_t *ppos)
{
struct iec_record rec;
unsigned long lflags;
int got, ret;
if (count < sizeof(rec))
return -EINVAL;
for (;;) {
spin_lock_irqsave(&iec_fifo_lock, lflags);
got = kfifo_get(&iec_fifo, &rec);
spin_unlock_irqrestore(&iec_fifo_lock, lflags);
if (got)
break;
if (f->f_flags & O_NONBLOCK)
return -EAGAIN;
ret = wait_event_interruptible(iec_read_wq, !kfifo_is_empty(&iec_fifo));
if (ret)
return ret; /* -ERESTARTSYS */
}
if (copy_to_user(buf, &rec, sizeof(rec)))
return -EFAULT;
return sizeof(rec);
}
static __poll_t iec_poll(struct file *f, poll_table *wait)
{
poll_wait(f, &iec_read_wq, wait);
return kfifo_is_empty(&iec_fifo) ? 0 : (EPOLLIN | EPOLLRDNORM);
}
static u32 run_selftest(void)
{
u32 r = 0;
/* drive DATA low, read it back */
iec_data_assert();
udelay(5);
if (iec_data_asserted())
r |= IEC_SELFTEST_DATA_ASSERT_OK;
/* release to Hi-Z; bus pull-up should float it high */
iec_data_release();
udelay(5);
if (!iec_data_asserted())
r |= IEC_SELFTEST_DATA_FLOAT_OK;
if (!iec_atn_asserted())
r |= IEC_SELFTEST_ATN_RELEASED;
if (!iec_clk_asserted())
r |= IEC_SELFTEST_CLK_RELEASED;
if (!iec_reset_asserted())
r |= IEC_SELFTEST_RESET_RELEASED;
return r;
}
static long iec_ioctl(struct file *f, unsigned int cmd, unsigned long arg)
{
u32 v;
switch (cmd) {
case IEC_IOC_SET_ADDRESS:
if (copy_from_user(&v, (void __user *)arg, sizeof(v)))
return -EFAULT;
if (v > 30)
return -EINVAL;
address = v;
return 0;
case IEC_IOC_GET_ADDRESS:
v = address;
return copy_to_user((void __user *)arg, &v, sizeof(v)) ? -EFAULT : 0;
case IEC_IOC_SELFTEST:
v = run_selftest();
return copy_to_user((void __user *)arg, &v, sizeof(v)) ? -EFAULT : 0;
default:
return -ENOTTY;
}
}
static const struct file_operations iec_fops = {
.owner = THIS_MODULE,
.read = iec_read,
.poll = iec_poll,
.unlocked_ioctl = iec_ioctl,
/* nonseekable_open() marks the fd non-seekable; no_llseek was removed in
* 6.12 so we deliberately leave .llseek unset (works on 6.6 and 6.12). */
.open = nonseekable_open,
};
/* --- module init / exit -------------------------------------------------- */
static int __init iec_init(void)
{
int ret;
if (address < 0 || address > 30) {
pr_err("iec: invalid address %d (0-30)\n", address);
return -EINVAL;
}
INIT_KFIFO(iec_fifo);
/* map GPIO block for the direct-register hot path */
iec_gpio = ioremap(IEC_GPIO_BASE, IEC_GPIO_LEN);
if (!iec_gpio)
return -ENOMEM;
/* claim GPIOs via descriptor API (init/exit only) */
gd_atn = gpio_to_desc(IEC_GPIO_ATN);
gd_clk = gpio_to_desc(IEC_GPIO_CLK);
gd_reset = gpio_to_desc(IEC_GPIO_RESET);
gd_data = gpio_to_desc(IEC_GPIO_DATA);
if (!gd_atn || !gd_clk || !gd_reset || !gd_data) {
ret = -ENODEV;
goto err_unmap;
}
gpiod_direction_input(gd_atn);
gpiod_direction_input(gd_clk);
gpiod_direction_input(gd_reset);
/* DATA starts released (Hi-Z / input) so the bus is free */
iec_data_release();
/* char device + class */
ret = alloc_chrdev_region(&iec_devno, 0, 1, "iec");
if (ret)
goto err_unmap;
cdev_init(&iec_cdev, &iec_fops);
ret = cdev_add(&iec_cdev, iec_devno, 1);
if (ret)
goto err_region;
iec_class = class_create("iec");
if (IS_ERR(iec_class)) {
ret = PTR_ERR(iec_class);
goto err_cdev;
}
device_create(iec_class, NULL, iec_devno, NULL, "iec0");
/* worker thread */
iec_worker_task = kthread_run(iec_worker, NULL, "iec_worker");
if (IS_ERR(iec_worker_task)) {
ret = PTR_ERR(iec_worker_task);
goto err_class;
}
/* ATN falling-edge interrupt */
irq_atn = gpiod_to_irq(gd_atn);
if (irq_atn < 0) {
ret = irq_atn;
goto err_thread;
}
ret = request_irq(irq_atn, atn_isr, IRQF_TRIGGER_FALLING, "iec_atn", NULL);
if (ret)
goto err_thread;
pr_info("iec: listener ready, address %d, /dev/iec0\n", address);
return 0;
err_thread:
kthread_stop(iec_worker_task);
err_class:
device_destroy(iec_class, iec_devno);
class_destroy(iec_class);
err_cdev:
cdev_del(&iec_cdev);
err_region:
unregister_chrdev_region(iec_devno, 1);
err_unmap:
iounmap(iec_gpio);
return ret;
}
static void __exit iec_exit(void)
{
free_irq(irq_atn, NULL);
if (iec_worker_task)
kthread_stop(iec_worker_task);
/* CRITICAL: release DATA so a stuck-low bus does not hang the C64 */
iec_data_release();
device_destroy(iec_class, iec_devno);
class_destroy(iec_class);
cdev_del(&iec_cdev);
unregister_chrdev_region(iec_devno, 1);
iounmap(iec_gpio);
pr_info("iec: unloaded, DATA released\n");
}
module_init(iec_init);
module_exit(iec_exit);

67
kernel/iec_listener.h Normal file
View File

@ -0,0 +1,67 @@
/* SPDX-License-Identifier: GPL-2.0 */
/*
* iec_listener.h - shared record format & ioctls for the IEC listener module.
*
* This header is included by both the kernel module (iec_listener.c) and,
* conceptually, mirrored by the Python userspace (iecpoc/device.py). The
* struct layout below MUST stay in sync with RECORD_FMT in device.py
* ("<BBBxQ", 12 bytes, little-endian).
*
* See _plans/poc-listener-printer-PLAN.md §5.2.
*/
#ifndef _IEC_LISTENER_H
#define _IEC_LISTENER_H
#include <linux/types.h>
#include <linux/ioctl.h>
/* record.kind */
#define IEC_KIND_COMMAND 0 /* command byte received during ATN phase */
#define IEC_KIND_DATA 1 /* data byte received in the data phase */
#define IEC_KIND_EVENT 2 /* state event; value is an IEC_EV_* code */
/* record.flags (bitmask) */
#define IEC_FLAG_EOI 0x01 /* last byte in the stream (EOI signalled) */
#define IEC_FLAG_ADDRESSED 0x02 /* transfer/transition concerns our address */
/* event codes (record.value when kind == IEC_KIND_EVENT) */
#define IEC_EV_IDLE 0 /* entered IDLE, waiting for ATN */
#define IEC_EV_ATN_ASSERTED 1 /* ATN low from IDLE -> pulled DATA (ack) */
#define IEC_EV_ATN_COMMAND 2 /* ATN re-asserted mid-stream -> cmd phase */
#define IEC_EV_ATN_RELEASED 3 /* ATN released (FLAG_ADDRESSED => listener)*/
#define IEC_EV_RESET 4 /* RESET asserted -> bus released, cleared */
/*
* One record pushed up through /dev/iec0. Fixed size, __packed so the on-wire
* layout is identical to the Python struct format.
*/
struct iec_record {
__u8 kind; /* IEC_KIND_* */
__u8 value; /* the byte, or an IEC_EV_* code */
__u8 flags; /* IEC_FLAG_* bitmask */
__u8 _pad; /* keep struct 4-byte aligned */
__u64 ts_ns; /* ktime_get_ns() at receive time */
} __packed;
/* ioctl interface */
#define IEC_IOC_MAGIC 'I'
/* set/get our primary device address (default 4 = printer) */
#define IEC_IOC_SET_ADDRESS _IOW(IEC_IOC_MAGIC, 1, __u32)
#define IEC_IOC_GET_ADDRESS _IOR(IEC_IOC_MAGIC, 2, __u32)
/*
* Selftest (Phase 1, PLAN.md §10): assert DATA (drive low), release to Hi-Z and
* read it back, then read ATN/CLK/RESET. Returns a bitmask of line states so
* the wiring can be checked with the C64 disconnected.
*/
#define IEC_IOC_SELFTEST _IOR(IEC_IOC_MAGIC, 3, __u32)
/* selftest result bits */
#define IEC_SELFTEST_DATA_ASSERT_OK 0x01 /* DATA read low while driven low */
#define IEC_SELFTEST_DATA_FLOAT_OK 0x02 /* DATA read high after release (Hi-Z) */
#define IEC_SELFTEST_ATN_RELEASED 0x04 /* ATN currently released (high) */
#define IEC_SELFTEST_CLK_RELEASED 0x08 /* CLK currently released (high) */
#define IEC_SELFTEST_RESET_RELEASED 0x10 /* RESET currently released (high) */
#endif /* _IEC_LISTENER_H */

42
kernel/iec_timing.h Normal file
View File

@ -0,0 +1,42 @@
/* SPDX-License-Identifier: GPL-2.0 */
/*
* iec_timing.h - starting timing constants for the listener handshake.
*
* Cross-referenced from the IEC spec, sd2iec empirical values, and ninepin /
* raspbiec observed values. See the "Starting Constants Table" and §1.7 of
* _research/pi-kernel-module-rt-gpio-2026-06-18.md. Tune in Phase 2 against a
* real C64 / logic analyser.
*/
#ifndef _IEC_TIMING_H
#define _IEC_TIMING_H
/* Per-CLK-transition busy-poll timeout. Generous; abort the byte and report a
* frame error if a CLK edge does not arrive within this window. Bounds the
* IRQ-off interval at ~8 ms worst case (8 bits), ~200 us in the normal case. */
#define IEC_CLK_TIMEOUT_US 1000
/* EOI detection: after releasing DATA (ready-for-data), if CLK does not assert
* within this window the talker is signalling EOI. Spec >= 200 us; sd2iec uses
* 256 us. We use 250 us (+25% margin over spec). */
#define IEC_EOI_DETECT_US 250
/* EOI acknowledge hold: pull DATA low for this long to ack EOI. Spec (Tei)
* >= 60 us; sd2iec uses 73 us (AVR instruction-calibrated). Pi A53 is faster;
* 80 us gives margin. */
#define IEC_EOI_ACK_HOLD_US 80
/* Debounce between consecutive bus reads (mirrors sd2iec / 1571 ROM). */
#define IEC_DEBOUNCE_US 2
/* Between-bytes minimum hold before the talker releases CLK for the next byte
* (Tbb). We simply wait for the next CLK transition; this is documentary. */
#define IEC_BETWEEN_BYTES_US 100
/*
* ATN response (Tat <= 1000 us) is NOT a udelay: it is performed by the ATN
* falling-edge ISR pulling DATA low immediately (hardirq latency ~5-20 us on
* the A53). Byte acknowledge (Tf <= 1000 us) is likewise immediate -- a single
* GPIO write at the end of the IRQ-off bit loop.
*/
#endif /* _IEC_TIMING_H */

25
pyproject.toml Normal file
View File

@ -0,0 +1,25 @@
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[project]
name = "iecpoc"
version = "0.1.0"
description = "Userspace decoder/trace for the Commodore IEC listener PoC (Raspberry Pi Zero 2 W)"
readme = "README.md"
requires-python = ">=3.10"
authors = [{ name = "Christian Werner" }]
dependencies = []
[project.optional-dependencies]
dev = ["pytest>=7"]
[project.scripts]
iecpoc = "iecpoc.main:main"
[tool.setuptools]
packages = ["iecpoc"]
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q"

0
tests/__init__.py Normal file
View File

Binary file not shown.

118
tests/fixtures.py Normal file
View File

@ -0,0 +1,118 @@
"""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())

116
tests/test_decode.py Normal file
View File

@ -0,0 +1,116 @@
"""Command decoding, PETSCII glyphs, and data-byte rendering (PLAN.md Phase 0)."""
from __future__ import annotations
import pytest
from iecpoc import decode, device
from iecpoc.petscii import CAT_CONTROL, CAT_GRAPHIC, CAT_PRINTABLE, to_glyph
# --- command decoding -------------------------------------------------------
@pytest.mark.parametrize(
"value, mnemonic, primary, secondary",
[
(0x24, decode.LISTEN, 4, None), # LISTEN 4
(0x20, decode.LISTEN, 0, None), # LISTEN 0
(0x3E, decode.LISTEN, 30, None), # LISTEN 30 (top of range)
(0x3F, decode.UNLISTEN, None, None),
(0x48, decode.TALK, 8, None), # TALK 8
(0x5F, decode.UNTALK, None, None),
(0x60, decode.SECOND, None, 0), # DATA/reopen SA=0
(0x6F, decode.SECOND, None, 15),
(0xE0, decode.CLOSE, None, 0),
(0xEF, decode.CLOSE, None, 15),
(0xF0, decode.OPEN, None, 0),
(0xFF, decode.OPEN, None, 15),
],
)
def test_decode_command(value, mnemonic, primary, secondary):
cmd = decode.decode_command(value)
assert cmd.value == value
assert cmd.mnemonic == mnemonic
assert cmd.primary == primary
assert cmd.secondary == secondary
def test_targets_address():
assert decode.decode_command(0x24).targets(4) is True
assert decode.decode_command(0x24).targets(8) is False
# UNLISTEN has no primary address, targets nobody specifically
assert decode.decode_command(0x3F).targets(4) is False
def test_describe():
assert decode.decode_command(0x24).describe() == "LISTEN 4"
assert decode.decode_command(0x3F).describe() == "UNLISTEN"
assert decode.decode_command(0xF0).describe() == "OPEN SA=0"
assert decode.decode_command(0xE0).describe() == "CLOSE SA=0"
def test_decode_command_range_check():
with pytest.raises(ValueError):
decode.decode_command(0x100)
# --- PETSCII ----------------------------------------------------------------
def test_petscii_letters_and_digits():
for ch in "HELLO WORLD0123456789":
g = to_glyph(ord(ch))
assert g.category == CAT_PRINTABLE
assert g.text == ch
def test_petscii_commodore_glyphs():
assert to_glyph(0x5C).text == "£"
assert to_glyph(0x5E).text == ""
assert to_glyph(0x5F).text == ""
def test_petscii_control_codes():
assert to_glyph(0x0D) == to_glyph(0x0D)
cr = to_glyph(0x0D)
assert cr.category == CAT_CONTROL
assert cr.text == "<CR>"
assert to_glyph(0x12).text == "<RVS-ON>"
# unknown control code falls back to the hex form
assert to_glyph(0x07).category == CAT_CONTROL
assert to_glyph(0x07).text == "<$07>"
def test_petscii_graphics_render_as_dot():
g = to_glyph(0x70) # graphics range in the upper-case set
assert g.category == CAT_GRAPHIC
assert g.text == "."
assert not g.printable
def test_petscii_range_check():
with pytest.raises(ValueError):
to_glyph(256)
# --- data-byte rendering ----------------------------------------------------
def test_describe_data_printable():
assert decode.describe_data(0x48) == "$48 'H'"
def test_describe_data_control():
assert decode.describe_data(0x0D) == "$0D <CR>"
# --- event description ------------------------------------------------------
def test_describe_event_atn_released_addressed():
rec = device.IecRecord(device.KIND_EVENT, device.EV_ATN_RELEASED,
device.FLAG_ADDRESSED, 0)
assert decode.describe_event(rec) == "released -> LISTENER"
def test_describe_event_atn_released_not_addressed():
rec = device.IecRecord(device.KIND_EVENT, device.EV_ATN_RELEASED, 0, 0)
assert decode.describe_event(rec) == "released -> not addressed -> IDLE"
def test_describe_event_idle():
rec = device.IecRecord(device.KIND_EVENT, device.EV_IDLE, 0, 0)
assert decode.describe_event(rec) == "waiting for ATN"

103
tests/test_device.py Normal file
View File

@ -0,0 +1,103 @@
"""Replay a captured /dev/iec0 record stream and check the produced trace.
End-to-end Phase 0 test: build the canonical session, round-trip it through the
binary wire format, and assert the formatted trace matches the documented PoC
output (PLAN.md §8).
"""
from __future__ import annotations
import io
from iecpoc import device
from iecpoc.log import TraceFormatter
from . import fixtures
def _trace(records, my_address=4, raw=False):
return list(TraceFormatter(my_address=my_address, raw=raw).format_stream(records))
# --- wire-format round trip -------------------------------------------------
def test_record_roundtrip():
rec = device.IecRecord(device.KIND_DATA, 0x48, device.FLAG_EOI, 123456789)
assert device.IecRecord.from_bytes(rec.to_bytes()) == rec
def test_iter_records_roundtrip():
records = fixtures.build_session()
buf = io.BytesIO()
device.write_records(buf, records)
buf.seek(0)
assert list(device.iter_records(buf)) == records
def test_iter_records_rejects_truncated():
buf = io.BytesIO(b"\x00\x01\x02") # < RECORD_SIZE
import pytest
with pytest.raises(ValueError):
list(device.iter_records(buf))
# --- captured binary fixture matches the in-memory builder ------------------
def test_capture_file_matches_builder():
with open(fixtures.CAPTURE_PATH, "rb") as fh:
from_file = list(device.iter_records(fh))
assert from_file == fixtures.build_session()
# --- trace formatting -------------------------------------------------------
def test_trace_contains_plan_section8_lines():
trace = _trace(fixtures.build_session())
# the §8 command-phase lines
assert "[IDLE] waiting for ATN" in trace
assert "[ATN] asserted -> DATA low (ack)" in trace
assert any(
line.startswith("[CMD] $24 LISTEN 4") and line.endswith("(addressed: ME)")
for line in trace
)
assert "[CMD] $F0 OPEN SA=0" in trace
assert "[ATN] released -> LISTENER" in trace
assert "[ATN] asserted -> command phase" in trace
assert "[CMD] $3F UNLISTEN" in trace
assert "[CMD] $E0 CLOSE SA=0" in trace
assert "[ATN] released -> not addressed -> IDLE" in trace
def test_trace_data_bytes_and_eoi():
trace = _trace(fixtures.build_session())
# HELLO WORLD bytes appear as glyphs
assert "[DATA] $48 'H'" in trace
assert "[DATA] $4F 'O'" in trace
assert "[DATA] $20 ' '" in trace
# each PRINT# record ends with a CR carrying EOI
assert "[DATA] $0D <CR> <EOI>" in trace
# exactly two EOI markers (two PRINT# statements)
assert sum("<EOI>" in line for line in trace) == 2
def test_trace_full_text_reconstructable():
"""The data bytes, in order, reconstruct the two printed lines."""
data_bytes = [r.value for r in fixtures.build_session() if r.is_data]
text = bytes(data_bytes).decode("ascii")
assert text == "HELLO WORLD\rLINE TWO\r"
def test_command_not_for_us_has_no_me_annotation():
# if we are device 8, LISTEN 4 is not for us
trace = _trace(fixtures.build_session(), my_address=8)
assert "[CMD] $24 LISTEN 4" in trace
assert not any("(addressed: ME)" in line for line in trace)
def test_raw_mode_is_hex_only():
trace = _trace(fixtures.build_session(), raw=True)
# no bracketed tags in raw mode
assert all(not line.startswith("[") for line in trace)
# first emitted byte is the LISTEN 4 command byte $24
assert trace[0] == "24"
# data bytes present as hex
assert "48" in trace # 'H'