mirror of
https://github.com/retroSoC/retroSoC.git
synced 2026-11-05 06:32:37 +00:00
feat(piolite): implement P1 ISA reference model and assembler
Add the frozen ISA1.0 host tooling for TINY: checked 16-bit encode/decode with relocation and atomic 32-slot allocation/sharing (scripts/piolite_isa.py), a two-SM PCLK reference model with DIV/delay, wrap/HALT, lifecycle, shift/autopull/autopush, pre-edge FIFOs and latched faults (scripts/piolite_model.py), and a strict .piolite assembler with deterministic C objects and atomic output (scripts/piolite_asm.py). Ten handwritten application programs under app/asm/piolite/ prove real 32-word footprints: pulse generate+capture 20/32, UART TX+RX 24/32, IR TX+RX 23/32. No RTL, HAL, profile, dependency or public-ABI changes; no protocol/rate qualification is claimed. Validation: 176 focused tests including exhaustive 65536-word ISA checking (22257 legal/43279 rejected), full 16-delay matrix, lifecycle/publication/authorization coverage and independent application behavior tests; full Pytest 2300 passed, 1 skipped; ruff, C11 syntax checks and byte-identical rebuilds pass. Seven independent review rounds closed twelve defects and three coverage gaps, all re-verified by independent probes; findings and closure evidence are recorded in docs/ip/piolite-verification.md with phase_accepted=false.
This commit is contained in:
19 files changed
+3912
-2
No files matched your search
@@ -0,0 +1,24 @@
|
||||
; Latch once, then clock eight data samples into one automatic RX word.
|
||||
.program gamepad
|
||||
.entry scan
|
||||
.wrap_bottom scan
|
||||
.wrap_top again
|
||||
.in 22, 1
|
||||
.out 0, 0
|
||||
.set 20, 2
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 8, right, auto
|
||||
.test none, 0
|
||||
.packing 0, 1, 8
|
||||
.clock_ticks 4, stretchable
|
||||
scan: SET_PIN 2
|
||||
SET_PIN 0
|
||||
IN PINS, 1
|
||||
SET_PIN 1
|
||||
SET_PIN 0
|
||||
SET_COUNT X, 7
|
||||
bit: IN PINS, 1
|
||||
SET_PIN 1
|
||||
SET_PIN 0
|
||||
BRANCH DEC_X_NONZERO, bit
|
||||
again: BRANCH ALWAYS, scan
|
||||
@@ -0,0 +1,26 @@
|
||||
; Emit complemented high and low countdowns for software pulse framing.
|
||||
.program ir_rx
|
||||
.entry wait_high
|
||||
.wrap_bottom wait_high
|
||||
.wrap_top again
|
||||
.in 13, 1
|
||||
.out 0, 0
|
||||
.set 0, 0
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test 13, 0
|
||||
.packing 0, 0, 16
|
||||
.clock_ticks 2, stretchable
|
||||
wait_high: WAIT PIN, 1, 13
|
||||
MOVE X, ONES
|
||||
high: BRANCH TEST_PIN_HIGH, high_count
|
||||
BRANCH ALWAYS, high_done
|
||||
high_count: BRANCH DEC_X_NONZERO, high
|
||||
high_done: MOVE ISR, X
|
||||
FIFO PUSH, BLOCKING
|
||||
MOVE X, ONES
|
||||
low: BRANCH TEST_PIN_HIGH, low_done
|
||||
low_count: BRANCH DEC_X_NONZERO, low
|
||||
low_done: MOVE ISR, X
|
||||
FIFO PUSH, BLOCKING
|
||||
again: BRANCH ALWAYS, wait_high
|
||||
@@ -0,0 +1,22 @@
|
||||
.program ir_tx
|
||||
.entry load_mark
|
||||
.wrap_bottom load_mark
|
||||
.wrap_top again
|
||||
.in 0, 0
|
||||
.out 0, 0
|
||||
.set 12, 1
|
||||
.tx_bits 16, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test none, 0
|
||||
.packing 2, 0, 16
|
||||
.clock_ticks 3, stretchable
|
||||
load_mark: FIFO PULL, BLOCKING
|
||||
MOVE X, OSR
|
||||
load_space: FIFO PULL, BLOCKING
|
||||
MOVE Y, OSR
|
||||
mark: SET_PIN 1
|
||||
SET_PIN 0
|
||||
BRANCH DEC_X_NONZERO, mark
|
||||
space: SET_PIN 0
|
||||
BRANCH DEC_Y_NONZERO, space
|
||||
again: BRANCH ALWAYS, load_mark
|
||||
@@ -0,0 +1,27 @@
|
||||
; Blank before every blocking refill, then shift 16 column/row bits.
|
||||
.program led_scan
|
||||
.entry scan
|
||||
.wrap_bottom scan
|
||||
.wrap_top again
|
||||
.in 0, 0
|
||||
.out 19, 1
|
||||
.set 16, 3
|
||||
.tx_bits 16, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test none, 0
|
||||
.packing 1, 0, 16
|
||||
.clock_ticks 4, stretchable
|
||||
scan: SET_PIN 4
|
||||
FIFO PULL, BLOCKING
|
||||
OUT PINS, 1
|
||||
SET_PIN 1
|
||||
SET_PIN 0
|
||||
SET_COUNT X, 15
|
||||
bit: OUT PINS, 1
|
||||
SET_PIN 1
|
||||
SET_PIN 0
|
||||
BRANCH DEC_X_NONZERO, bit
|
||||
SET_PIN 6
|
||||
SET_PIN 0
|
||||
CTRL NOP [7]
|
||||
again: BRANCH ALWAYS, scan
|
||||
@@ -0,0 +1,22 @@
|
||||
; One initial sample plus 32767 two-tick loop samples = 32768 samples.
|
||||
.program logic_capture
|
||||
.entry initialize
|
||||
.wrap_bottom initialize
|
||||
.wrap_top done
|
||||
.in 8, 8
|
||||
.out 0, 0
|
||||
.set 0, 0
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 32, right, auto
|
||||
.test none, 0
|
||||
.packing 0, 32768, 8
|
||||
.clock_ticks 2, stretchable
|
||||
initialize: MOVE X, ONES
|
||||
MOVE OSR, X
|
||||
OUT DISCARD, 1
|
||||
MOVE X, OSR
|
||||
IN PINS, 8
|
||||
CTRL NOP
|
||||
loop: IN PINS, 8
|
||||
BRANCH DEC_X_NONZERO, loop
|
||||
done: CTRL HALT
|
||||
@@ -0,0 +1,21 @@
|
||||
; Emit 65535 minus the number of high-phase countdown passes.
|
||||
.program pulse_capture
|
||||
.entry wait_high
|
||||
.wrap_bottom wait_high
|
||||
.wrap_top again
|
||||
.in 3, 1
|
||||
.out 0, 0
|
||||
.set 0, 0
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test 3, 0
|
||||
.packing 0, 0, 16
|
||||
.clock_ticks 2, stretchable
|
||||
wait_high: WAIT PIN, 1, 3
|
||||
MOVE X, ONES
|
||||
high: BRANCH TEST_PIN_HIGH, count
|
||||
BRANCH ALWAYS, complete
|
||||
count: BRANCH DEC_X_NONZERO, high
|
||||
complete: MOVE ISR, X
|
||||
FIFO PUSH, BLOCKING
|
||||
again: BRANCH ALWAYS, wait_high
|
||||
@@ -0,0 +1,25 @@
|
||||
; Two 16-bit FIFO words program one finite HIGH then LOW pulse in ticks.
|
||||
.program pulse_generate
|
||||
.entry load_high
|
||||
.wrap_bottom load_high
|
||||
.wrap_top halt
|
||||
.in 0, 0
|
||||
.out 0, 0
|
||||
.set 2, 1
|
||||
.tx_bits 16, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test none, 0
|
||||
.packing 2, 0, 16
|
||||
.clock_ticks 1, stretchable
|
||||
load_high: FIFO PULL, BLOCKING
|
||||
MOVE X, OSR
|
||||
load_low: FIFO PULL, BLOCKING
|
||||
MOVE Y, OSR
|
||||
prepare_high: BRANCH DEC_X_NONZERO, prepare_low
|
||||
prepare_low: BRANCH DEC_Y_NONZERO, high
|
||||
high: SET_PIN 1
|
||||
high_wait: BRANCH DEC_X_NONZERO, high_wait
|
||||
low: SET_PIN 0
|
||||
low_wait: BRANCH DEC_Y_NONZERO, low_wait
|
||||
done: SET_PIN 1
|
||||
halt: CTRL HALT
|
||||
@@ -0,0 +1,28 @@
|
||||
; Mode-0 eight-bit full-duplex frame; CS is SET window bit 1.
|
||||
.program spi_master
|
||||
.entry idle
|
||||
.wrap_bottom idle
|
||||
.wrap_top again
|
||||
.in 7, 1
|
||||
.out 6, 1
|
||||
.set 4, 2
|
||||
.tx_bits 8, right, manual
|
||||
.rx_bits 8, right, auto
|
||||
.test none, 0
|
||||
.packing 1, 1, 8
|
||||
.clock_ticks 5, stretchable
|
||||
idle: SET_PIN 2
|
||||
frame: FIFO PULL, BLOCKING
|
||||
SET_PIN 0
|
||||
OUT PINS, 1
|
||||
SET_PIN 1
|
||||
IN PINS, 1
|
||||
SET_PIN 0
|
||||
SET_COUNT X, 7
|
||||
bit: OUT PINS, 1
|
||||
SET_PIN 1
|
||||
IN PINS, 1
|
||||
SET_PIN 0
|
||||
BRANCH DEC_X_NONZERO, bit
|
||||
SET_PIN 2
|
||||
again: BRANCH ALWAYS, frame
|
||||
@@ -0,0 +1,24 @@
|
||||
.program uart_rx
|
||||
.entry wait_start
|
||||
.wrap_bottom wait_start
|
||||
.wrap_top again
|
||||
.in 1, 1
|
||||
.out 0, 0
|
||||
.set 0, 0
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 8, right, auto
|
||||
.test none, 0
|
||||
.packing 0, 1, 8
|
||||
.clock_ticks 8, stretchable
|
||||
wait_start: WAIT PIN, 0, 1
|
||||
CTRL NOP [10]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
IN PINS, 1 [7]
|
||||
WAIT PIN, 1, 1
|
||||
again: BRANCH ALWAYS, wait_start
|
||||
@@ -0,0 +1,24 @@
|
||||
.program uart_tx
|
||||
.entry idle
|
||||
.wrap_bottom idle
|
||||
.wrap_top again
|
||||
.in 0, 0
|
||||
.out 0, 1
|
||||
.set 0, 1
|
||||
.tx_bits 8, right, auto
|
||||
.rx_bits 32, right, manual
|
||||
.test none, 0
|
||||
.packing 1, 0, 8
|
||||
.clock_ticks 8, stretchable
|
||||
idle: SET_PIN 1 [7]
|
||||
start: SET_PIN 0 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
OUT PINS, 1 [7]
|
||||
SET_PIN 1 [6]
|
||||
again: BRANCH ALWAYS, start
|
||||
@@ -133,12 +133,234 @@ verdict; UART output is diagnostic. Run both Verilator and Icarus pin-level
|
||||
checks; preserve simulator-specific unsupported coverage rather than replacing
|
||||
one with the other. Neither a dry-run nor a skipped fixture is a passing test.
|
||||
|
||||
## PIOLITE-P1 Host Implementation Evidence
|
||||
|
||||
The 2026-10-28 working-tree implementation based on source revision
|
||||
`557f8b452745ed2cc28a73f08a452804f0a3cc74` adds the host-only ISA, model,
|
||||
assembler, handwritten programs, and focused tests. It has
|
||||
`phase_accepted: false`: this evidence is not phase acceptance, Tiny
|
||||
integration, protocol-waveform/rate qualification, or a firmware/RTL result.
|
||||
The Tiny verification ledger is intentionally unchanged at this stage.
|
||||
|
||||
The focused commands and observed results are:
|
||||
|
||||
```sh
|
||||
python3 -m pytest -q tests/test_piolite_isa.py tests/test_piolite_model.py tests/test_piolite_assembler.py
|
||||
python3 -m pytest -q
|
||||
ruff check .
|
||||
git diff --check
|
||||
for object in build/piolite-p1-2026-10-28/generated/piolite/*.c; do
|
||||
cc -std=c11 -fsyntax-only "$object"
|
||||
done
|
||||
```
|
||||
|
||||
The focused Pytest run passed 176 tests: 23 ISA, 99 model/application, and 54
|
||||
assembler/allocation/footprint tests. The full repository run passed 2300 tests
|
||||
with one skip in 4767.84 seconds; the skip is the existing optional `pypdf`
|
||||
import in `test_publication_retrieval.py`, not a PIOLite case. Ruff and C11
|
||||
syntax checks passed. The
|
||||
ISA test independently enumerates all 65536 encodings: 22257 legal forms
|
||||
round-trip and 43279 illegal forms are rejected. Model traces cover the frozen
|
||||
PCLK/divider/delay, dual-SM, lifecycle, counter, wrap, fault, shift, packing,
|
||||
and pre-edge FIFO behavior; they do not use the model as its own oracle.
|
||||
|
||||
### Independent application-source review closure
|
||||
|
||||
The independent review cleared the ISA/model/assembler/allocation and identified
|
||||
three application-source defects. Their host behavioral closures are:
|
||||
|
||||
- **P1-1 UART RX schedule.** The old four-tick post-WAIT path sampled the start
|
||||
bit and then D0..D6. `uart_rx.piolite` now waits ten delay ticks after the NOP
|
||||
commit. With its declared eight ticks/bit, independent 8N1 stimulus verifies
|
||||
data samples at edges 12, 20, 28, 36, 44, 52, 60, and 68, followed by the STOP
|
||||
WAIT at 76. `test_uart_rx_application_samples_independent_8n1_stimulus`
|
||||
verifies exact received bytes for A5, 00, and FF, not only a metadata claim.
|
||||
- **P1-2 programmable pulse phases.** The old program copied one OSR value to
|
||||
both counters and drove LOW inside the HIGH loop, producing one-tick pulses.
|
||||
`pulse_generate.piolite` now consumes separate 16-bit HIGH and LOW duration
|
||||
words, pre-decrements both counts, holds each output level throughout its
|
||||
countdown, returns HIGH, and HALTs.
|
||||
`test_pulse_generate_application_has_independent_exact_phase_durations`
|
||||
checks every output transition for HIGH/LOW settings 3/5 and 8/2 ticks.
|
||||
- **P1-3 uniform logic capture.** The old paired-IN loop alternated one- and
|
||||
two-tick sample intervals. `logic_capture.piolite` now derives 32767 loop
|
||||
iterations through OSR, adds one initial sample, and executes a strict
|
||||
IN/branch two-tick loop. `test_logic_capture_application_uniform_cadence_and_exact_counts`
|
||||
drives an independent changing GPIO8..15 pattern and checks all 32768 sample
|
||||
edges, all 8192 packed words, HALT, and empty residual ISR at DIV 1 and 3.
|
||||
|
||||
These seven new cases are deterministic host-model application tests with
|
||||
independent stimulus and expectations. They close the reviewed P1 source
|
||||
defects; they are not RTL pin-BFM, electrical, baud/rate, or P4 protocol
|
||||
qualification.
|
||||
|
||||
### FIX_P0_P1 review closure
|
||||
|
||||
A second independent review found two P1 defects and one P2 coverage group.
|
||||
Their closures are:
|
||||
|
||||
- **P1-1 logic-capture item metadata.** `.packing 0, 8192, 32` described FIFO
|
||||
words rather than application items. The source now declares
|
||||
`.packing 0, 32768, 8` with `.clock_ticks 2, stretchable` per 8-bit sample.
|
||||
`test_logic_capture_metadata_describes_application_samples` checks the
|
||||
program and emitted C metadata. The behavioral capture test derives its
|
||||
expected word count as `32768*8/32 = 8192` and its four-lane layout from the
|
||||
item width rather than hardcoding either one.
|
||||
- **P1-2 selected-but-unused test pin.** Selection previously enabled the pin
|
||||
even without a test-pin branch, while claim metadata followed actual use.
|
||||
Rule **(a)** is selected: `test_pin_is_used` infers enable only from an actual
|
||||
`TEST_PIN_HIGH` branch. This follows the register contract's "GPIO checked
|
||||
only when enabled/used" qualification and referenced-resource metadata rule;
|
||||
an unused selection does not invent a claim. The C emitter and default
|
||||
`SMConfig.from_allocation` share that query. Two
|
||||
`test_selected_test_pin_enable_follows_actual_branch_use` cases run the full
|
||||
assemble/C/default-config/START path: selected-unused starts with enable 0
|
||||
and claim 0; selected-used starts with enable 1 and GPIO5 claimed, then takes
|
||||
the false-condition path to HALT. START authorization checks are unchanged.
|
||||
- **P2 confirmed-behavior probes.** Eight permanent model cases now cover a
|
||||
false-condition out-of-region branch; START TX preservation, nonempty-RX and
|
||||
partial-IMEM rejection, sticky history and initial XY/data/OE; automatic PUSH
|
||||
clearing last-FIFO-failed; PULL high-bit zeroing; PAUSE acceptance during a
|
||||
stalled WAIT at DIV 65535; and runtime IN/OUT window-authorization faults.
|
||||
These tests codify review-confirmed behavior without changing the model.
|
||||
|
||||
These eleven new tests join the prior behavioral coverage. The second review's
|
||||
enable/used wording allowed either consistent rule; rule (a) is an explicit
|
||||
judgment call, not an unresolved specification conflict or a weakened check.
|
||||
|
||||
### Fourth independent review closure
|
||||
|
||||
The fourth review found one allocator defect and one assembler-diagnostics
|
||||
defect. Five regressions close them:
|
||||
|
||||
- **P1 foreign allocation handles.** `unload()` previously trusted the caller's
|
||||
base/length after checking only the numeric handle, so equal handles in two
|
||||
allocators could remove the wrong live record while retaining stale bitmap
|
||||
bits. It now resolves the handle to this allocator's canonical record and
|
||||
requires the exact canonical allocation object before changing references or
|
||||
bitmap state; all clearing uses the canonical base/length.
|
||||
`test_foreign_allocation_handle_collision_is_rejected_atomically` uses two
|
||||
allocators with colliding handle 1 and checks unchanged bitmap, live
|
||||
references and allocations, followed by successful legitimate unload/reload.
|
||||
- **P2 pathological numeric diagnostics.** A 5000-digit decimal can raise the
|
||||
interpreter integer-conversion-limit `ValueError` before the normal range
|
||||
check. `_number()` now converts inside a checked exception boundary and
|
||||
raises `AssemblyError` with its original source line. Two library and two CLI
|
||||
cases cover 5000-digit decimal and equivalent-length hexadecimal tokens;
|
||||
every case returns a normal `line 13: SET immediate ...` diagnostic, CLI
|
||||
status 2, no traceback and no generated output file.
|
||||
|
||||
No application source or footprint changed in this closure, so the program
|
||||
hashes and allocation evidence remain valid; only the validation logs and this
|
||||
ledger require refresh.
|
||||
|
||||
### Fifth independent review closure
|
||||
|
||||
The fifth review, including 140k-edge randomized differential validation, found
|
||||
three P1 defects. Six directed regressions close them:
|
||||
|
||||
- **P1-1 narrow OUT pin preservation.** OUT PINS/DIRECTIONS used the whole
|
||||
window mask when committing a narrow chunk. The model now forms the update
|
||||
mask from the emitted count and window base while retaining whole-window
|
||||
authorization. Two `test_narrow_out_preserves_unrelated_output_bits` cases
|
||||
use a four-bit window with nonzero initial upper bits; each destination
|
||||
produces D0 for the changed field and preserves C0 in the other field.
|
||||
- **P1-2 IMEM publication lifecycle.** Publication now requires every SM to be
|
||||
DISABLED and rejects already-valid overlapping slots before changing IMEM.
|
||||
Scoped host-model `unpublish()` clears exactly an unloaded image's validity
|
||||
region; it is the model counterpart of program unload, not a new HAL ABI.
|
||||
Replacement requires invalidation before publication, so a two-word image
|
||||
replaced by one word leaves the vacated slot invalid. Tests prove live
|
||||
publish/unpublish rejection is atomic, overlap rejection is atomic, unrelated
|
||||
valid words survive, vacated APB reads return zero, and forced fetch of the
|
||||
vacated slot raises INVALID_FETCH.
|
||||
- **P1-3 atomic CLI output.** Direct destination writes could expose truncated
|
||||
C files after an output error. The CLI now writes a same-directory temporary
|
||||
file, closes it successfully, then uses `os.replace`; every output-failure
|
||||
path attempts temporary-file cleanup. Injected mid-write failures after 32
|
||||
bytes prove no destination appears for a new output and a pre-existing
|
||||
known-good destination remains unchanged, with no temporary files left.
|
||||
|
||||
No application program or pair footprint changed, so existing program hashes
|
||||
and allocation evidence remain valid; validation logs and this ledger are
|
||||
refreshed for the closure.
|
||||
|
||||
### Sixth independent review closure
|
||||
|
||||
The sixth review found three P1 defects and three P2 items. Their 32 new or
|
||||
expanded regression cases close them:
|
||||
|
||||
- **P1-1 guarded full invalidation.** `invalidate_imem()` now applies the same
|
||||
all-SMs-DISABLED check as scoped publication changes and clears publication
|
||||
identity only after that check succeeds. A running-SM test proves payload,
|
||||
validity and execution state remain unchanged after rejection.
|
||||
- **P1-2 authenticated unpublish and sharing.** The model now tracks exact
|
||||
published allocation objects and publication reference counts independently
|
||||
of allocator-local numeric handles. Unknown, foreign, shortened and
|
||||
same-length/different-content images are rejected before validity changes.
|
||||
Shared publication keeps its region valid until the final reference is
|
||||
removed. Full invalidation clears this identity state.
|
||||
- **P1-3 disabled SET windows.** SET_PIN and SET_DIR now raise
|
||||
PIN_AUTHORIZATION before effects when SET width is zero, matching assembler
|
||||
rejection. Directed tests cover both destinations and zero retirement.
|
||||
- **P2-1 bounded temporary names.** The CLI uses the fixed `.piolite-`
|
||||
temporary prefix rather than the full destination basename. Outputs with 242-
|
||||
and 255-character basenames assemble successfully without temporary residue.
|
||||
- **P2-2 destination modes.** Before replacement, the temporary file receives
|
||||
the existing destination's mode or `0666 & ~umask` for a new destination.
|
||||
Tests verify a new 0644 file under umask 022 and preservation of existing
|
||||
0644 and 0444 modes while replacing content atomically.
|
||||
- **P2-3 confirmed behavior matrix.** Delay tests now cover every value 0..15;
|
||||
further cases cover START of one DISABLED SM while the other RUNNING,
|
||||
unclaimed/non-output initial outputs, WAIT FLAG pre-edge/non-clearing at all
|
||||
four indices, and a delay-15 taken branch winning over wrap. These probes
|
||||
assert confirmed behavior without changing it.
|
||||
|
||||
No application source or pair footprint changed. Existing program hashes and
|
||||
allocation evidence remain valid; validation logs and this ledger are refreshed.
|
||||
|
||||
Every application was assembled twice and each pair of C outputs compared
|
||||
byte-for-byte:
|
||||
|
||||
```sh
|
||||
out=build/piolite-p1-2026-10-28
|
||||
for source in app/asm/piolite/*.piolite; do
|
||||
name=${source##*/}
|
||||
stem=${name%.piolite}
|
||||
python3 scripts/piolite_asm.py "$source" "$out/generated/piolite/$stem.c"
|
||||
python3 scripts/piolite_asm.py "$source" "$out/rebuild/piolite/$stem.c"
|
||||
cmp "$out/generated/piolite/$stem.c" "$out/rebuild/piolite/$stem.c"
|
||||
done
|
||||
```
|
||||
|
||||
All ten comparisons passed. `build/piolite-p1-2026-10-28/assembly.log` records
|
||||
the assembler output; `footprints.json` binds source/C hashes, metadata,
|
||||
relocation sites, pair allocations, and `phase_accepted: false`.
|
||||
|
||||
| Program | Words | Required pair | Relocated total |
|
||||
| --- | ---: | --- | ---: |
|
||||
| `logic_capture` | 9 | — | — |
|
||||
| `pulse_generate` | 12 | `pulse_generate+pulse_capture` | 20 |
|
||||
| `pulse_capture` | 8 | `pulse_generate+pulse_capture` | 20 |
|
||||
| `uart_tx` | 12 | `uart_tx+uart_rx` | 24 |
|
||||
| `uart_rx` | 12 | `uart_tx+uart_rx` | 24 |
|
||||
| `spi_master` | 15 | — | — |
|
||||
| `ir_tx` | 10 | `ir_tx+ir_rx` | 23 |
|
||||
| `ir_rx` | 13 | `ir_tx+ir_rx` | 23 |
|
||||
| `led_scan` | 14 | — | — |
|
||||
| `gamepad` | 11 | — | — |
|
||||
|
||||
The pulse pair occupies slots 0..19, UART duplex slots 0..23, and the IR pair
|
||||
slots 0..22 in their respective independent allocations. No IMEM enlargement,
|
||||
padding demonstration, protocol claim, or simultaneous all-application claim
|
||||
is made. P2-P5 requirements remain open as listed below.
|
||||
|
||||
## Freeze Record and Remaining Gaps
|
||||
|
||||
| Phase | Status at this freeze |
|
||||
| Phase | Status |
|
||||
| --- | --- |
|
||||
| PIOLITE-P0 | Documentation freeze complete; local links/anchor, command dry-runs, unchanged legacy/R2 IDs and QFN64/ALT tables, documentation-only scope and whitespace checks passed on 2026-10-04 |
|
||||
| PIOLITE-P1 | Not started: model, assembler and actual 32-word program evidence absent |
|
||||
| PIOLITE-P1 | Host model, assembler, focused tests and actual 32-word footprints added on 2026-10-28; `phase_accepted: false`; Tiny integration and protocol qualification remain absent |
|
||||
| PIOLITE-P2 | Not started: core, CSR/ISA parity and standalone RTL/formal/synthesis evidence absent |
|
||||
| PIOLITE-P3 | Not started: R2-dependent Tiny wiring, DMA V2.1 and GPIO/RCU/HAL lifecycle absent |
|
||||
| PIOLITE-P4 | Not started: complete SDK/application and dual-simulator/contention qualification absent |
|
||||
|
||||
@@ -10,6 +10,18 @@ bit-accurate P4 primitive model. They accept only the APU instruction language
|
||||
and deliberately have no C, ELF, RV32, dynamic linking, or runtime
|
||||
code-generation path.
|
||||
|
||||
`piolite_isa.py`, `piolite_model.py`, and `piolite_asm.py` implement the frozen
|
||||
Tiny PIO-lite ISA1.0 encoding/allocation objects, two-SM PCLK reference model,
|
||||
and strict deterministic assembler/C-object emitter. They are host-only P1
|
||||
verification tools: they do not integrate a peripheral, provide a HAL ABI, or
|
||||
qualify protocol waveforms or rates. Assemble a handwritten source into a
|
||||
selected build variant with:
|
||||
|
||||
```sh
|
||||
python3 scripts/piolite_asm.py app/asm/piolite/uart_tx.piolite \
|
||||
build/<variant>/generated/piolite/uart_tx.c
|
||||
```
|
||||
|
||||
`setup_apu_reference.py` installs the checksum-pinned libFLAC and official
|
||||
FLAC corpus as host-only P5 verification inputs and builds libFLAC with Ogg
|
||||
disabled below the selected build variant. These inputs never enter firmware,
|
||||
|
||||
@@ -0,0 +1,503 @@
|
||||
"""Strict deterministic assembler and C-object emitter for PIO-lite ISA1.0."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import stat
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from piolite_isa import (
|
||||
BRANCH_CONDITIONS,
|
||||
CTRL_COMMANDS,
|
||||
FIFO_MODES,
|
||||
FIFO_OPERATIONS,
|
||||
FLAG_ACTIONS,
|
||||
IN_SOURCES,
|
||||
INSTRUCTION_WORDS,
|
||||
ISA_VERSION,
|
||||
MOVE_REGISTERS,
|
||||
MOVE_SOURCES,
|
||||
OUT_DESTINATIONS,
|
||||
WAIT_SOURCES,
|
||||
Instruction,
|
||||
ISAError,
|
||||
Opcode,
|
||||
PackingSpec,
|
||||
PinWindow,
|
||||
Program,
|
||||
ShiftSpec,
|
||||
TestSpec,
|
||||
TimingSpec,
|
||||
referenced_resources,
|
||||
test_pin_is_used,
|
||||
)
|
||||
|
||||
|
||||
IDENTIFIER = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z", re.ASCII)
|
||||
NUMBER = re.compile(r"(?:0x[0-9a-fA-F]+|[0-9]+)\Z", re.ASCII)
|
||||
DIRECTIVES = (
|
||||
".program",
|
||||
".entry",
|
||||
".wrap_bottom",
|
||||
".wrap_top",
|
||||
".in",
|
||||
".out",
|
||||
".set",
|
||||
".tx_bits",
|
||||
".rx_bits",
|
||||
".test",
|
||||
".packing",
|
||||
".clock_ticks",
|
||||
)
|
||||
DIRECTIVE_OPERANDS = {
|
||||
".program": 1,
|
||||
".entry": 1,
|
||||
".wrap_bottom": 1,
|
||||
".wrap_top": 1,
|
||||
".in": 2,
|
||||
".out": 2,
|
||||
".set": 2,
|
||||
".tx_bits": 3,
|
||||
".rx_bits": 3,
|
||||
".test": 2,
|
||||
".packing": 3,
|
||||
".clock_ticks": 2,
|
||||
}
|
||||
INSTRUCTION_OPERANDS = {
|
||||
Opcode.CTRL: 1,
|
||||
Opcode.SET_PIN: 1,
|
||||
Opcode.SET_DIR: 1,
|
||||
Opcode.SET_COUNT: 2,
|
||||
Opcode.IN: 2,
|
||||
Opcode.OUT: 2,
|
||||
Opcode.MOVE: 2,
|
||||
Opcode.FIFO: 2,
|
||||
Opcode.WAIT: 3,
|
||||
Opcode.BRANCH: 2,
|
||||
Opcode.FLAG: 2,
|
||||
}
|
||||
MNEMONICS = {opcode.name: opcode for opcode in Opcode}
|
||||
|
||||
|
||||
class AssemblyError(ValueError):
|
||||
"""A source error with a deterministic one-based source line."""
|
||||
|
||||
def __init__(self, line: int, message: str):
|
||||
super().__init__(f"line {line}: {message}")
|
||||
self.line = line
|
||||
self.message = message
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Statement:
|
||||
line: int
|
||||
opcode: Opcode
|
||||
args: tuple[str, ...]
|
||||
delay: int
|
||||
|
||||
|
||||
def _number(token: str, line: int, name: str, minimum: int, maximum: int) -> int:
|
||||
if NUMBER.fullmatch(token) is None:
|
||||
raise AssemblyError(line, f"{name} must be unsigned decimal or 0x hexadecimal")
|
||||
try:
|
||||
value = int(token, 16) if token.startswith("0x") else int(token, 10)
|
||||
except ValueError as exc:
|
||||
raise AssemblyError(line, f"{name} exceeds the supported integer conversion") from exc
|
||||
if not minimum <= value <= maximum:
|
||||
raise AssemblyError(line, f"{name} must be in {minimum}..{maximum}")
|
||||
return value
|
||||
|
||||
|
||||
def _identifier(token: str, line: int, name: str) -> str:
|
||||
if IDENTIFIER.fullmatch(token) is None:
|
||||
raise AssemblyError(line, f"{name} must be an ASCII identifier")
|
||||
return token
|
||||
|
||||
|
||||
def _symbol(token: str, symbols: dict[str, int], line: int, name: str) -> int:
|
||||
if token not in symbols:
|
||||
raise AssemblyError(line, f"unknown {name} symbol {token!r}")
|
||||
return symbols[token]
|
||||
|
||||
|
||||
def _args(text: str, count: int, line: int, construct: str) -> tuple[str, ...]:
|
||||
if count == 1:
|
||||
if "," in text or not text or len(text.split()) != 1:
|
||||
raise AssemblyError(line, f"{construct} requires exactly one operand")
|
||||
return (text,)
|
||||
args = tuple(item.strip() for item in text.split(","))
|
||||
if len(args) != count or any(not item or len(item.split()) != 1 for item in args):
|
||||
raise AssemblyError(line, f"{construct} requires {count} comma-separated operands")
|
||||
return args
|
||||
|
||||
|
||||
def _delay(text: str, line: int) -> tuple[str, int]:
|
||||
if "[" not in text and "]" not in text:
|
||||
return text, 0
|
||||
match = re.fullmatch(r"(.*?)\s*\[([^\[\]]+)\]", text)
|
||||
if match is None or not match.group(1).strip():
|
||||
raise AssemblyError(line, "delay must be one trailing unsigned value in brackets")
|
||||
return match.group(1).strip(), _number(match.group(2).strip(), line, "delay", 0, 15)
|
||||
|
||||
|
||||
def _instruction(statement: _Statement, labels: dict[str, int]) -> Instruction:
|
||||
line = statement.line
|
||||
args = statement.args
|
||||
opcode = statement.opcode
|
||||
operand = 0
|
||||
if opcode == Opcode.CTRL:
|
||||
operand = _symbol(args[0], CTRL_COMMANDS, line, "CTRL") << 5
|
||||
elif opcode in (Opcode.SET_PIN, Opcode.SET_DIR):
|
||||
operand = _number(args[0], line, "SET immediate", 0, 0xFF)
|
||||
elif opcode == Opcode.SET_COUNT:
|
||||
destination = _symbol(args[0], {"X": 0, "Y": 1}, line, "SET_COUNT destination")
|
||||
operand = (destination << 7) | _number(args[1], line, "SET_COUNT immediate", 0, 127)
|
||||
elif opcode == Opcode.IN:
|
||||
source = _symbol(args[0], IN_SOURCES, line, "IN source")
|
||||
count = _number(args[1], line, "IN count", 1, 32)
|
||||
operand = (source << 5) | (count & 0x1F)
|
||||
elif opcode == Opcode.OUT:
|
||||
destination = _symbol(args[0], OUT_DESTINATIONS, line, "OUT destination")
|
||||
count = _number(args[1], line, "OUT count", 1, 32)
|
||||
operand = (destination << 5) | (count & 0x1F)
|
||||
elif opcode == Opcode.MOVE:
|
||||
destination = _symbol(args[0], MOVE_REGISTERS, line, "MOVE destination")
|
||||
source = _symbol(args[1], MOVE_SOURCES, line, "MOVE source")
|
||||
operand = (destination << 6) | (source << 3)
|
||||
elif opcode == Opcode.FIFO:
|
||||
operation = _symbol(args[0], FIFO_OPERATIONS, line, "FIFO operation")
|
||||
mode = _symbol(args[1], FIFO_MODES, line, "FIFO mode")
|
||||
operand = (operation << 7) | (mode << 6)
|
||||
elif opcode == Opcode.WAIT:
|
||||
source = _symbol(args[0], WAIT_SOURCES, line, "WAIT source")
|
||||
level = _number(args[1], line, "WAIT level", 0, 1)
|
||||
maximum_index = 31 if source == WAIT_SOURCES["PIN"] else 3
|
||||
index = _number(args[2], line, "WAIT index", 0, maximum_index)
|
||||
operand = (source << 7) | (level << 6) | (index << 1)
|
||||
elif opcode == Opcode.BRANCH:
|
||||
condition = _symbol(args[0], BRANCH_CONDITIONS, line, "branch condition")
|
||||
label = _identifier(args[1], line, "branch target")
|
||||
if label not in labels:
|
||||
raise AssemblyError(line, f"unknown label {label!r}")
|
||||
operand = (condition << 5) | labels[label]
|
||||
elif opcode == Opcode.FLAG:
|
||||
action = _symbol(args[0], FLAG_ACTIONS, line, "FLAG action")
|
||||
index = _number(args[1], line, "FLAG index", 0, 3)
|
||||
operand = (action << 7) | (index << 5)
|
||||
instruction = Instruction(opcode, statement.delay, operand)
|
||||
try:
|
||||
instruction.validate()
|
||||
except ISAError as exc:
|
||||
raise AssemblyError(line, str(exc)) from exc
|
||||
return instruction
|
||||
|
||||
|
||||
def assemble(source: str) -> Program:
|
||||
"""Assemble one strict .piolite source string into an unrelocated program."""
|
||||
|
||||
if not isinstance(source, str):
|
||||
raise TypeError("assembly source must be text")
|
||||
labels: dict[str, int] = {}
|
||||
statements: list[_Statement] = []
|
||||
directives: dict[str, tuple[tuple[str, ...], int]] = {}
|
||||
lines = source.splitlines()
|
||||
for line_number, raw_line in enumerate(lines, 1):
|
||||
text = raw_line.split(";", 1)[0].strip()
|
||||
if not text:
|
||||
continue
|
||||
label_match = re.match(r"([A-Za-z_][A-Za-z0-9_]*):(.*)\Z", text, re.ASCII)
|
||||
if label_match is not None:
|
||||
label = label_match.group(1)
|
||||
if label in labels:
|
||||
raise AssemblyError(line_number, f"duplicate label {label!r}")
|
||||
text = label_match.group(2).strip()
|
||||
if not text:
|
||||
raise AssemblyError(line_number, "a label requires an instruction on the same line")
|
||||
if text.startswith("."):
|
||||
raise AssemblyError(line_number, "a label cannot prefix a directive")
|
||||
labels[label] = len(statements)
|
||||
elif ":" in text:
|
||||
raise AssemblyError(line_number, "malformed label")
|
||||
|
||||
if text.startswith("."):
|
||||
parts = text.split(None, 1)
|
||||
directive = parts[0]
|
||||
if directive not in DIRECTIVES:
|
||||
raise AssemblyError(line_number, f"unknown directive {directive!r}")
|
||||
if directive in directives:
|
||||
raise AssemblyError(line_number, f"duplicate directive {directive}")
|
||||
operand_text = parts[1].strip() if len(parts) == 2 else ""
|
||||
args = _args(
|
||||
operand_text,
|
||||
DIRECTIVE_OPERANDS[directive],
|
||||
line_number,
|
||||
directive,
|
||||
)
|
||||
directives[directive] = (args, line_number)
|
||||
continue
|
||||
|
||||
if len(statements) >= INSTRUCTION_WORDS:
|
||||
raise AssemblyError(line_number, "program exceeds 32 instruction words")
|
||||
body, delay = _delay(text, line_number)
|
||||
parts = body.split(None, 1)
|
||||
mnemonic = parts[0]
|
||||
if mnemonic not in MNEMONICS:
|
||||
raise AssemblyError(line_number, f"unknown instruction mnemonic {mnemonic!r}")
|
||||
opcode = MNEMONICS[mnemonic]
|
||||
operand_text = parts[1].strip() if len(parts) == 2 else ""
|
||||
args = _args(operand_text, INSTRUCTION_OPERANDS[opcode], line_number, mnemonic)
|
||||
statements.append(_Statement(line_number, opcode, args, delay))
|
||||
|
||||
fallback_line = len(lines) if lines else 1
|
||||
if not statements:
|
||||
raise AssemblyError(fallback_line, "program has no instructions")
|
||||
for directive in DIRECTIVES:
|
||||
if directive not in directives:
|
||||
raise AssemblyError(fallback_line, f"missing required directive {directive}")
|
||||
|
||||
def label_offset(directive: str) -> int:
|
||||
args, line = directives[directive]
|
||||
label = _identifier(args[0], line, f"{directive} label")
|
||||
if label not in labels:
|
||||
raise AssemblyError(line, f"unknown label {label!r}")
|
||||
return labels[label]
|
||||
|
||||
def window(directive: str, maximum_width: int) -> PinWindow:
|
||||
args, line = directives[directive]
|
||||
return PinWindow(
|
||||
_number(args[0], line, f"{directive} base", 0, 31),
|
||||
_number(args[1], line, f"{directive} width", 0, maximum_width),
|
||||
)
|
||||
|
||||
def shift(directive: str) -> ShiftSpec:
|
||||
args, line = directives[directive]
|
||||
return ShiftSpec(
|
||||
_number(args[0], line, f"{directive} count", 1, 32),
|
||||
_choice(args[1], ("left", "right"), line, f"{directive} direction"),
|
||||
_choice(args[2], ("manual", "auto"), line, f"{directive} mode"),
|
||||
)
|
||||
|
||||
program_args, program_line = directives[".program"]
|
||||
name = _identifier(program_args[0], program_line, "program name")
|
||||
test_args, test_line = directives[".test"]
|
||||
if test_args[0] == "none":
|
||||
test_pin = None
|
||||
else:
|
||||
test_pin = _number(test_args[0], test_line, "test pin", 0, 31)
|
||||
test = TestSpec(test_pin, _number(test_args[1], test_line, "test flag", 0, 3))
|
||||
packing_args, packing_line = directives[".packing"]
|
||||
packing = PackingSpec(
|
||||
_number(packing_args[0], packing_line, "TX_ITEMS", 0, 0xFFFFFFFF),
|
||||
_number(packing_args[1], packing_line, "RX_ITEMS", 0, 0xFFFFFFFF),
|
||||
_number(packing_args[2], packing_line, "ITEM_BITS", 1, 32),
|
||||
)
|
||||
timing_args, timing_line = directives[".clock_ticks"]
|
||||
timing = TimingSpec(
|
||||
_number(timing_args[0], timing_line, "clock_ticks", 1, 0xFFFFFFFF),
|
||||
_choice(timing_args[1], ("fixed", "stretchable"), timing_line, "timing behavior"),
|
||||
)
|
||||
in_window = window(".in", 32)
|
||||
out_window = window(".out", 32)
|
||||
set_window = window(".set", 8)
|
||||
tx = shift(".tx_bits")
|
||||
rx = shift(".rx_bits")
|
||||
instructions = tuple(_instruction(statement, labels) for statement in statements)
|
||||
try:
|
||||
claim_mask, flag_mask = referenced_resources(
|
||||
instructions, in_window, out_window, set_window, test
|
||||
)
|
||||
except ISAError as exc:
|
||||
line = next(
|
||||
statement.line
|
||||
for statement in statements
|
||||
if statement.opcode == Opcode.BRANCH
|
||||
and statement.args[0] == "TEST_PIN_HIGH"
|
||||
)
|
||||
raise AssemblyError(line, str(exc)) from exc
|
||||
program = Program(
|
||||
name=name,
|
||||
instructions=instructions,
|
||||
entry=label_offset(".entry"),
|
||||
wrap_bottom=label_offset(".wrap_bottom"),
|
||||
wrap_top=label_offset(".wrap_top"),
|
||||
in_window=in_window,
|
||||
out_window=out_window,
|
||||
set_window=set_window,
|
||||
tx=tx,
|
||||
rx=rx,
|
||||
test=test,
|
||||
packing=packing,
|
||||
timing=timing,
|
||||
claim_mask=claim_mask,
|
||||
flag_mask=flag_mask,
|
||||
source_lines=tuple(statement.line for statement in statements),
|
||||
)
|
||||
try:
|
||||
program.validate()
|
||||
except ISAError as exc:
|
||||
match = re.search(r"instruction (\d+)", str(exc))
|
||||
line = program.source_lines[int(match.group(1))] if match else program_line
|
||||
raise AssemblyError(line, str(exc)) from exc
|
||||
return program
|
||||
|
||||
|
||||
def _choice(token: str, choices: tuple[str, ...], line: int, name: str) -> str:
|
||||
if token not in choices:
|
||||
raise AssemblyError(line, f"{name} must be {' or '.join(choices)}")
|
||||
return token
|
||||
|
||||
|
||||
def emit_c(program: Program) -> str:
|
||||
"""Emit a self-contained deterministic P1 C object, not a future HAL ABI."""
|
||||
|
||||
program.validate()
|
||||
symbol = f"piolite_p1_{program.name}"
|
||||
words = program.words
|
||||
sites = program.relocation_sites
|
||||
lines = [
|
||||
"/* Deterministic PIO-lite P1 program object; no HAL ABI is implied. */",
|
||||
"#include <stdint.h>",
|
||||
"",
|
||||
"typedef struct {",
|
||||
" uint32_t isa_version;",
|
||||
" const uint16_t *words;",
|
||||
" uint32_t word_count;",
|
||||
" uint8_t entry;",
|
||||
" uint8_t wrap_bottom;",
|
||||
" uint8_t wrap_top;",
|
||||
" const uint8_t *relocation_sites;",
|
||||
" uint32_t relocation_count;",
|
||||
" uint8_t in_base;",
|
||||
" uint8_t in_width;",
|
||||
" uint8_t out_base;",
|
||||
" uint8_t out_width;",
|
||||
" uint8_t set_base;",
|
||||
" uint8_t set_width;",
|
||||
" uint8_t tx_bits;",
|
||||
" uint8_t tx_right;",
|
||||
" uint8_t autopull;",
|
||||
" uint8_t rx_bits;",
|
||||
" uint8_t rx_right;",
|
||||
" uint8_t autopush;",
|
||||
" uint8_t test_pin;",
|
||||
" uint8_t test_flag;",
|
||||
" uint8_t test_pin_enable;",
|
||||
" uint32_t claim_mask;",
|
||||
" uint32_t flag_mask;",
|
||||
" uint32_t tx_items;",
|
||||
" uint32_t rx_items;",
|
||||
" uint8_t item_bits;",
|
||||
" uint32_t clock_ticks;",
|
||||
" uint8_t stretchable;",
|
||||
"} piolite_p1_program_object_t;",
|
||||
"",
|
||||
f"static const uint16_t {symbol}_words[{len(words)}] = {{",
|
||||
]
|
||||
lines.extend(f" 0x{word:04X}u," for word in words)
|
||||
lines.extend(
|
||||
[
|
||||
"};",
|
||||
f"static const uint8_t {symbol}_relocation_sites[{max(1, len(sites))}] = {{",
|
||||
]
|
||||
)
|
||||
if sites:
|
||||
lines.extend(f" {site}u," for site in sites)
|
||||
else:
|
||||
lines.append(" 0u,")
|
||||
lines.extend(
|
||||
[
|
||||
"};",
|
||||
f"static const piolite_p1_program_object_t {symbol} = {{",
|
||||
f" .isa_version = 0x{ISA_VERSION:08X}u,",
|
||||
f" .words = {symbol}_words,",
|
||||
f" .word_count = {len(words)}u,",
|
||||
f" .entry = {program.entry}u,",
|
||||
f" .wrap_bottom = {program.wrap_bottom}u,",
|
||||
f" .wrap_top = {program.wrap_top}u,",
|
||||
f" .relocation_sites = {symbol}_relocation_sites,",
|
||||
f" .relocation_count = {len(sites)}u,",
|
||||
f" .in_base = {program.in_window.base}u,",
|
||||
f" .in_width = {program.in_window.width}u,",
|
||||
f" .out_base = {program.out_window.base}u,",
|
||||
f" .out_width = {program.out_window.width}u,",
|
||||
f" .set_base = {program.set_window.base}u,",
|
||||
f" .set_width = {program.set_window.width}u,",
|
||||
f" .tx_bits = {program.tx.bits}u,",
|
||||
f" .tx_right = {int(program.tx.direction == 'right')}u,",
|
||||
f" .autopull = {int(program.tx.mode == 'auto')}u,",
|
||||
f" .rx_bits = {program.rx.bits}u,",
|
||||
f" .rx_right = {int(program.rx.direction == 'right')}u,",
|
||||
f" .autopush = {int(program.rx.mode == 'auto')}u,",
|
||||
f" .test_pin = {0xFF if program.test.pin is None else program.test.pin}u,",
|
||||
f" .test_flag = {program.test.flag}u,",
|
||||
f" .test_pin_enable = {int(test_pin_is_used(program.instructions))}u,",
|
||||
f" .claim_mask = 0x{program.claim_mask:08X}u,",
|
||||
f" .flag_mask = 0x{program.flag_mask:X}u,",
|
||||
f" .tx_items = {program.packing.tx_items}u,",
|
||||
f" .rx_items = {program.packing.rx_items}u,",
|
||||
f" .item_bits = {program.packing.item_bits}u,",
|
||||
f" .clock_ticks = {program.timing.clock_ticks}u,",
|
||||
f" .stretchable = {int(program.timing.behavior == 'stretchable')}u,",
|
||||
"};",
|
||||
"",
|
||||
]
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("input", type=Path, help="strict .piolite source")
|
||||
parser.add_argument("output", type=Path, help="generated C object below a build/test directory")
|
||||
args = parser.parse_args(argv)
|
||||
try:
|
||||
source = args.input.read_text(encoding="utf-8")
|
||||
program = assemble(source)
|
||||
generated = emit_c(program)
|
||||
except (OSError, UnicodeError, AssemblyError, ISAError) as exc:
|
||||
print(f"{args.input}: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
temp_path = None
|
||||
try:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
if args.output.exists():
|
||||
destination_mode = stat.S_IMODE(args.output.stat().st_mode)
|
||||
else:
|
||||
current_umask = os.umask(0)
|
||||
os.umask(current_umask)
|
||||
destination_mode = 0o666 & ~current_umask
|
||||
with tempfile.NamedTemporaryFile(
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
newline="\n",
|
||||
dir=args.output.parent,
|
||||
prefix=".piolite-",
|
||||
suffix=".tmp",
|
||||
delete=False,
|
||||
) as stream:
|
||||
temp_path = Path(stream.name)
|
||||
stream.write(generated)
|
||||
os.chmod(temp_path, destination_mode)
|
||||
os.replace(temp_path, args.output)
|
||||
except OSError as exc:
|
||||
if temp_path is not None:
|
||||
try:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
print(f"{args.output}: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
print(
|
||||
f"{program.name}: {len(program.instructions)} words, "
|
||||
f"{len(program.relocation_sites)} relocation sites, ISA 0x{ISA_VERSION:08X}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,462 @@
|
||||
"""Frozen PIO-lite ISA1.0 objects, encoding, relocation, and slot allocation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from enum import IntEnum
|
||||
|
||||
|
||||
ISA_VERSION = 0x00010000
|
||||
INSTRUCTION_WORDS = 32
|
||||
FIFO_DEPTH = 8
|
||||
SM_COUNT = 2
|
||||
|
||||
OPCODE_SHIFT = 12
|
||||
DELAY_SHIFT = 8
|
||||
BRANCH_TARGET_MASK = 0x1F
|
||||
|
||||
|
||||
class Opcode(IntEnum):
|
||||
CTRL = 0x0
|
||||
SET_PIN = 0x1
|
||||
SET_DIR = 0x2
|
||||
SET_COUNT = 0x3
|
||||
IN = 0x4
|
||||
OUT = 0x5
|
||||
MOVE = 0x6
|
||||
FIFO = 0x7
|
||||
WAIT = 0x8
|
||||
BRANCH = 0x9
|
||||
FLAG = 0xA
|
||||
|
||||
|
||||
CTRL_COMMANDS = {"NOP": 0, "HALT": 1, "CLEAR_ISR": 2, "CLEAR_OSR": 3}
|
||||
IN_SOURCES = {"PINS": 0, "X": 1, "Y": 2, "ZERO": 3}
|
||||
OUT_DESTINATIONS = {"PINS": 0, "DIRECTIONS": 1, "X": 2, "Y": 3, "DISCARD": 4}
|
||||
MOVE_REGISTERS = {"ISR": 0, "OSR": 1, "X": 2, "Y": 3}
|
||||
MOVE_SOURCES = {**MOVE_REGISTERS, "ZERO": 4, "ONES": 5}
|
||||
FIFO_OPERATIONS = {"PULL": 0, "PUSH": 1}
|
||||
FIFO_MODES = {"NONBLOCKING": 0, "BLOCKING": 1}
|
||||
BRANCH_CONDITIONS = {
|
||||
"ALWAYS": 0,
|
||||
"X_ZERO": 1,
|
||||
"Y_ZERO": 2,
|
||||
"DEC_X_NONZERO": 3,
|
||||
"DEC_Y_NONZERO": 4,
|
||||
"TEST_PIN_HIGH": 5,
|
||||
"FIFO_FAILED": 6,
|
||||
"TEST_FLAG_SET": 7,
|
||||
}
|
||||
WAIT_SOURCES = {"PIN": 0, "FLAG": 1}
|
||||
FLAG_ACTIONS = {"CLEAR": 0, "SET": 1}
|
||||
|
||||
|
||||
class ISAError(ValueError):
|
||||
"""An instruction or program violates the frozen ISA."""
|
||||
|
||||
|
||||
class AllocationError(ValueError):
|
||||
"""A program cannot be allocated in the shared instruction store."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Instruction:
|
||||
opcode: int
|
||||
delay: int = 0
|
||||
operand: int = 0
|
||||
|
||||
def validate(self) -> None:
|
||||
if not isinstance(self.opcode, int) or not 0 <= self.opcode <= 0xF:
|
||||
raise ISAError("opcode must be an unsigned four-bit value")
|
||||
if not isinstance(self.delay, int) or not 0 <= self.delay <= 15:
|
||||
raise ISAError("delay must be in 0..15")
|
||||
if not isinstance(self.operand, int) or not 0 <= self.operand <= 0xFF:
|
||||
raise ISAError("operand must be an unsigned eight-bit value")
|
||||
if self.opcode > Opcode.FLAG:
|
||||
raise ISAError("reserved opcode is an illegal instruction")
|
||||
|
||||
operand = self.operand
|
||||
if self.opcode == Opcode.CTRL:
|
||||
if operand & 0x1F or (operand >> 5) > 3:
|
||||
raise ISAError("CTRL contains reserved bits or a reserved command")
|
||||
if operand >> 5 == CTRL_COMMANDS["HALT"] and self.delay != 0:
|
||||
raise ISAError("HALT requires delay zero")
|
||||
elif self.opcode in (Opcode.SET_PIN, Opcode.SET_DIR, Opcode.SET_COUNT):
|
||||
return
|
||||
elif self.opcode == Opcode.IN:
|
||||
if operand >> 5 > 3:
|
||||
raise ISAError("IN source is reserved")
|
||||
elif self.opcode == Opcode.OUT:
|
||||
destination = operand >> 5
|
||||
if destination > 4:
|
||||
raise ISAError("OUT destination is reserved")
|
||||
if destination in (2, 3) and encoded_count(operand & 0x1F) > 16:
|
||||
raise ISAError("OUT to X/Y is limited to 16 bits")
|
||||
elif self.opcode == Opcode.MOVE:
|
||||
if operand & 0x7 or ((operand >> 3) & 0x7) > 5:
|
||||
raise ISAError("MOVE contains reserved bits or a reserved source")
|
||||
elif self.opcode == Opcode.FIFO:
|
||||
if operand & 0x3F:
|
||||
raise ISAError("FIFO reserved bits must be zero")
|
||||
elif self.opcode == Opcode.WAIT:
|
||||
if operand & 1:
|
||||
raise ISAError("WAIT reserved bit must be zero")
|
||||
if operand & 0x80 and ((operand >> 1) & 0x1F) > 3:
|
||||
raise ISAError("WAIT flag index is outside 0..3")
|
||||
elif self.opcode == Opcode.BRANCH:
|
||||
return
|
||||
elif self.opcode == Opcode.FLAG:
|
||||
if operand & 0x1F:
|
||||
raise ISAError("FLAG reserved bits must be zero")
|
||||
|
||||
def encode(self) -> int:
|
||||
self.validate()
|
||||
return (self.opcode << OPCODE_SHIFT) | (self.delay << DELAY_SHIFT) | self.operand
|
||||
|
||||
@classmethod
|
||||
def decode(cls, word: int) -> Instruction:
|
||||
if not isinstance(word, int) or not 0 <= word <= 0xFFFF:
|
||||
raise ISAError("instruction word must be an unsigned 16-bit value")
|
||||
instruction = cls((word >> OPCODE_SHIFT) & 0xF, (word >> DELAY_SHIFT) & 0xF, word & 0xFF)
|
||||
instruction.validate()
|
||||
return instruction
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PinWindow:
|
||||
base: int
|
||||
width: int
|
||||
|
||||
def validate(self, name: str, maximum_width: int = 32) -> None:
|
||||
if not isinstance(self.base, int) or not 0 <= self.base <= 31:
|
||||
raise ISAError(f"{name} base must be in 0..31")
|
||||
if not isinstance(self.width, int) or not 0 <= self.width <= maximum_width:
|
||||
raise ISAError(f"{name} width must be in 0..{maximum_width}")
|
||||
if self.base + self.width > 32:
|
||||
raise ISAError(f"{name} window exceeds GPIO31")
|
||||
|
||||
def mask(self) -> int:
|
||||
if self.width == 0:
|
||||
return 0
|
||||
return ((1 << self.width) - 1) << self.base
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ShiftSpec:
|
||||
bits: int
|
||||
direction: str
|
||||
mode: str
|
||||
|
||||
def validate(self, name: str) -> None:
|
||||
if not isinstance(self.bits, int) or not 1 <= self.bits <= 32:
|
||||
raise ISAError(f"{name} bit count must be in 1..32")
|
||||
if self.direction not in ("left", "right"):
|
||||
raise ISAError(f"{name} direction must be left or right")
|
||||
if self.mode not in ("manual", "auto"):
|
||||
raise ISAError(f"{name} mode must be manual or auto")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TestSpec:
|
||||
pin: int | None
|
||||
flag: int
|
||||
|
||||
def validate(self) -> None:
|
||||
if self.pin is not None and (not isinstance(self.pin, int) or not 0 <= self.pin <= 31):
|
||||
raise ISAError("test pin must be none or a GPIO index in 0..31")
|
||||
if not isinstance(self.flag, int) or not 0 <= self.flag <= 3:
|
||||
raise ISAError("test flag must be in 0..3")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PackingSpec:
|
||||
tx_items: int
|
||||
rx_items: int
|
||||
item_bits: int
|
||||
|
||||
def validate(self) -> None:
|
||||
for name, value in (("TX_ITEMS", self.tx_items), ("RX_ITEMS", self.rx_items)):
|
||||
if not isinstance(value, int) or not 0 <= value <= 0xFFFFFFFF:
|
||||
raise ISAError(f"packing {name} must be an unsigned 32-bit value")
|
||||
if not isinstance(self.item_bits, int) or not 1 <= self.item_bits <= 32:
|
||||
raise ISAError("packing ITEM_BITS must be in 1..32")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TimingSpec:
|
||||
clock_ticks: int
|
||||
behavior: str
|
||||
|
||||
def validate(self) -> None:
|
||||
if not isinstance(self.clock_ticks, int) or not 1 <= self.clock_ticks <= 0xFFFFFFFF:
|
||||
raise ISAError("clock_ticks must be an unsigned nonzero 32-bit value")
|
||||
if self.behavior not in ("fixed", "stretchable"):
|
||||
raise ISAError("clock_ticks behavior must be fixed or stretchable")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Program:
|
||||
name: str
|
||||
instructions: tuple[Instruction, ...]
|
||||
entry: int
|
||||
wrap_bottom: int
|
||||
wrap_top: int
|
||||
in_window: PinWindow
|
||||
out_window: PinWindow
|
||||
set_window: PinWindow
|
||||
tx: ShiftSpec
|
||||
rx: ShiftSpec
|
||||
test: TestSpec
|
||||
packing: PackingSpec
|
||||
timing: TimingSpec
|
||||
claim_mask: int
|
||||
flag_mask: int
|
||||
isa_version: int = ISA_VERSION
|
||||
source_lines: tuple[int, ...] = field(default=(), compare=False)
|
||||
|
||||
@property
|
||||
def words(self) -> tuple[int, ...]:
|
||||
return tuple(instruction.encode() for instruction in self.instructions)
|
||||
|
||||
@property
|
||||
def relocation_sites(self) -> tuple[int, ...]:
|
||||
return tuple(
|
||||
offset
|
||||
for offset, instruction in enumerate(self.instructions)
|
||||
if instruction.opcode == Opcode.BRANCH
|
||||
)
|
||||
|
||||
def validate(self) -> None:
|
||||
if self.isa_version != ISA_VERSION:
|
||||
raise ISAError(
|
||||
f"unsupported PIO-lite ISA version 0x{self.isa_version:08X}; expected 0x{ISA_VERSION:08X}"
|
||||
)
|
||||
if not self.name or not self.name.isascii() or not self.name.replace("_", "a").isalnum():
|
||||
raise ISAError("program name must be an ASCII identifier")
|
||||
if not (self.name[0].isalpha() or self.name[0] == "_"):
|
||||
raise ISAError("program name must start with a letter or underscore")
|
||||
if not 1 <= len(self.instructions) <= INSTRUCTION_WORDS:
|
||||
raise ISAError("program must contain 1..32 instructions")
|
||||
if self.source_lines and len(self.source_lines) != len(self.instructions):
|
||||
raise ISAError("source-line metadata does not match the instruction count")
|
||||
for instruction in self.instructions:
|
||||
instruction.validate()
|
||||
for name, offset in (
|
||||
("entry", self.entry),
|
||||
("wrap_bottom", self.wrap_bottom),
|
||||
("wrap_top", self.wrap_top),
|
||||
):
|
||||
if not isinstance(offset, int) or not 0 <= offset < len(self.instructions):
|
||||
raise ISAError(f"{name} is outside the program")
|
||||
if self.wrap_bottom > self.wrap_top:
|
||||
raise ISAError("wrap_bottom must not exceed wrap_top")
|
||||
self.in_window.validate("IN")
|
||||
self.out_window.validate("OUT")
|
||||
self.set_window.validate("SET", maximum_width=8)
|
||||
self.tx.validate("TX_BITS")
|
||||
self.rx.validate("RX_BITS")
|
||||
self.test.validate()
|
||||
self.packing.validate()
|
||||
self.timing.validate()
|
||||
|
||||
for offset, instruction in enumerate(self.instructions):
|
||||
operand = instruction.operand
|
||||
if instruction.opcode == Opcode.BRANCH and (operand & 0x1F) >= len(self.instructions):
|
||||
raise ISAError(f"branch at instruction {offset} is outside the program")
|
||||
if instruction.opcode == Opcode.IN and operand >> 5 == IN_SOURCES["PINS"]:
|
||||
if encoded_count(operand & 0x1F) > self.in_window.width:
|
||||
raise ISAError(f"IN at instruction {offset} exceeds the configured input window")
|
||||
if instruction.opcode == Opcode.OUT and operand >> 5 in (
|
||||
OUT_DESTINATIONS["PINS"],
|
||||
OUT_DESTINATIONS["DIRECTIONS"],
|
||||
):
|
||||
if encoded_count(operand & 0x1F) > self.out_window.width:
|
||||
raise ISAError(f"OUT at instruction {offset} exceeds the configured output window")
|
||||
if (
|
||||
instruction.opcode in (Opcode.SET_PIN, Opcode.SET_DIR)
|
||||
and self.set_window.width == 0
|
||||
):
|
||||
raise ISAError(f"SET at instruction {offset} requires an enabled SET window")
|
||||
if (
|
||||
instruction.opcode == Opcode.BRANCH
|
||||
and operand >> 5 == BRANCH_CONDITIONS["TEST_PIN_HIGH"]
|
||||
and self.test.pin is None
|
||||
):
|
||||
raise ISAError(f"test-pin branch at instruction {offset} has no selected pin")
|
||||
|
||||
expected_claim_mask, expected_flag_mask = referenced_resources(
|
||||
self.instructions, self.in_window, self.out_window, self.set_window, self.test
|
||||
)
|
||||
if self.claim_mask != expected_claim_mask:
|
||||
raise ISAError("claim-mask metadata does not match program resources")
|
||||
if self.flag_mask != expected_flag_mask:
|
||||
raise ISAError("flag metadata does not match program resources")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RelocatedProgram:
|
||||
program: Program
|
||||
base: int
|
||||
words: tuple[int, ...]
|
||||
entry: int
|
||||
wrap_bottom: int
|
||||
wrap_top: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Allocation:
|
||||
handle: int
|
||||
image: RelocatedProgram
|
||||
|
||||
@property
|
||||
def base(self) -> int:
|
||||
return self.image.base
|
||||
|
||||
@property
|
||||
def words(self) -> tuple[int, ...]:
|
||||
return self.image.words
|
||||
|
||||
|
||||
def encoded_count(field_value: int) -> int:
|
||||
if not isinstance(field_value, int) or not 0 <= field_value <= 31:
|
||||
raise ISAError("encoded count must be a five-bit value")
|
||||
return 32 if field_value == 0 else field_value
|
||||
|
||||
|
||||
def test_pin_is_used(instructions: tuple[Instruction, ...]) -> bool:
|
||||
return any(
|
||||
instruction.opcode == Opcode.BRANCH
|
||||
and instruction.operand >> 5 == BRANCH_CONDITIONS["TEST_PIN_HIGH"]
|
||||
for instruction in instructions
|
||||
)
|
||||
|
||||
|
||||
def referenced_resources(
|
||||
instructions: tuple[Instruction, ...],
|
||||
in_window: PinWindow,
|
||||
out_window: PinWindow,
|
||||
set_window: PinWindow,
|
||||
test: TestSpec,
|
||||
) -> tuple[int, int]:
|
||||
claim_mask = in_window.mask() | out_window.mask() | set_window.mask()
|
||||
flag_mask = 0
|
||||
for instruction in instructions:
|
||||
operand = instruction.operand
|
||||
if instruction.opcode == Opcode.WAIT:
|
||||
index = (operand >> 1) & 0x1F
|
||||
if operand & 0x80:
|
||||
flag_mask |= 1 << index
|
||||
else:
|
||||
claim_mask |= 1 << index
|
||||
elif instruction.opcode == Opcode.FLAG:
|
||||
flag_mask |= 1 << ((operand >> 5) & 0x3)
|
||||
elif instruction.opcode == Opcode.BRANCH:
|
||||
condition = operand >> 5
|
||||
if condition == BRANCH_CONDITIONS["TEST_PIN_HIGH"]:
|
||||
if test.pin is None:
|
||||
raise ISAError("test-pin branch requires a selected test pin")
|
||||
claim_mask |= 1 << test.pin
|
||||
elif condition == BRANCH_CONDITIONS["TEST_FLAG_SET"]:
|
||||
flag_mask |= 1 << test.flag
|
||||
return claim_mask, flag_mask
|
||||
|
||||
|
||||
def relocate_program(program: Program, base: int) -> RelocatedProgram:
|
||||
program.validate()
|
||||
if not isinstance(base, int) or not 0 <= base < INSTRUCTION_WORDS:
|
||||
raise AllocationError("allocation base must be in 0..31")
|
||||
if base + len(program.instructions) > INSTRUCTION_WORDS:
|
||||
raise AllocationError("allocation base plus program length exceeds 32 slots")
|
||||
|
||||
relocated_words = []
|
||||
for offset, instruction in enumerate(program.instructions):
|
||||
word = instruction.encode()
|
||||
if instruction.opcode == Opcode.BRANCH:
|
||||
local_target = instruction.operand & BRANCH_TARGET_MASK
|
||||
absolute_target = base + local_target
|
||||
if absolute_target > 31 or not base <= absolute_target < base + len(program.instructions):
|
||||
raise AllocationError(f"relocated branch at instruction {offset} is outside the image")
|
||||
word = (word & ~BRANCH_TARGET_MASK) | absolute_target
|
||||
relocated = Instruction.decode(word)
|
||||
relocated_words.append(relocated.encode())
|
||||
return RelocatedProgram(
|
||||
program=program,
|
||||
base=base,
|
||||
words=tuple(relocated_words),
|
||||
entry=base + program.entry,
|
||||
wrap_bottom=base + program.wrap_bottom,
|
||||
wrap_top=base + program.wrap_top,
|
||||
)
|
||||
|
||||
|
||||
class InstructionAllocator:
|
||||
"""Atomic first-fit allocator for the shared 32-slot instruction store."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._bitmap = 0
|
||||
self._allocations: dict[int, Allocation] = {}
|
||||
self._references: dict[int, int] = {}
|
||||
self._next_handle = 1
|
||||
|
||||
@property
|
||||
def valid_bitmap(self) -> int:
|
||||
return self._bitmap
|
||||
|
||||
@property
|
||||
def allocations(self) -> tuple[Allocation, ...]:
|
||||
return tuple(sorted(self._allocations.values(), key=lambda item: item.base))
|
||||
|
||||
def read_word(self, index: int) -> int:
|
||||
if not isinstance(index, int) or not 0 <= index < INSTRUCTION_WORDS:
|
||||
raise AllocationError("instruction-store index must be in 0..31")
|
||||
for allocation in self._allocations.values():
|
||||
if allocation.base <= index < allocation.base + len(allocation.words):
|
||||
return allocation.words[index - allocation.base]
|
||||
return 0
|
||||
|
||||
def load(self, program: Program, base: int | None = None) -> Allocation:
|
||||
program.validate()
|
||||
for allocation in self._allocations.values():
|
||||
if allocation.image.program == program:
|
||||
if base is not None and allocation.base != base:
|
||||
raise AllocationError("identical image is already published at another base")
|
||||
self._references[allocation.handle] += 1
|
||||
return allocation
|
||||
|
||||
length = len(program.instructions)
|
||||
if base is None:
|
||||
base = self._first_fit(length)
|
||||
if not isinstance(base, int) or not 0 <= base < INSTRUCTION_WORDS:
|
||||
raise AllocationError("allocation base must be in 0..31")
|
||||
if base + length > INSTRUCTION_WORDS:
|
||||
raise AllocationError("allocation base plus program length exceeds 32 slots")
|
||||
region_mask = ((1 << length) - 1) << base
|
||||
if self._bitmap & region_mask:
|
||||
raise AllocationError("requested instruction slots are already allocated")
|
||||
|
||||
image = relocate_program(program, base)
|
||||
for word in image.words:
|
||||
Instruction.decode(word)
|
||||
allocation = Allocation(self._next_handle, image)
|
||||
self._allocations[allocation.handle] = allocation
|
||||
self._references[allocation.handle] = 1
|
||||
self._bitmap |= region_mask
|
||||
self._next_handle += 1
|
||||
return allocation
|
||||
|
||||
def unload(self, allocation: Allocation) -> None:
|
||||
canonical = self._allocations.get(allocation.handle)
|
||||
if canonical is None or allocation is not canonical:
|
||||
raise AllocationError("unknown or foreign allocation")
|
||||
self._references[canonical.handle] -= 1
|
||||
if self._references[canonical.handle] == 0:
|
||||
length = len(canonical.words)
|
||||
self._bitmap &= ~(((1 << length) - 1) << canonical.base)
|
||||
del self._references[canonical.handle]
|
||||
del self._allocations[canonical.handle]
|
||||
|
||||
def _first_fit(self, length: int) -> int:
|
||||
for base in range(INSTRUCTION_WORDS - length + 1):
|
||||
if self._bitmap & (((1 << length) - 1) << base) == 0:
|
||||
return base
|
||||
raise AllocationError("no contiguous instruction region has enough free slots")
|
||||
@@ -0,0 +1,793 @@
|
||||
"""Cycle-stepped host reference model for the frozen two-SM PIO-lite ISA."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from collections import deque
|
||||
from dataclasses import dataclass, field
|
||||
from enum import IntEnum
|
||||
from typing import Iterable, Mapping
|
||||
|
||||
from piolite_isa import (
|
||||
BRANCH_CONDITIONS,
|
||||
CTRL_COMMANDS,
|
||||
FIFO_DEPTH,
|
||||
FIFO_MODES,
|
||||
FIFO_OPERATIONS,
|
||||
FLAG_ACTIONS,
|
||||
IN_SOURCES,
|
||||
INSTRUCTION_WORDS,
|
||||
OUT_DESTINATIONS,
|
||||
SM_COUNT,
|
||||
WAIT_SOURCES,
|
||||
Allocation,
|
||||
Instruction,
|
||||
Opcode,
|
||||
PinWindow,
|
||||
encoded_count,
|
||||
test_pin_is_used,
|
||||
)
|
||||
|
||||
|
||||
class SMState(IntEnum):
|
||||
DISABLED = 0
|
||||
RUNNING = 1
|
||||
PAUSED = 2
|
||||
HALTED = 3
|
||||
CLOSING = 4
|
||||
|
||||
|
||||
class FaultCode(IntEnum):
|
||||
NONE = 0
|
||||
ILLEGAL_INSTRUCTION = 1
|
||||
INVALID_FETCH = 2
|
||||
PROGRAM_BOUNDS = 3
|
||||
SHIFT_BOUNDS = 4
|
||||
PIN_AUTHORIZATION = 5
|
||||
|
||||
|
||||
class StallKind(IntEnum):
|
||||
NONE = 0
|
||||
PIN = 1
|
||||
FLAG = 2
|
||||
TX = 3
|
||||
RX = 4
|
||||
|
||||
|
||||
EVENT_HALT = 1 << 0
|
||||
EVENT_ABORT = 1 << 1
|
||||
EVENT_FAULT = 1 << 2
|
||||
EVENT_TX_STALL = 1 << 3
|
||||
EVENT_RX_STALL = 1 << 4
|
||||
EVENT_FIFO_FAIL = 1 << 5
|
||||
|
||||
|
||||
class FIFOError(ValueError):
|
||||
"""An external FIFO access has no pre-edge space or data."""
|
||||
|
||||
|
||||
class ControlError(ValueError):
|
||||
"""A lifecycle command is invalid for the selected SM state."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class FIFO:
|
||||
items: deque[int] = field(default_factory=deque)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if len(self.items) > FIFO_DEPTH or any(not 0 <= word <= 0xFFFFFFFF for word in self.items):
|
||||
raise FIFOError("FIFO contents exceed the 8x32 contract")
|
||||
|
||||
def push(self, word: int) -> None:
|
||||
if not isinstance(word, int) or not 0 <= word <= 0xFFFFFFFF:
|
||||
raise FIFOError("FIFO word must be an unsigned 32-bit value")
|
||||
if len(self.items) >= FIFO_DEPTH:
|
||||
raise FIFOError("FIFO is full")
|
||||
self.items.append(word)
|
||||
|
||||
def pop(self) -> int:
|
||||
if not self.items:
|
||||
raise FIFOError("FIFO is empty")
|
||||
return self.items.popleft()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SMConfig:
|
||||
base: int
|
||||
length: int
|
||||
entry: int
|
||||
wrap_bottom: int
|
||||
wrap_top: int
|
||||
divider: int = 1
|
||||
autopull: bool = False
|
||||
autopush: bool = False
|
||||
tx_right: bool = True
|
||||
rx_right: bool = True
|
||||
tx_bits: int = 32
|
||||
rx_bits: int = 32
|
||||
in_window: PinWindow = PinWindow(0, 0)
|
||||
out_window: PinWindow = PinWindow(0, 0)
|
||||
set_window: PinWindow = PinWindow(0, 0)
|
||||
claim_mask: int = 0
|
||||
output_mask: int = 0
|
||||
initial_value: int = 0
|
||||
initial_oe: int = 0
|
||||
initial_x: int = 0
|
||||
initial_y: int = 0
|
||||
test_pin_enable: bool = False
|
||||
test_pin: int | None = None
|
||||
test_flag: int = 0
|
||||
|
||||
@classmethod
|
||||
def from_allocation(
|
||||
cls,
|
||||
allocation: Allocation,
|
||||
*,
|
||||
divider: int = 1,
|
||||
initial_value: int = 0,
|
||||
initial_oe: int = 0,
|
||||
initial_x: int = 0,
|
||||
initial_y: int = 0,
|
||||
claim_mask: int | None = None,
|
||||
output_mask: int | None = None,
|
||||
test_pin_enable: bool | None = None,
|
||||
) -> SMConfig:
|
||||
image = allocation.image
|
||||
program = image.program
|
||||
return cls(
|
||||
base=image.base,
|
||||
length=len(image.words),
|
||||
entry=image.entry,
|
||||
wrap_bottom=image.wrap_bottom,
|
||||
wrap_top=image.wrap_top,
|
||||
divider=divider,
|
||||
autopull=program.tx.mode == "auto",
|
||||
autopush=program.rx.mode == "auto",
|
||||
tx_right=program.tx.direction == "right",
|
||||
rx_right=program.rx.direction == "right",
|
||||
tx_bits=program.tx.bits,
|
||||
rx_bits=program.rx.bits,
|
||||
in_window=program.in_window,
|
||||
out_window=program.out_window,
|
||||
set_window=program.set_window,
|
||||
claim_mask=program.claim_mask if claim_mask is None else claim_mask,
|
||||
output_mask=(program.out_window.mask() | program.set_window.mask())
|
||||
if output_mask is None
|
||||
else output_mask,
|
||||
initial_value=initial_value,
|
||||
initial_oe=initial_oe,
|
||||
initial_x=initial_x,
|
||||
initial_y=initial_y,
|
||||
test_pin_enable=(
|
||||
test_pin_is_used(program.instructions) if test_pin_enable is None else test_pin_enable
|
||||
),
|
||||
test_pin=program.test.pin,
|
||||
test_flag=program.test.flag,
|
||||
)
|
||||
|
||||
def validate(self) -> None:
|
||||
if not 0 <= self.base < INSTRUCTION_WORDS or not 1 <= self.length <= INSTRUCTION_WORDS:
|
||||
raise ControlError("program base/length is invalid")
|
||||
if self.base + self.length > INSTRUCTION_WORDS:
|
||||
raise ControlError("program region exceeds instruction storage")
|
||||
for name, value in (
|
||||
("entry", self.entry),
|
||||
("wrap_bottom", self.wrap_bottom),
|
||||
("wrap_top", self.wrap_top),
|
||||
):
|
||||
if not self.base <= value < self.base + self.length:
|
||||
raise ControlError(f"{name} is outside the program region")
|
||||
if self.wrap_bottom > self.wrap_top:
|
||||
raise ControlError("wrap_bottom exceeds wrap_top")
|
||||
if not 1 <= self.divider <= 65535:
|
||||
raise ControlError("divider must be in 1..65535")
|
||||
if not 1 <= self.tx_bits <= 32 or not 1 <= self.rx_bits <= 32:
|
||||
raise ControlError("shift limits must be in 1..32")
|
||||
self.in_window.validate("IN")
|
||||
self.out_window.validate("OUT")
|
||||
self.set_window.validate("SET", maximum_width=8)
|
||||
for name, value in (
|
||||
("claim_mask", self.claim_mask),
|
||||
("output_mask", self.output_mask),
|
||||
("initial_value", self.initial_value),
|
||||
("initial_oe", self.initial_oe),
|
||||
):
|
||||
if not isinstance(value, int) or not 0 <= value <= 0xFFFFFFFF:
|
||||
raise ControlError(f"{name} must be an unsigned 32-bit value")
|
||||
if self.output_mask & ~self.claim_mask:
|
||||
raise ControlError("output pins are not claimed")
|
||||
windows = self.in_window.mask() | self.out_window.mask() | self.set_window.mask()
|
||||
if windows & ~self.claim_mask:
|
||||
raise ControlError("an enabled pin window is not claimed")
|
||||
if (self.out_window.mask() | self.set_window.mask()) & ~self.output_mask:
|
||||
raise ControlError("OUT/SET windows are not authorized outputs")
|
||||
if self.initial_value & ~self.output_mask or self.initial_oe & ~self.output_mask:
|
||||
raise ControlError("initial pin state names a non-output pin")
|
||||
if not 0 <= self.initial_x <= 0xFFFF or not 0 <= self.initial_y <= 0xFFFF:
|
||||
raise ControlError("initial X/Y values must fit 16 bits")
|
||||
if self.test_pin_enable:
|
||||
if self.test_pin is None or not 0 <= self.test_pin <= 31:
|
||||
raise ControlError("enabled test pin must be in 0..31")
|
||||
if not self.claim_mask & (1 << self.test_pin):
|
||||
raise ControlError("test pin is not claimed")
|
||||
if not 0 <= self.test_flag <= 3:
|
||||
raise ControlError("test flag must be in 0..3")
|
||||
|
||||
|
||||
@dataclass
|
||||
class SM:
|
||||
config: SMConfig
|
||||
state: SMState = SMState.DISABLED
|
||||
pc: int = 0
|
||||
isr: int = 0
|
||||
osr: int = 0
|
||||
isr_count: int = 0
|
||||
osr_count: int = 0
|
||||
x: int = 0
|
||||
y: int = 0
|
||||
tx: FIFO = field(default_factory=FIFO)
|
||||
rx: FIFO = field(default_factory=FIFO)
|
||||
last_fifo_failed: bool = False
|
||||
fault_code: FaultCode = FaultCode.NONE
|
||||
fault_pc: int = 0
|
||||
stall: StallKind = StallKind.NONE
|
||||
divider_remaining: int = 1
|
||||
delay_remaining: int = 0
|
||||
tick_count: int = 0
|
||||
tx_stall_count: int = 0
|
||||
rx_stall_count: int = 0
|
||||
retired_count: int = 0
|
||||
events: int = 0
|
||||
output_value: int = 0
|
||||
output_oe: int = 0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EdgeResult:
|
||||
committed: tuple[bool, bool]
|
||||
tx_write_accepted: dict[int, bool]
|
||||
rx_read_values: dict[int, int | None]
|
||||
|
||||
|
||||
class _ExecutionFault(Exception):
|
||||
def __init__(self, code: FaultCode):
|
||||
super().__init__(code.name)
|
||||
self.code = code
|
||||
|
||||
|
||||
class _ExecutionStall(Exception):
|
||||
def __init__(self, kind: StallKind):
|
||||
super().__init__(kind.name)
|
||||
self.kind = kind
|
||||
|
||||
|
||||
class Machine:
|
||||
"""Two independent SMs stepping on explicit PCLK edges, without a watchdog."""
|
||||
|
||||
def __init__(self, configs: Iterable[SMConfig] | None = None) -> None:
|
||||
if configs is None:
|
||||
configs = (
|
||||
SMConfig(0, 1, 0, 0, 0),
|
||||
SMConfig(0, 1, 0, 0, 0),
|
||||
)
|
||||
config_tuple = tuple(configs)
|
||||
if len(config_tuple) != SM_COUNT:
|
||||
raise ControlError("Tiny PIO-lite has exactly two SM configurations")
|
||||
self.sms = [SM(config) for config in config_tuple]
|
||||
self.imem = [0] * INSTRUCTION_WORDS
|
||||
self.imem_valid = 0
|
||||
self._published_allocations: dict[int, Allocation] = {}
|
||||
self._published_references: dict[int, int] = {}
|
||||
self.flags = 0
|
||||
self.pin_inputs = 0
|
||||
self.pclk_edges = 0
|
||||
|
||||
def publish(self, allocation: Allocation) -> None:
|
||||
if any(sm.state != SMState.DISABLED for sm in self.sms):
|
||||
raise ControlError("IMEM publication requires all SMs DISABLED")
|
||||
image = allocation.image
|
||||
staged = [Instruction.decode(word).encode() for word in image.words]
|
||||
if image.base + len(staged) > INSTRUCTION_WORDS:
|
||||
raise ControlError("published image exceeds instruction storage")
|
||||
key = id(allocation)
|
||||
if key in self._published_allocations:
|
||||
if self._published_allocations[key] is not allocation:
|
||||
raise ControlError("foreign published allocation identity")
|
||||
self._published_references[key] += 1
|
||||
return
|
||||
region_mask = ((1 << len(staged)) - 1) << image.base
|
||||
if self.imem_valid & region_mask:
|
||||
raise ControlError("published instruction slots must be invalidated before replacement")
|
||||
self.imem[image.base : image.base + len(staged)] = staged
|
||||
self.imem_valid |= region_mask
|
||||
self._published_allocations[key] = allocation
|
||||
self._published_references[key] = 1
|
||||
|
||||
def unpublish(self, allocation: Allocation) -> None:
|
||||
if any(sm.state != SMState.DISABLED for sm in self.sms):
|
||||
raise ControlError("IMEM invalidation requires all SMs DISABLED")
|
||||
key = id(allocation)
|
||||
canonical = self._published_allocations.get(key)
|
||||
if canonical is None or canonical is not allocation:
|
||||
raise ControlError("unknown or foreign published allocation")
|
||||
if self._published_references[key] > 1:
|
||||
self._published_references[key] -= 1
|
||||
return
|
||||
image = canonical.image
|
||||
region_mask = ((1 << len(image.words)) - 1) << image.base
|
||||
del self._published_references[key]
|
||||
del self._published_allocations[key]
|
||||
self.imem_valid &= ~region_mask
|
||||
|
||||
def invalidate_imem(self) -> None:
|
||||
if any(sm.state != SMState.DISABLED for sm in self.sms):
|
||||
raise ControlError("IMEM invalidation requires all SMs DISABLED")
|
||||
self.imem = [0] * INSTRUCTION_WORDS
|
||||
self.imem_valid = 0
|
||||
self._published_allocations.clear()
|
||||
self._published_references.clear()
|
||||
|
||||
def read_imem(self, index: int) -> int:
|
||||
if not 0 <= index < INSTRUCTION_WORDS:
|
||||
raise ControlError("instruction index must be in 0..31")
|
||||
return self.imem[index] if self.imem_valid & (1 << index) else 0
|
||||
|
||||
def write_tx(self, sm_index: int, word: int) -> None:
|
||||
self._sm(sm_index).tx.push(word)
|
||||
|
||||
def read_rx(self, sm_index: int) -> int:
|
||||
return self._sm(sm_index).rx.pop()
|
||||
|
||||
def start(self, mask: int) -> None:
|
||||
selected = self._validate_mask(mask)
|
||||
if not selected:
|
||||
return
|
||||
for index in selected:
|
||||
sm = self.sms[index]
|
||||
if sm.state != SMState.DISABLED or sm.fault_code != FaultCode.NONE:
|
||||
raise ControlError("START requires a clean DISABLED SM")
|
||||
sm.config.validate()
|
||||
if sm.rx.items:
|
||||
raise ControlError("START requires an empty RX FIFO")
|
||||
region = ((1 << sm.config.length) - 1) << sm.config.base
|
||||
if self.imem_valid & region != region:
|
||||
raise ControlError("START requires every configured instruction word to be valid")
|
||||
for first, second in ((0, 1),):
|
||||
if self.sms[first].config.output_mask & self.sms[second].config.output_mask:
|
||||
raise ControlError("SM output masks must be disjoint")
|
||||
for index in selected:
|
||||
sm = self.sms[index]
|
||||
config = sm.config
|
||||
sm.pc = config.entry
|
||||
sm.isr = sm.osr = 0
|
||||
sm.isr_count = sm.osr_count = 0
|
||||
sm.x = config.initial_x
|
||||
sm.y = config.initial_y
|
||||
sm.last_fifo_failed = False
|
||||
sm.stall = StallKind.NONE
|
||||
sm.divider_remaining = config.divider
|
||||
sm.delay_remaining = 0
|
||||
sm.tick_count = 0
|
||||
sm.tx_stall_count = sm.rx_stall_count = sm.retired_count = 0
|
||||
sm.output_value = config.initial_value
|
||||
sm.output_oe = config.initial_oe
|
||||
sm.state = SMState.RUNNING
|
||||
|
||||
def sm_reset(self, mask: int) -> None:
|
||||
selected = self._validate_mask(mask)
|
||||
for index in selected:
|
||||
if self.sms[index].state not in (SMState.DISABLED, SMState.CLOSING):
|
||||
raise ControlError("SM_RESET requires DISABLED or CLOSING")
|
||||
for index in selected:
|
||||
sm = self.sms[index]
|
||||
config = sm.config
|
||||
self.sms[index] = SM(config=config, pc=config.entry, divider_remaining=config.divider)
|
||||
|
||||
def step(
|
||||
self,
|
||||
*,
|
||||
pause_mask: int = 0,
|
||||
resume_mask: int = 0,
|
||||
disable_mask: int = 0,
|
||||
flag_set: int = 0,
|
||||
flag_clear: int = 0,
|
||||
tx_writes: Mapping[int, int] | None = None,
|
||||
rx_reads: Iterable[int] = (),
|
||||
) -> EdgeResult:
|
||||
pauses = set(self._validate_mask(pause_mask))
|
||||
resumes = set(self._validate_mask(resume_mask))
|
||||
disables = set(self._validate_mask(disable_mask))
|
||||
if (pauses & resumes) or (pauses & disables) or (resumes & disables):
|
||||
raise ControlError("conflicting lifecycle commands for one SM")
|
||||
if not 0 <= flag_set <= 0xF or not 0 <= flag_clear <= 0xF:
|
||||
raise ControlError("flag masks must fit four bits")
|
||||
tx_writes = {} if tx_writes is None else dict(tx_writes)
|
||||
rx_indices = tuple(rx_reads)
|
||||
if any(index not in (0, 1) for index in (*tx_writes, *rx_indices)):
|
||||
raise ControlError("FIFO command names an unimplemented SM")
|
||||
if len(set(rx_indices)) != len(rx_indices):
|
||||
raise ControlError("only one external RX read is allowed per SM per edge")
|
||||
for word in tx_writes.values():
|
||||
if not isinstance(word, int) or not 0 <= word <= 0xFFFFFFFF:
|
||||
raise FIFOError("FIFO word must be an unsigned 32-bit value")
|
||||
for index in pauses:
|
||||
if self.sms[index].state != SMState.RUNNING:
|
||||
raise ControlError("PAUSE requires RUNNING")
|
||||
for index in resumes:
|
||||
if self.sms[index].state != SMState.PAUSED:
|
||||
raise ControlError("RESUME requires PAUSED")
|
||||
for index in disables:
|
||||
if self.sms[index].state == SMState.CLOSING:
|
||||
raise ControlError("DISABLE cannot escape CLOSING")
|
||||
|
||||
flags_before = self.flags
|
||||
pins_before = self.pin_inputs
|
||||
tx_had_space = tuple(len(sm.tx.items) < FIFO_DEPTH for sm in self.sms)
|
||||
rx_had_data = tuple(bool(sm.rx.items) for sm in self.sms)
|
||||
|
||||
for index in disables:
|
||||
sm = self.sms[index]
|
||||
sm.state = SMState.DISABLED
|
||||
sm.output_oe = 0
|
||||
sm.stall = StallKind.NONE
|
||||
for index in pauses:
|
||||
self.sms[index].state = SMState.PAUSED
|
||||
self.sms[index].stall = StallKind.NONE
|
||||
for index in resumes:
|
||||
self.sms[index].state = SMState.RUNNING
|
||||
|
||||
committed = [False, False]
|
||||
instruction_flag_set = 0
|
||||
instruction_flag_clear = 0
|
||||
for index, sm in enumerate(tuple(self.sms)):
|
||||
if sm.state != SMState.RUNNING:
|
||||
continue
|
||||
sm.divider_remaining -= 1
|
||||
if sm.divider_remaining != 0:
|
||||
continue
|
||||
sm.divider_remaining = sm.config.divider
|
||||
sm.tick_count = _saturating_increment(sm.tick_count)
|
||||
set_mask, clear_mask, did_commit = self._execution_tick(index, flags_before, pins_before)
|
||||
committed[index] = did_commit
|
||||
instruction_flag_set |= set_mask
|
||||
instruction_flag_clear |= clear_mask
|
||||
|
||||
self.flags = (flags_before & ~(flag_clear | instruction_flag_clear)) | flag_set | instruction_flag_set
|
||||
tx_accepted: dict[int, bool] = {}
|
||||
for index, word in tx_writes.items():
|
||||
accepted = tx_had_space[index]
|
||||
tx_accepted[index] = accepted
|
||||
if accepted:
|
||||
self.sms[index].tx.push(word)
|
||||
rx_values: dict[int, int | None] = {}
|
||||
for index in rx_indices:
|
||||
if rx_had_data[index]:
|
||||
rx_values[index] = self.sms[index].rx.pop()
|
||||
else:
|
||||
rx_values[index] = None
|
||||
self.pclk_edges += 1
|
||||
return EdgeResult(tuple(committed), tx_accepted, rx_values)
|
||||
|
||||
def _execution_tick(self, index: int, flags_before: int, pins_before: int) -> tuple[int, int, bool]:
|
||||
sm = self.sms[index]
|
||||
if sm.delay_remaining:
|
||||
sm.delay_remaining -= 1
|
||||
sm.stall = StallKind.NONE
|
||||
return 0, 0, False
|
||||
if not sm.config.base <= sm.pc < sm.config.base + sm.config.length:
|
||||
self._fault(sm, FaultCode.PROGRAM_BOUNDS)
|
||||
return 0, 0, False
|
||||
if not self.imem_valid & (1 << sm.pc):
|
||||
self._fault(sm, FaultCode.INVALID_FETCH)
|
||||
return 0, 0, False
|
||||
try:
|
||||
instruction = Instruction.decode(self.imem[sm.pc])
|
||||
except ValueError:
|
||||
self._fault(sm, FaultCode.ILLEGAL_INSTRUCTION)
|
||||
return 0, 0, False
|
||||
|
||||
candidate = copy.deepcopy(sm)
|
||||
try:
|
||||
set_mask, clear_mask = self._apply(candidate, instruction, flags_before, pins_before)
|
||||
try:
|
||||
next_pc, halted = self._successor(candidate, instruction, flags_before, pins_before)
|
||||
except _ExecutionFault:
|
||||
raise
|
||||
except _ExecutionStall as stall:
|
||||
sm.stall = stall.kind
|
||||
if stall.kind == StallKind.TX:
|
||||
sm.events |= EVENT_TX_STALL
|
||||
sm.tx_stall_count = _saturating_increment(sm.tx_stall_count)
|
||||
elif stall.kind == StallKind.RX:
|
||||
sm.events |= EVENT_RX_STALL
|
||||
sm.rx_stall_count = _saturating_increment(sm.rx_stall_count)
|
||||
return 0, 0, False
|
||||
except _ExecutionFault as primary:
|
||||
try:
|
||||
self._successor(candidate, instruction, flags_before, pins_before)
|
||||
except _ExecutionFault as successor_fault:
|
||||
self._fault(sm, FaultCode(min(primary.code, successor_fault.code)))
|
||||
else:
|
||||
self._fault(sm, primary.code)
|
||||
return 0, 0, False
|
||||
except _ExecutionFault:
|
||||
raise
|
||||
|
||||
candidate.delay_remaining = instruction.delay
|
||||
if halted:
|
||||
candidate.state = SMState.HALTED
|
||||
candidate.events |= EVENT_HALT
|
||||
else:
|
||||
candidate.pc = next_pc
|
||||
candidate.stall = StallKind.NONE
|
||||
candidate.retired_count = _saturating_increment(candidate.retired_count)
|
||||
self.sms[index] = candidate
|
||||
return set_mask, clear_mask, True
|
||||
|
||||
def _successor(
|
||||
self, sm: SM, instruction: Instruction, flags: int, pins: int
|
||||
) -> tuple[int, bool]:
|
||||
operand = instruction.operand
|
||||
if instruction.opcode == Opcode.CTRL and operand >> 5 == CTRL_COMMANDS["HALT"]:
|
||||
return sm.pc, True
|
||||
if instruction.opcode == Opcode.BRANCH:
|
||||
target = operand & 0x1F
|
||||
if not sm.config.base <= target < sm.config.base + sm.config.length:
|
||||
raise _ExecutionFault(FaultCode.PROGRAM_BOUNDS)
|
||||
condition = operand >> 5
|
||||
if condition == BRANCH_CONDITIONS["ALWAYS"]:
|
||||
taken = True
|
||||
elif condition == BRANCH_CONDITIONS["X_ZERO"]:
|
||||
taken = sm.x == 0
|
||||
elif condition == BRANCH_CONDITIONS["Y_ZERO"]:
|
||||
taken = sm.y == 0
|
||||
elif condition == BRANCH_CONDITIONS["DEC_X_NONZERO"]:
|
||||
taken = sm.x != 0
|
||||
elif condition == BRANCH_CONDITIONS["DEC_Y_NONZERO"]:
|
||||
taken = sm.y != 0
|
||||
elif condition == BRANCH_CONDITIONS["TEST_PIN_HIGH"]:
|
||||
if not sm.config.test_pin_enable or sm.config.test_pin is None:
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
if not sm.config.claim_mask & (1 << sm.config.test_pin):
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
taken = bool(pins & (1 << sm.config.test_pin))
|
||||
elif condition == BRANCH_CONDITIONS["FIFO_FAILED"]:
|
||||
taken = sm.last_fifo_failed
|
||||
elif condition == BRANCH_CONDITIONS["TEST_FLAG_SET"]:
|
||||
taken = bool(flags & (1 << sm.config.test_flag))
|
||||
else:
|
||||
raise _ExecutionFault(FaultCode.ILLEGAL_INSTRUCTION)
|
||||
if taken:
|
||||
return target, False
|
||||
next_pc = sm.config.wrap_bottom if sm.pc == sm.config.wrap_top else sm.pc + 1
|
||||
if not sm.config.base <= next_pc < sm.config.base + sm.config.length:
|
||||
raise _ExecutionFault(FaultCode.PROGRAM_BOUNDS)
|
||||
return next_pc, False
|
||||
|
||||
def _apply(
|
||||
self, sm: SM, instruction: Instruction, flags: int, pins: int
|
||||
) -> tuple[int, int]:
|
||||
opcode = instruction.opcode
|
||||
operand = instruction.operand
|
||||
if opcode == Opcode.CTRL:
|
||||
command = operand >> 5
|
||||
if command == CTRL_COMMANDS["CLEAR_ISR"]:
|
||||
sm.isr = sm.isr_count = 0
|
||||
elif command == CTRL_COMMANDS["CLEAR_OSR"]:
|
||||
sm.osr = sm.osr_count = 0
|
||||
return 0, 0
|
||||
if opcode in (Opcode.SET_PIN, Opcode.SET_DIR):
|
||||
window = sm.config.set_window
|
||||
if window.width == 0:
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
self._require_output(sm, window)
|
||||
value = (operand & 0xFF) & ((1 << window.width) - 1) if window.width else 0
|
||||
aligned = value << window.base
|
||||
if opcode == Opcode.SET_PIN:
|
||||
sm.output_value = (sm.output_value & ~window.mask()) | aligned
|
||||
else:
|
||||
sm.output_oe = (sm.output_oe & ~window.mask()) | aligned
|
||||
return 0, 0
|
||||
if opcode == Opcode.SET_COUNT:
|
||||
if operand & 0x80:
|
||||
sm.y = operand & 0x7F
|
||||
else:
|
||||
sm.x = operand & 0x7F
|
||||
return 0, 0
|
||||
if opcode == Opcode.IN:
|
||||
self._apply_in(sm, operand, pins)
|
||||
return 0, 0
|
||||
if opcode == Opcode.OUT:
|
||||
self._apply_out(sm, operand)
|
||||
return 0, 0
|
||||
if opcode == Opcode.MOVE:
|
||||
source = (operand >> 3) & 0x7
|
||||
values = (sm.isr, sm.osr, sm.x, sm.y, 0, 0xFFFFFFFF)
|
||||
value = values[source] & 0xFFFFFFFF
|
||||
destination = operand >> 6
|
||||
if destination == 0:
|
||||
sm.isr = value
|
||||
sm.isr_count = 32
|
||||
elif destination == 1:
|
||||
sm.osr = value
|
||||
sm.osr_count = 32
|
||||
elif destination == 2:
|
||||
sm.x = value & 0xFFFF
|
||||
else:
|
||||
sm.y = value & 0xFFFF
|
||||
return 0, 0
|
||||
if opcode == Opcode.FIFO:
|
||||
self._apply_fifo(sm, operand)
|
||||
return 0, 0
|
||||
if opcode == Opcode.WAIT:
|
||||
source = (operand >> 7) & 1
|
||||
level = (operand >> 6) & 1
|
||||
index = (operand >> 1) & 0x1F
|
||||
if source == WAIT_SOURCES["PIN"]:
|
||||
if not sm.config.claim_mask & (1 << index):
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
observed = int(bool(pins & (1 << index)))
|
||||
if observed != level:
|
||||
raise _ExecutionStall(StallKind.PIN)
|
||||
else:
|
||||
observed = int(bool(flags & (1 << index)))
|
||||
if observed != level:
|
||||
raise _ExecutionStall(StallKind.FLAG)
|
||||
return 0, 0
|
||||
if opcode == Opcode.BRANCH:
|
||||
condition = operand >> 5
|
||||
if condition == BRANCH_CONDITIONS["DEC_X_NONZERO"] and sm.x:
|
||||
sm.x -= 1
|
||||
elif condition == BRANCH_CONDITIONS["DEC_Y_NONZERO"] and sm.y:
|
||||
sm.y -= 1
|
||||
return 0, 0
|
||||
if opcode == Opcode.FLAG:
|
||||
index = (operand >> 5) & 0x3
|
||||
if operand >> 7 == FLAG_ACTIONS["SET"]:
|
||||
return 1 << index, 0
|
||||
return 0, 1 << index
|
||||
raise _ExecutionFault(FaultCode.ILLEGAL_INSTRUCTION)
|
||||
|
||||
def _apply_fifo(self, sm: SM, operand: int) -> None:
|
||||
blocking = ((operand >> 6) & 1) == FIFO_MODES["BLOCKING"]
|
||||
if ((operand >> 7) & 1) == FIFO_OPERATIONS["PULL"]:
|
||||
if not sm.tx.items:
|
||||
if blocking:
|
||||
raise _ExecutionStall(StallKind.TX)
|
||||
self._fifo_failed(sm)
|
||||
return
|
||||
word = sm.tx.pop()
|
||||
sm.osr = word & _mask(sm.config.tx_bits)
|
||||
sm.osr_count = sm.config.tx_bits
|
||||
sm.last_fifo_failed = False
|
||||
return
|
||||
|
||||
if sm.isr_count == 0:
|
||||
raise _ExecutionFault(FaultCode.SHIFT_BOUNDS)
|
||||
word = _normalized_isr(sm)
|
||||
if len(sm.rx.items) >= FIFO_DEPTH:
|
||||
if blocking:
|
||||
raise _ExecutionStall(StallKind.RX)
|
||||
self._fifo_failed(sm)
|
||||
return
|
||||
sm.rx.push(word)
|
||||
sm.isr = sm.isr_count = 0
|
||||
sm.last_fifo_failed = False
|
||||
|
||||
def _apply_out(self, sm: SM, operand: int) -> None:
|
||||
count = encoded_count(operand & 0x1F)
|
||||
if sm.osr_count == 0:
|
||||
if not sm.config.autopull or count > sm.config.tx_bits:
|
||||
raise _ExecutionFault(FaultCode.SHIFT_BOUNDS)
|
||||
if not sm.tx.items:
|
||||
raise _ExecutionStall(StallKind.TX)
|
||||
sm.osr = sm.tx.pop() & _mask(sm.config.tx_bits)
|
||||
sm.osr_count = sm.config.tx_bits
|
||||
sm.last_fifo_failed = False
|
||||
elif sm.osr_count < count:
|
||||
raise _ExecutionFault(FaultCode.SHIFT_BOUNDS)
|
||||
|
||||
if sm.config.tx_right:
|
||||
chunk = sm.osr & _mask(count)
|
||||
sm.osr = (sm.osr >> count) & _mask(sm.osr_count - count)
|
||||
else:
|
||||
shift = sm.osr_count - count
|
||||
chunk = (sm.osr >> shift) & _mask(count)
|
||||
sm.osr &= _mask(shift)
|
||||
sm.osr_count -= count
|
||||
|
||||
destination = operand >> 5
|
||||
if destination in (OUT_DESTINATIONS["PINS"], OUT_DESTINATIONS["DIRECTIONS"]):
|
||||
window = sm.config.out_window
|
||||
if count > window.width:
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
self._require_output(sm, window)
|
||||
aligned = chunk << window.base
|
||||
update_mask = _mask(count) << window.base
|
||||
if destination == OUT_DESTINATIONS["PINS"]:
|
||||
sm.output_value = (sm.output_value & ~update_mask) | aligned
|
||||
else:
|
||||
sm.output_oe = (sm.output_oe & ~update_mask) | aligned
|
||||
elif destination == OUT_DESTINATIONS["X"]:
|
||||
sm.x = chunk & 0xFFFF
|
||||
elif destination == OUT_DESTINATIONS["Y"]:
|
||||
sm.y = chunk & 0xFFFF
|
||||
|
||||
def _apply_in(self, sm: SM, operand: int, pins: int) -> None:
|
||||
count = encoded_count(operand & 0x1F)
|
||||
resulting_count = sm.isr_count + count
|
||||
limit = sm.config.rx_bits if sm.config.autopush else 32
|
||||
if resulting_count > limit:
|
||||
raise _ExecutionFault(FaultCode.SHIFT_BOUNDS)
|
||||
if sm.config.autopush and resulting_count == sm.config.rx_bits and len(sm.rx.items) >= FIFO_DEPTH:
|
||||
raise _ExecutionStall(StallKind.RX)
|
||||
|
||||
source = operand >> 5
|
||||
if source == IN_SOURCES["PINS"]:
|
||||
window = sm.config.in_window
|
||||
if count > window.width or window.mask() & ~sm.config.claim_mask:
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
value = (pins >> window.base) & _mask(count)
|
||||
elif source == IN_SOURCES["X"]:
|
||||
value = sm.x & _mask(count)
|
||||
elif source == IN_SOURCES["Y"]:
|
||||
value = sm.y & _mask(count)
|
||||
else:
|
||||
value = 0
|
||||
|
||||
if sm.config.rx_right:
|
||||
sm.isr = ((sm.isr >> count) | (value << (32 - count))) & 0xFFFFFFFF
|
||||
else:
|
||||
sm.isr = ((sm.isr << count) | value) & 0xFFFFFFFF
|
||||
sm.isr_count = resulting_count
|
||||
if sm.config.autopush and resulting_count == sm.config.rx_bits:
|
||||
sm.rx.push(_normalized_isr(sm))
|
||||
sm.isr = sm.isr_count = 0
|
||||
sm.last_fifo_failed = False
|
||||
|
||||
@staticmethod
|
||||
def _require_output(sm: SM, window: PinWindow) -> None:
|
||||
if window.mask() & ~sm.config.claim_mask or window.mask() & ~sm.config.output_mask:
|
||||
raise _ExecutionFault(FaultCode.PIN_AUTHORIZATION)
|
||||
|
||||
@staticmethod
|
||||
def _fifo_failed(sm: SM) -> None:
|
||||
sm.last_fifo_failed = True
|
||||
sm.events |= EVENT_FIFO_FAIL
|
||||
|
||||
@staticmethod
|
||||
def _fault(sm: SM, code: FaultCode) -> None:
|
||||
if sm.fault_code == FaultCode.NONE:
|
||||
sm.fault_code = code
|
||||
sm.fault_pc = sm.pc
|
||||
sm.state = SMState.CLOSING
|
||||
sm.output_oe = 0
|
||||
sm.stall = StallKind.NONE
|
||||
sm.events |= EVENT_FAULT
|
||||
|
||||
def _sm(self, index: int) -> SM:
|
||||
if index not in (0, 1):
|
||||
raise ControlError("SM index must be zero or one")
|
||||
return self.sms[index]
|
||||
|
||||
@staticmethod
|
||||
def _validate_mask(mask: int) -> tuple[int, ...]:
|
||||
if not isinstance(mask, int) or not 0 <= mask <= 0x3:
|
||||
raise ControlError("Tiny SM mask must contain only bits 0 and 1")
|
||||
return tuple(index for index in range(SM_COUNT) if mask & (1 << index))
|
||||
|
||||
|
||||
def _normalized_isr(sm: SM) -> int:
|
||||
if sm.isr_count == 0:
|
||||
raise _ExecutionFault(FaultCode.SHIFT_BOUNDS)
|
||||
if sm.config.rx_right:
|
||||
return (sm.isr >> (32 - sm.isr_count)) & _mask(sm.isr_count)
|
||||
return sm.isr & _mask(sm.isr_count)
|
||||
|
||||
|
||||
def _mask(width: int) -> int:
|
||||
return 0xFFFFFFFF if width == 32 else (1 << width) - 1
|
||||
|
||||
|
||||
def _saturating_increment(value: int) -> int:
|
||||
return min(0xFFFFFFFF, value + 1)
|
||||
@@ -4,6 +4,13 @@ This directory contains tests for deterministic SDK behavior and build/quality
|
||||
tooling.
|
||||
|
||||
- `c/test_runtime.c` is the host C test program used by `make sw-host-test`.
|
||||
- `test_piolite_isa.py`, `test_piolite_model.py`, and
|
||||
`test_piolite_assembler.py` cover every ISA1.0 encoding, independent two-SM
|
||||
cycle/packing traces, strict source diagnostics, deterministic C objects,
|
||||
atomic shared-store allocation, application/pair footprints, and independent
|
||||
UART sampling, pulse-phase, and logic-capture cadence/count behavior. These
|
||||
are P1 host-model checks, not RTL, protocol-waveform, firmware, or
|
||||
operating-rate qualification.
|
||||
- `test_npu_framework_reference.py` checks model-specific Softmax scaling,
|
||||
all 65536 VWW two-class input combinations, seeded KWS vectors, unchanged
|
||||
hardware artifacts and complete sample-model outputs against the locked
|
||||
|
||||
@@ -0,0 +1,445 @@
|
||||
"""Strict-source, deterministic-object, allocation, and footprint tests for PIO-lite."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
import piolite_asm # noqa: E402
|
||||
import piolite_isa # noqa: E402
|
||||
from piolite_asm import AssemblyError, assemble, emit_c # noqa: E402
|
||||
from piolite_isa import AllocationError, InstructionAllocator, ISAError # noqa: E402
|
||||
|
||||
|
||||
SOURCE = """.program demo
|
||||
.entry start
|
||||
.wrap_bottom start
|
||||
.wrap_top done
|
||||
.in 2, 1
|
||||
.out 4, 2
|
||||
.set 6, 3
|
||||
.tx_bits 8, right, auto
|
||||
.rx_bits 16, left, manual
|
||||
.test 2, 1
|
||||
.packing 4, 0, 8
|
||||
.clock_ticks 7, stretchable
|
||||
start: CTRL NOP [1]
|
||||
done: CTRL HALT
|
||||
"""
|
||||
|
||||
|
||||
def make_source(name: str, body: str, *, top: str = "start") -> str:
|
||||
return f""".program {name}
|
||||
.entry start
|
||||
.wrap_bottom start
|
||||
.wrap_top {top}
|
||||
.in 0, 0
|
||||
.out 0, 0
|
||||
.set 0, 0
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test none, 0
|
||||
.packing 0, 0, 8
|
||||
.clock_ticks 1, fixed
|
||||
{body}
|
||||
"""
|
||||
|
||||
|
||||
def sized_program(name: str, length: int) -> piolite_isa.Program:
|
||||
if length == 1:
|
||||
return assemble(make_source(name, "start: CTRL HALT"))
|
||||
lines = ["start: CTRL NOP"]
|
||||
lines.extend("CTRL NOP" for _ in range(length - 2))
|
||||
lines.append("last: CTRL HALT")
|
||||
return assemble(make_source(name, "\n".join(lines), top="last"))
|
||||
|
||||
|
||||
def test_deterministic_assembly_and_independent_c_object_golden() -> None:
|
||||
first = assemble(SOURCE)
|
||||
second = assemble(SOURCE)
|
||||
assert first == second
|
||||
assert first.words == (0x0100, 0x0020)
|
||||
assert first.relocation_sites == ()
|
||||
assert first.claim_mask == 0x000001F4
|
||||
assert first.flag_mask == 0
|
||||
generated = emit_c(first).encode("utf-8")
|
||||
assert generated == emit_c(second).encode("utf-8")
|
||||
assert len(generated) == 1780
|
||||
assert hashlib.sha256(generated).hexdigest() == (
|
||||
"07c41f5875b17c8d9fb439d4b6169c26ef47739eca44af35dd9ee61783d675f6"
|
||||
)
|
||||
assert b"0x00010000u" in generated
|
||||
assert b".test_pin_enable = 0u" in generated
|
||||
assert b"0x0100u,\n 0x0020u," in generated
|
||||
assert b".entry = 0u" in generated
|
||||
assert b".wrap_top = 1u" in generated
|
||||
assert str(ROOT).encode() not in generated
|
||||
|
||||
|
||||
def test_resource_metadata_tracks_actual_wait_flag_and_test_references() -> None:
|
||||
program = assemble(
|
||||
""".program resources
|
||||
.entry start
|
||||
.wrap_bottom start
|
||||
.wrap_top last
|
||||
.in 0, 0
|
||||
.out 0, 0
|
||||
.set 0, 0
|
||||
.tx_bits 32, right, manual
|
||||
.rx_bits 32, right, manual
|
||||
.test 31, 2
|
||||
.packing 0, 0, 8
|
||||
.clock_ticks 1, stretchable
|
||||
start: WAIT PIN, 1, 7
|
||||
WAIT FLAG, 0, 3
|
||||
FLAG SET, 1
|
||||
BRANCH TEST_PIN_HIGH, start
|
||||
BRANCH TEST_FLAG_SET, start
|
||||
last: CTRL HALT
|
||||
"""
|
||||
)
|
||||
assert program.claim_mask == (1 << 31) | (1 << 7)
|
||||
assert program.flag_mask == (1 << 3) | (1 << 2) | (1 << 1)
|
||||
assert program.relocation_sites == (3, 4)
|
||||
|
||||
|
||||
BAD_SOURCES = (
|
||||
("duplicate label", make_source("p", "start: CTRL NOP\nstart: CTRL HALT"), "duplicate label"),
|
||||
("unknown label", make_source("p", "start: BRANCH ALWAYS, missing"), "unknown label"),
|
||||
("label only", make_source("p", "start:"), "same line"),
|
||||
("label directive", ".entry x\nstart: .program p", "cannot prefix"),
|
||||
("lowercase mnemonic", make_source("p", "start: ctrl NOP"), "unknown instruction"),
|
||||
("lowercase operand", make_source("p", "start: CTRL nop"), "unknown CTRL"),
|
||||
("raw branch", make_source("p", "start: BRANCH ALWAYS, 0"), "ASCII identifier"),
|
||||
("signed number", make_source("p", "start: SET_PIN -1"), "unsigned decimal"),
|
||||
("expression", make_source("p", "start: SET_PIN 1 + 1"), "exactly one operand"),
|
||||
("uppercase hex introducer", make_source("p", "start: SET_PIN 0X1"), "unsigned decimal"),
|
||||
("hash comment", make_source("p", "start: CTRL HALT # no"), "exactly one operand"),
|
||||
("missing comma", make_source("p", "start: WAIT PIN 1 2"), "comma-separated"),
|
||||
("delay range", make_source("p", "start: CTRL NOP [16]"), "0..15"),
|
||||
("signed delay", make_source("p", "start: CTRL NOP [-1]"), "unsigned decimal"),
|
||||
("include", ".include other\n", "unknown directive"),
|
||||
("missing metadata", SOURCE.replace(".test 2, 1\n", ""), "missing required directive .test"),
|
||||
("duplicate metadata", SOURCE.replace(".out 4, 2", ".out 4, 2\n.in 2, 1"), "duplicate directive .in"),
|
||||
("bad wrap", make_source("p", "start: CTRL HALT", top="missing"), "unknown label"),
|
||||
("disabled input", make_source("p", "start: IN PINS, 1"), "exceeds the configured input"),
|
||||
("wide counter OUT", make_source("p", "start: OUT X, 32"), "limited to 16"),
|
||||
("counter immediate", make_source("p", "start: SET_COUNT X, 128"), "0..127"),
|
||||
("missing test pin", make_source("p", "start: BRANCH TEST_PIN_HIGH, start"), "selected test pin"),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("name", "source", "message"), BAD_SOURCES, ids=[case[0] for case in BAD_SOURCES])
|
||||
def test_strict_source_rejection(name: str, source: str, message: str) -> None:
|
||||
del name
|
||||
with pytest.raises(AssemblyError, match=message):
|
||||
assemble(source)
|
||||
|
||||
|
||||
def test_oversize_and_line_diagnostics_are_deterministic() -> None:
|
||||
with pytest.raises(AssemblyError) as caught:
|
||||
sized_program("too_large", 33)
|
||||
assert caught.value.line == 45
|
||||
assert "exceeds 32" in str(caught.value)
|
||||
with pytest.raises(AssemblyError) as duplicate:
|
||||
assemble(make_source("p", "start: CTRL NOP\nstart: CTRL HALT"))
|
||||
assert str(duplicate.value) == "line 14: duplicate label 'start'"
|
||||
|
||||
|
||||
def test_allocator_relocation_exhaustion_fragmentation_and_sharing() -> None:
|
||||
at_zero = InstructionAllocator()
|
||||
at_31 = InstructionAllocator()
|
||||
branch = assemble(make_source("branch", "start: BRANCH ALWAYS, start"))
|
||||
assert at_zero.load(branch, 0).words == (0x9000,)
|
||||
assert at_31.load(branch, 31).words == (0x901F,)
|
||||
|
||||
full = InstructionAllocator()
|
||||
for index in range(32):
|
||||
full.load(sized_program(f"p{index}", 1))
|
||||
assert full.valid_bitmap == 0xFFFFFFFF
|
||||
with pytest.raises(AllocationError, match="no contiguous"):
|
||||
full.load(sized_program("overflow", 1))
|
||||
|
||||
fragmented = InstructionAllocator()
|
||||
for index, base in enumerate(range(0, 32, 2)):
|
||||
fragmented.load(sized_program(f"f{index}", 1), base)
|
||||
assert fragmented.valid_bitmap == 0x55555555
|
||||
with pytest.raises(AllocationError, match="no contiguous"):
|
||||
fragmented.load(sized_program("pair", 2))
|
||||
|
||||
shared = InstructionAllocator()
|
||||
program = sized_program("shared", 3)
|
||||
first = shared.load(program)
|
||||
second = shared.load(program)
|
||||
assert first == second
|
||||
assert shared.valid_bitmap == 0b111
|
||||
shared.unload(first)
|
||||
assert shared.valid_bitmap == 0b111
|
||||
shared.unload(second)
|
||||
assert shared.valid_bitmap == 0
|
||||
|
||||
|
||||
def test_invalid_regions_and_interrupted_load_do_not_publish(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
allocator = InstructionAllocator()
|
||||
valid = sized_program("valid", 2)
|
||||
for invalid in (
|
||||
replace(valid, entry=2),
|
||||
replace(valid, wrap_bottom=1, wrap_top=0),
|
||||
replace(valid, wrap_top=2),
|
||||
replace(valid, isa_version=0x00020000),
|
||||
):
|
||||
with pytest.raises(ISAError):
|
||||
allocator.load(invalid)
|
||||
assert allocator.valid_bitmap == 0
|
||||
|
||||
def interrupt(program: piolite_isa.Program, base: int) -> piolite_isa.RelocatedProgram:
|
||||
del program, base
|
||||
raise KeyboardInterrupt
|
||||
|
||||
monkeypatch.setattr(piolite_isa, "relocate_program", interrupt)
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
allocator.load(valid)
|
||||
assert allocator.valid_bitmap == 0
|
||||
assert allocator.allocations == ()
|
||||
|
||||
|
||||
def test_cli_emits_byte_identical_files_and_source_diagnostics(tmp_path: Path) -> None:
|
||||
source_path = tmp_path / "demo.piolite"
|
||||
source_path.write_text(SOURCE, encoding="utf-8")
|
||||
outputs = []
|
||||
for index in range(2):
|
||||
output = tmp_path / f"demo-{index}.c"
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(ROOT / "scripts" / "piolite_asm.py"), str(source_path), str(output)],
|
||||
cwd=ROOT,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "demo: 2 words, 0 relocation sites" in result.stdout
|
||||
outputs.append(output.read_bytes())
|
||||
assert outputs[0] == outputs[1]
|
||||
|
||||
bad_path = tmp_path / "bad.piolite"
|
||||
bad_path.write_text(make_source("bad", "start: BRANCH ALWAYS, nowhere"), encoding="utf-8")
|
||||
failed = subprocess.run(
|
||||
[sys.executable, str(ROOT / "scripts" / "piolite_asm.py"), str(bad_path), str(tmp_path / "bad.c")],
|
||||
cwd=ROOT,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert failed.returncode == 2
|
||||
assert "line 13: unknown label 'nowhere'" in failed.stderr
|
||||
assert not (tmp_path / "bad.c").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("preexisting", (False, True), ids=("new-destination", "existing-destination"))
|
||||
def test_cli_write_failure_cleans_temporary_and_preserves_destination(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
preexisting: bool,
|
||||
) -> None:
|
||||
source_path = tmp_path / "input.piolite"
|
||||
output_path = tmp_path / "result.c"
|
||||
source_path.write_text(SOURCE, encoding="utf-8")
|
||||
original = b"known-good\n"
|
||||
if preexisting:
|
||||
output_path.write_bytes(original)
|
||||
real_factory = tempfile.NamedTemporaryFile
|
||||
|
||||
class FailingStream:
|
||||
def __init__(self, stream):
|
||||
self.stream = stream
|
||||
self.name = stream.name
|
||||
|
||||
def __enter__(self):
|
||||
self.stream.__enter__()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, traceback):
|
||||
return self.stream.__exit__(exc_type, exc_value, traceback)
|
||||
|
||||
def write(self, text):
|
||||
self.stream.write(text[:32])
|
||||
self.stream.flush()
|
||||
raise OSError("injected write failure")
|
||||
|
||||
monkeypatch.setattr(
|
||||
piolite_asm.tempfile,
|
||||
"NamedTemporaryFile",
|
||||
lambda *args, **kwargs: FailingStream(real_factory(*args, **kwargs)),
|
||||
)
|
||||
result = piolite_asm.main([str(source_path), str(output_path)])
|
||||
assert result == 2
|
||||
assert "injected write failure" in capsys.readouterr().err
|
||||
if preexisting:
|
||||
assert output_path.read_bytes() == original
|
||||
else:
|
||||
assert not output_path.exists()
|
||||
assert list(tmp_path.glob(".piolite-*.tmp")) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name_length", (242, 255))
|
||||
def test_cli_near_name_max_output_uses_bounded_temporary_name(
|
||||
tmp_path: Path, name_length: int
|
||||
) -> None:
|
||||
source_path = tmp_path / "input.piolite"
|
||||
output_path = tmp_path / ("p" * name_length)
|
||||
source_path.write_text(SOURCE, encoding="utf-8")
|
||||
result = piolite_asm.main([str(source_path), str(output_path)])
|
||||
assert result == 0
|
||||
assert output_path.read_bytes() == emit_c(assemble(SOURCE)).encode("utf-8")
|
||||
assert list(tmp_path.glob(".piolite-*.tmp")) == []
|
||||
|
||||
|
||||
def test_cli_new_output_mode_follows_umask(tmp_path: Path) -> None:
|
||||
source_path = tmp_path / "input.piolite"
|
||||
output_path = tmp_path / "new.c"
|
||||
source_path.write_text(SOURCE, encoding="utf-8")
|
||||
previous_umask = os.umask(0o022)
|
||||
try:
|
||||
result = piolite_asm.main([str(source_path), str(output_path)])
|
||||
finally:
|
||||
os.umask(previous_umask)
|
||||
assert result == 0
|
||||
assert output_path.stat().st_mode & 0o777 == 0o644
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", (0o644, 0o444))
|
||||
def test_cli_existing_output_mode_is_preserved(tmp_path: Path, mode: int) -> None:
|
||||
source_path = tmp_path / "input.piolite"
|
||||
output_path = tmp_path / "existing.c"
|
||||
source_path.write_text(SOURCE, encoding="utf-8")
|
||||
output_path.write_bytes(b"old\n")
|
||||
os.chmod(output_path, mode)
|
||||
result = piolite_asm.main([str(source_path), str(output_path)])
|
||||
assert result == 0
|
||||
assert output_path.stat().st_mode & 0o777 == mode
|
||||
assert output_path.read_bytes() == emit_c(assemble(SOURCE)).encode("utf-8")
|
||||
|
||||
|
||||
def test_foreign_allocation_handle_collision_is_rejected_atomically() -> None:
|
||||
allocator_a = InstructionAllocator()
|
||||
allocator_b = InstructionAllocator()
|
||||
program_a = sized_program("allocator_a", 2)
|
||||
program_b = sized_program("allocator_b", 1)
|
||||
allocation_a = allocator_a.load(program_a, 0)
|
||||
allocation_b = allocator_b.load(program_b, 5)
|
||||
assert allocation_a.handle == allocation_b.handle
|
||||
before = (
|
||||
allocator_a.valid_bitmap,
|
||||
allocator_a.allocations,
|
||||
dict(allocator_a._references),
|
||||
)
|
||||
|
||||
with pytest.raises(AllocationError, match="foreign allocation"):
|
||||
allocator_a.unload(allocation_b)
|
||||
assert (
|
||||
allocator_a.valid_bitmap,
|
||||
allocator_a.allocations,
|
||||
dict(allocator_a._references),
|
||||
) == before
|
||||
assert allocator_b.valid_bitmap == 1 << 5
|
||||
|
||||
allocator_a.unload(allocation_a)
|
||||
assert allocator_a.valid_bitmap == 0
|
||||
replacement = allocator_a.load(program_a, 0)
|
||||
assert replacement.base == 0
|
||||
assert allocator_a.valid_bitmap == 0b11
|
||||
allocator_b.unload(allocation_b)
|
||||
assert allocator_b.valid_bitmap == 0
|
||||
|
||||
|
||||
PATHOLOGICAL_NUMBERS = ("9" * 5000, "0x" + "F" * 5000)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("token", PATHOLOGICAL_NUMBERS, ids=("long-decimal", "long-hexadecimal"))
|
||||
def test_pathological_numbers_have_normal_assembly_diagnostics(token: str) -> None:
|
||||
with pytest.raises(AssemblyError) as caught:
|
||||
assemble(make_source("p", f"start: SET_PIN {token}"))
|
||||
assert caught.value.line == 13
|
||||
assert "SET immediate" in caught.value.message
|
||||
|
||||
|
||||
@pytest.mark.parametrize("token", PATHOLOGICAL_NUMBERS, ids=("long-decimal", "long-hexadecimal"))
|
||||
def test_cli_pathological_numbers_have_normal_diagnostics(tmp_path: Path, token: str) -> None:
|
||||
source_path = tmp_path / "pathological.piolite"
|
||||
output_path = tmp_path / "pathological.c"
|
||||
source_path.write_text(make_source("p", f"start: SET_PIN {token}"), encoding="utf-8")
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(ROOT / "scripts" / "piolite_asm.py"), str(source_path), str(output_path)],
|
||||
cwd=ROOT,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 2
|
||||
assert "line 13: SET immediate" in result.stderr
|
||||
assert "Traceback" not in result.stderr
|
||||
assert not output_path.exists()
|
||||
|
||||
|
||||
EXPECTED_FOOTPRINTS = {
|
||||
"logic_capture": 9,
|
||||
"pulse_generate": 12,
|
||||
"pulse_capture": 8,
|
||||
"uart_tx": 12,
|
||||
"uart_rx": 12,
|
||||
"spi_master": 15,
|
||||
"ir_tx": 10,
|
||||
"ir_rx": 13,
|
||||
"led_scan": 14,
|
||||
"gamepad": 11,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("name", "words"), sorted(EXPECTED_FOOTPRINTS.items()))
|
||||
def test_actual_application_footprint(name: str, words: int) -> None:
|
||||
source = (ROOT / "app" / "asm" / "piolite" / f"{name}.piolite").read_text(encoding="utf-8")
|
||||
program = assemble(source)
|
||||
assert len(program.instructions) == words
|
||||
assert emit_c(program).encode("utf-8") == emit_c(assemble(source)).encode("utf-8")
|
||||
|
||||
|
||||
def test_logic_capture_metadata_describes_application_samples() -> None:
|
||||
source = (ROOT / "app" / "asm" / "piolite" / "logic_capture.piolite").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
program = assemble(source)
|
||||
assert program.packing.rx_items == 32768
|
||||
assert program.packing.item_bits == 8
|
||||
assert program.timing.clock_ticks == 2
|
||||
generated = emit_c(program)
|
||||
assert ".rx_items = 32768u" in generated
|
||||
assert ".item_bits = 8u" in generated
|
||||
assert ".clock_ticks = 2u" in generated
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("first", "second", "total"),
|
||||
(("pulse_generate", "pulse_capture", 20), ("uart_tx", "uart_rx", 24), ("ir_tx", "ir_rx", 23)),
|
||||
)
|
||||
def test_required_dual_sm_images_fit_together(first: str, second: str, total: int) -> None:
|
||||
allocator = InstructionAllocator()
|
||||
for name in (first, second):
|
||||
source = (ROOT / "app" / "asm" / "piolite" / f"{name}.piolite").read_text(encoding="utf-8")
|
||||
allocator.load(assemble(source))
|
||||
assert sum(len(allocation.words) for allocation in allocator.allocations) == total
|
||||
assert allocator.valid_bitmap == (1 << total) - 1
|
||||
for allocation in allocator.allocations:
|
||||
for word in allocation.words:
|
||||
piolite_isa.Instruction.decode(word)
|
||||
@@ -0,0 +1,157 @@
|
||||
"""Independent frozen encodings and instruction-store checks for PIO-lite P1."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from piolite_isa import ( # noqa: E402
|
||||
ISA_VERSION,
|
||||
AllocationError,
|
||||
Instruction,
|
||||
InstructionAllocator,
|
||||
ISAError,
|
||||
Opcode,
|
||||
PackingSpec,
|
||||
PinWindow,
|
||||
Program,
|
||||
ShiftSpec,
|
||||
TestSpec as PIOTestSpec,
|
||||
TimingSpec,
|
||||
encoded_count,
|
||||
relocate_program,
|
||||
)
|
||||
|
||||
|
||||
GOLDEN = (
|
||||
(0x0000, Instruction(0x0, 0, 0x00)),
|
||||
(0x0020, Instruction(0x0, 0, 0x20)),
|
||||
(0x0F40, Instruction(0x0, 15, 0x40)),
|
||||
(0x0060, Instruction(0x0, 0, 0x60)),
|
||||
(0x1FAB, Instruction(0x1, 15, 0xAB)),
|
||||
(0x2001, Instruction(0x2, 0, 0x01)),
|
||||
(0x30FF, Instruction(0x3, 0, 0xFF)),
|
||||
(0x4000, Instruction(0x4, 0, 0x00)),
|
||||
(0x417F, Instruction(0x4, 1, 0x7F)),
|
||||
(0x509F, Instruction(0x5, 0, 0x9F)),
|
||||
(0x504F, Instruction(0x5, 0, 0x4F)),
|
||||
(0x60E8, Instruction(0x6, 0, 0xE8)),
|
||||
(0x70C0, Instruction(0x7, 0, 0xC0)),
|
||||
(0x80C6, Instruction(0x8, 0, 0xC6)),
|
||||
(0x927F, Instruction(0x9, 2, 0x7F)),
|
||||
(0xA0E0, Instruction(0xA, 0, 0xE0)),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("word", "instruction"), GOLDEN)
|
||||
def test_independent_golden_encoding(word: int, instruction: Instruction) -> None:
|
||||
assert instruction.encode() == word
|
||||
assert Instruction.decode(word) == instruction
|
||||
|
||||
|
||||
def _legal_form(opcode: int, delay: int, operand: int) -> bool:
|
||||
if opcode == 0:
|
||||
return operand & 0x1F == 0 and operand >> 5 <= 3 and not (
|
||||
operand == 0x20 and delay != 0
|
||||
)
|
||||
if opcode in (1, 2, 3, 9):
|
||||
return True
|
||||
if opcode == 4:
|
||||
return operand >> 5 <= 3
|
||||
if opcode == 5:
|
||||
count = 32 if operand & 0x1F == 0 else operand & 0x1F
|
||||
return operand >> 5 <= 4 and not (operand >> 5 in (2, 3) and count > 16)
|
||||
if opcode == 6:
|
||||
return operand & 7 == 0 and (operand >> 3) & 7 <= 5
|
||||
if opcode == 7:
|
||||
return operand & 0x3F == 0
|
||||
if opcode == 8:
|
||||
return operand & 1 == 0 and not (operand & 0x80 and (operand >> 1) & 0x1F > 3)
|
||||
if opcode == 10:
|
||||
return operand & 0x1F == 0
|
||||
return False
|
||||
|
||||
|
||||
def test_every_legal_and_illegal_encoding_form() -> None:
|
||||
legal_count = 0
|
||||
illegal_count = 0
|
||||
for opcode in range(16):
|
||||
for delay in range(16):
|
||||
for operand in range(256):
|
||||
word = (opcode << 12) | (delay << 8) | operand
|
||||
if _legal_form(opcode, delay, operand):
|
||||
legal_count += 1
|
||||
assert Instruction.decode(word).encode() == word
|
||||
else:
|
||||
illegal_count += 1
|
||||
with pytest.raises(ISAError):
|
||||
Instruction.decode(word)
|
||||
assert legal_count == 22257
|
||||
assert illegal_count == 43279
|
||||
|
||||
|
||||
@pytest.mark.parametrize("word", (-1, 0x10000, 0x100000))
|
||||
def test_word_width_is_checked(word: int) -> None:
|
||||
with pytest.raises(ISAError):
|
||||
Instruction.decode(word)
|
||||
|
||||
|
||||
def test_count_zero_is_32_only_for_in_out() -> None:
|
||||
assert encoded_count(0) == 32
|
||||
assert encoded_count(1) == 1
|
||||
assert encoded_count(31) == 31
|
||||
assert Instruction(4, 0, 0).operand & 0x1F == 0
|
||||
assert Instruction(3, 0, 0).encode() == 0x3000
|
||||
with pytest.raises(ISAError):
|
||||
encoded_count(32)
|
||||
|
||||
|
||||
def program(instructions: tuple[Instruction, ...] = (Instruction(0, 0, 0x20),)) -> Program:
|
||||
return Program(
|
||||
name="golden",
|
||||
instructions=instructions,
|
||||
entry=0,
|
||||
wrap_bottom=0,
|
||||
wrap_top=len(instructions) - 1,
|
||||
in_window=PinWindow(0, 0),
|
||||
out_window=PinWindow(0, 0),
|
||||
set_window=PinWindow(0, 0),
|
||||
tx=ShiftSpec(32, "right", "manual"),
|
||||
rx=ShiftSpec(32, "right", "manual"),
|
||||
test=PIOTestSpec(None, 0),
|
||||
packing=PackingSpec(0, 0, 8),
|
||||
timing=TimingSpec(1, "fixed"),
|
||||
claim_mask=0,
|
||||
flag_mask=0,
|
||||
)
|
||||
|
||||
|
||||
def test_relocation_at_slot_zero_and_31_patches_only_branch_target() -> None:
|
||||
branch_program = replace(program((Instruction(Opcode.BRANCH, 7, 0xE0),)), flag_mask=1)
|
||||
at_zero = relocate_program(branch_program, 0)
|
||||
at_31 = relocate_program(branch_program, 31)
|
||||
assert at_zero.words == (0x97E0,)
|
||||
assert at_31.words == (0x97FF,)
|
||||
assert at_zero.words[0] ^ at_31.words[0] == 0x1F
|
||||
assert at_31.entry == at_31.wrap_bottom == at_31.wrap_top == 31
|
||||
|
||||
|
||||
def test_isa_version_and_relocation_bounds_fail_before_publication() -> None:
|
||||
allocator = InstructionAllocator()
|
||||
with pytest.raises(ISAError, match="ISA version"):
|
||||
allocator.load(replace(program(), isa_version=ISA_VERSION + 1))
|
||||
bad_branch = program((Instruction(Opcode.BRANCH, 0, 1),))
|
||||
with pytest.raises(ISAError, match="outside the program"):
|
||||
allocator.load(bad_branch)
|
||||
with pytest.raises(AllocationError):
|
||||
relocate_program(program(), 32)
|
||||
with pytest.raises(AllocationError):
|
||||
relocate_program(program((Instruction(0), Instruction(0, 0, 0x20))), 31)
|
||||
assert allocator.valid_bitmap == 0
|
||||
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user