diff --git a/dau_driver/capability_decode.py b/dau_driver/capability_decode.py index 1f08ade..d305d17 100644 --- a/dau_driver/capability_decode.py +++ b/dau_driver/capability_decode.py @@ -11,18 +11,21 @@ from dau_core.capabilities import OperatorCapability from dau_core.capability_contract import CapabilityContract -from dau_core.registers import DEFAULT_STREAM_JOB_REGISTER_CONTRACT, RegisterBlockSnapshot, RegisterMapVersion, RegisterOffset, StreamProtocolVersion +from dau_core.registers import ( + CAPABILITY_WORDS_REGISTER_MAP_VERSION, + DEFAULT_STREAM_JOB_REGISTER_CONTRACT, + JOB_CAPACITY_REGISTER_MAP_VERSION, + RegisterBlockSnapshot, + RegisterMapVersion, + RegisterOffset, + StreamProtocolVersion, +) from dau_core.stream import LogicalType, OperationCode, output_type_for # the register-level type claim of the bitmap words — every # opcode-driven tile in the palette is INT32 (per-tile types travel in the # manifest capability contract, not in registers) REGISTER_BITMAP_INPUT_TYPES = (LogicalType.INT32,) -CAPABILITY_WORDS_REGISTER_MAP_VERSION = RegisterMapVersion(0, 2) -# job capacity arrived in 0.3. A 0.2 shell does not decode those offsets, so -# they are not read from one -- it reports no capacity and the host refuses -# to batch rather than guessing at depths it cannot see. -JOB_CAPACITY_REGISTER_MAP_VERSION = RegisterMapVersion(0, 3) # identity window candidates: shell base, then the stream-job contract window REGISTER_WINDOW_OFFSETS = (0x0, DEFAULT_STREAM_JOB_REGISTER_CONTRACT.register_window_offset) diff --git a/dau_driver/pipeline_execution.py b/dau_driver/pipeline_execution.py index f9aa51f..57d88c6 100644 --- a/dau_driver/pipeline_execution.py +++ b/dau_driver/pipeline_execution.py @@ -13,12 +13,21 @@ from dataclasses import dataclass from dau_core.operators import FilterPredicate, GroupAggregate, MapConfig +from dau_core.records import ( + BAR_RECORD_BYTES, + BAR_RECORD_FIELDS, + BAR_ROW_BYTES, + BAR_ROW_FIELDS, + GROUP_RECORD_BYTES, + GROUP_RECORD_FIELDS, + KEYED_ROW_FIELDS, + struct_format, +) from dau_core.registers import JobControlFlag, JobStatusFlag, RegisterOffset, StreamJobRegisterContract from .device import write_address from .errors import RegisterAccessError -GROUP_RECORD_BYTES = 40 DMA_ALIGN_BYTES = 64 @@ -71,14 +80,14 @@ def map_constant_word(self) -> int: def pack_keyed_rows(rows: list[tuple[int, int]]) -> bytes: """{value, key} pair rows on the 64-byte DMA grid; the logical length stays ``len(rows) * 8``.""" - payload = b"".join(struct.pack(" list[GroupAggregate]: groups = [] for offset in range(0, count * GROUP_RECORD_BYTES, GROUP_RECORD_BYTES): - key, first, last, minimum, maximum, _pad, total, rows = struct.unpack_from(" str: - """The little-endian ``struct`` format for a declared field layout.""" - return "<" + "".join(_STRUCT_CODE[kind] for _name, kind in fields) - - -def numpy_dtype_fields(fields: tuple[tuple[str, str], ...]) -> list[tuple[str, str]]: - """The same layout as numpy dtype pairs, for vectorized consumers.""" - return [(name, "<" + kind) for name, kind in fields] - - -BAR_ROW_BYTES = 16 -BAR_RECORD_BYTES = 48 - - def pack_bar_rows(rows: list[tuple[int, int, int, int]]) -> bytes: """(key, price, shares, frac) rows for the fused bar tile, on the DMA grid; the logical length stays ``len(rows) * 16``.""" @@ -441,7 +420,9 @@ def device_job_capacity(device) -> tuple[int, int]: reflashed inside one session -- the flywheel does exactly that -- and a cached capacity would then describe the previous shell. """ - from .capability_decode import JOB_CAPACITY_REGISTER_MAP_VERSION, decode_register_map_version + from dau_core.registers import JOB_CAPACITY_REGISTER_MAP_VERSION + + from .capability_decode import decode_register_map_version version = decode_register_map_version(device.read_register(RegisterOffset.REGISTER_MAP_VERSION)) if version < JOB_CAPACITY_REGISTER_MAP_VERSION: diff --git a/dau_driver/status.py b/dau_driver/status.py index e55fe2d..e7424be 100644 --- a/dau_driver/status.py +++ b/dau_driver/status.py @@ -2,33 +2,13 @@ from dataclasses import dataclass -LINK_READY_BIT = 1 << 0 -MIG_READY_BIT = 1 << 1 -OPERATOR_READY_BIT = 1 << 2 -RESET_ACTIVE_BIT = 1 << 3 - @dataclass(frozen=True) class DeviceStatus: + # STATUS as read. No shell assigns its bits yet, so it is not decoded raw_status: int last_error: int - @property - def link_ready(self) -> bool: - return bool(self.raw_status & LINK_READY_BIT) - - @property - def mig_ready(self) -> bool: - return bool(self.raw_status & MIG_READY_BIT) - - @property - def operator_ready(self) -> bool: - return bool(self.raw_status & OPERATOR_READY_BIT) - - @property - def reset_active(self) -> bool: - return bool(self.raw_status & RESET_ACTIVE_BIT) - @dataclass(frozen=True) class DeviceCounters: diff --git a/dau_driver/stream_codec.py b/dau_driver/stream_codec.py index 8dbaa3a..e017fc5 100644 --- a/dau_driver/stream_codec.py +++ b/dau_driver/stream_codec.py @@ -6,6 +6,7 @@ from dau_core.stream import ( DEFAULT_RECORD_BATCH_LIMITS, + LOGICAL_TYPE_BYTE_WIDTHS, BatchHeader, ColumnDescriptor, InvalidStreamError, @@ -17,18 +18,6 @@ ) _ALIGNMENT_BYTES = 64 -_SUPPORTED_WIDTHS_BYTES = { - LogicalType.INT8: 1, - LogicalType.INT16: 2, - LogicalType.INT32: 4, - LogicalType.INT64: 8, - LogicalType.UINT8: 1, - LogicalType.UINT16: 2, - LogicalType.UINT32: 4, - LogicalType.UINT64: 8, - LogicalType.FLOAT32: 4, - LogicalType.FLOAT64: 8, -} _SCALAR_RESULT_STRUCTS = { LogicalType.INT8: struct.Struct(" None: def _validate_column(header: BatchHeader, column: ColumnDescriptor, encoded_length: int) -> None: _normalize_supported_type(column.logical_type) - expected_width_bits = _SUPPORTED_WIDTHS_BYTES[column.logical_type] * 8 + expected_width_bits = LOGICAL_TYPE_BYTE_WIDTHS[column.logical_type] * 8 if column.physical_width_bits != expected_width_bits: raise InvalidStreamError(f"physical_width_bits must be {expected_width_bits} for {column.logical_type.name.lower()}") if column.null_count != 0 or column.validity_length != 0: @@ -408,7 +397,7 @@ def _validate_column(header: BatchHeader, column: ColumnDescriptor, encoded_leng raise InvalidStreamError("column row_count does not match batch header") if column.payload_offset != header.payload_offset or column.payload_length != header.payload_length: raise InvalidStreamError("column payload window does not match batch header") - expected_payload_length = column.row_count * _SUPPORTED_WIDTHS_BYTES[column.logical_type] + expected_payload_length = column.row_count * LOGICAL_TYPE_BYTE_WIDTHS[column.logical_type] if column.payload_length != expected_payload_length: raise InvalidStreamError("column payload length does not match row_count") _validate_payload_window(column.payload_offset, column.payload_length, encoded_length) @@ -435,7 +424,7 @@ def _normalize_supported_type(logical_type: LogicalType) -> LogicalType: normalized_type = LogicalType(logical_type) except ValueError as exc: raise InvalidStreamError(f"unsupported logical type {logical_type!r}") from exc - if normalized_type not in _SUPPORTED_WIDTHS_BYTES: + if normalized_type not in LOGICAL_TYPE_BYTE_WIDTHS: raise InvalidStreamError(f"unsupported logical type {normalized_type.name.lower()}") return normalized_type diff --git a/dau_driver/tests/test_discovery.py b/dau_driver/tests/test_discovery.py index b8ac022..531b5f6 100644 --- a/dau_driver/tests/test_discovery.py +++ b/dau_driver/tests/test_discovery.py @@ -216,10 +216,7 @@ def test_device_decodes_status_and_counters_from_register_block(tmp_path: Path) status = device.read_status() counters = device.read_counters() - assert status.link_ready - assert status.mig_ready - assert status.operator_ready - assert not status.reset_active + assert status.raw_status == 0b0111 assert status.last_error == 0x42 assert counters.h2c_bytes == 0x1_0000_0040 assert counters.c2h_bytes == 0x2_0000_0080 diff --git a/dau_driver/tests/test_no_board_geometry_constants.py b/dau_driver/tests/test_no_board_geometry_constants.py index 69efc34..72bed95 100644 --- a/dau_driver/tests/test_no_board_geometry_constants.py +++ b/dau_driver/tests/test_no_board_geometry_constants.py @@ -30,9 +30,6 @@ # or RECORD shape -- fixed by the format, identical on every board. WIRE_SHAPE_CONSTANTS = { "DMA_ALIGN_BYTES": "the DMA engine's alignment, a property of the transfer format", - "GROUP_RECORD_BYTES": "bytes per grouped-aggregation record on the wire", - "BAR_ROW_BYTES": "bytes per bar input row on the wire", - "BAR_RECORD_BYTES": "bytes per bar output record on the wire", } # Constants that DO describe board or shell geometry and have not moved to diff --git a/dau_driver/tests/test_pipeline_execution.py b/dau_driver/tests/test_pipeline_execution.py index 28737b9..7c47fd1 100644 --- a/dau_driver/tests/test_pipeline_execution.py +++ b/dau_driver/tests/test_pipeline_execution.py @@ -14,6 +14,7 @@ import pytest from dau_core.operators import BarAggregate, GroupAggregate +from dau_core.records import BAR_RECORD_BYTES, GROUP_RECORD_BYTES from dau_core.registers import ( JobStatusFlag, NocLaneRegisterOffset, @@ -24,8 +25,6 @@ ) from dau_driver.pipeline_execution import ( - BAR_RECORD_BYTES, - GROUP_RECORD_BYTES, PipelineJobError, bar_noc_lane_output_address, decode_bar_records, @@ -387,35 +386,6 @@ def test_execute_bar_noc_packed_error(tmp_path): assert str(exc.value) == "bar NoC job failed: status=0xc last_error=1" -def test_the_bar_layout_has_one_declaration() -> None: - """dau-polars builds numpy dtypes over these same bytes. When the layout - was declared in both places, a padding or field-order change could make - the two halves read one bitstream differently — so the derived formats - must reproduce the wire shapes exactly, and the byte sizes the job - contract advertises must follow from the fields rather than sit beside - them.""" - import struct as _struct - - from dau_driver.pipeline_execution import ( - BAR_RECORD_BYTES, - BAR_RECORD_FIELDS, - BAR_ROW_BYTES, - BAR_ROW_FIELDS, - numpy_dtype_fields, - struct_format, - ) - - assert struct_format(BAR_ROW_FIELDS) == " None: """A wide design keeps the high half across a LOW write, so a host that wrote only LOW would inherit whatever the previous job left there and diff --git a/docs/wiki/reference/Device-and-Discovery.md b/docs/wiki/reference/Device-and-Discovery.md index f5bc753..6775205 100644 --- a/docs/wiki/reference/Device-and-Discovery.md +++ b/docs/wiki/reference/Device-and-Discovery.md @@ -169,8 +169,8 @@ direction's channel count is zero. `DauDevice` passes its own `memory_bytes`. | `raw_status` | `int` | | `last_error` | `int` | -Properties decoding `raw_status`: `link_ready` (bit 0), `mig_ready` (bit 1), -`operator_ready` (bit 2), `reset_active` (bit 3). +`raw_status` is the STATUS word as read. No shell assigns its bits yet, so it +is not decoded into named flags. ### `DeviceCounters` @@ -196,7 +196,8 @@ hardware results. Host opcodes must never be submitted to the device. The composed-capability words (`LANE_COUNT`, `HOST_OPCODE_BITMAP`, `SORT_CAPACITY`) are only read on 0.2 and above; all-ones is normalized to not-advertised. The job-capacity words (`INPUT_ROW_CAPACITY`, `OUTPUT_RECORD_CAPACITY`) need 0.3 and -above — `JOB_CAPACITY_REGISTER_MAP_VERSION`. +above. Both gates, `CAPABILITY_WORDS_REGISTER_MAP_VERSION` and +`JOB_CAPACITY_REGISTER_MAP_VERSION`, are defined in `dau_core.registers`. ## Storage diff --git a/docs/wiki/reference/Execution-Contracts.md b/docs/wiki/reference/Execution-Contracts.md index 5cf6485..bff2899 100644 --- a/docs/wiki/reference/Execution-Contracts.md +++ b/docs/wiki/reference/Execution-Contracts.md @@ -257,9 +257,9 @@ identity. | `decode_group_records(data, count)` | 40-byte group records as `GroupAggregate` values. | | `execute_keyed_pipeline_job(device, rows, *, timeout_seconds=5.0, contract=None, max_rows=None)` | One batch through a configured pipeline. Returns `(group records, selected row count)`. | -Constants: `GROUP_RECORD_BYTES = 40`, `DMA_ALIGN_BYTES = 64`. These are **wire -and record shapes** — fixed by the format both sides compile against, not by -the board. +`DMA_ALIGN_BYTES = 64` is a **wire shape**: fixed by the transfer format, not by +the board. The record layouts and their sizes (`GROUP_RECORD_BYTES` is 40) are +defined in `dau_core.records`, next to the tiles that produce them. Capacity is not a constant. `device_job_capacity(device)` reads the input row and output record capacities the shell advertises; exceeding them raises @@ -270,10 +270,12 @@ module-level constants exactly, in both directions. ### Bar jobs -The bar wire layout is declared once, in `BAR_ROW_FIELDS` and -`BAR_RECORD_FIELDS`, with `struct_format(fields)` and `numpy_dtype_fields(fields)` -deriving the two consumers' views so a padding, signedness, or field-order change -cannot make them read one bitstream differently. +The bar wire layout is declared once, in `dau_core.records` as `BAR_ROW_FIELDS` +and `BAR_RECORD_FIELDS`, with `struct_format(fields)` and +`numpy_dtype_fields(fields)` deriving the two consumers' views so a padding, +signedness, or field-order change cannot make them read one bitstream +differently. dau-core's suite checks each layout's size against the tile's +registry entry. | Function | Behaviour | | ------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | @@ -286,7 +288,7 @@ cannot make them read one bitstream differently. | `execute_bar_ddr_packed(device, payload, row_count, *, staging, timeout_seconds=60.0)` | Stage, start, finish. | | `bar_ddr_max_rows_per_job(staging)` | The row budget the given staging layout supports. | -Constants: `BAR_ROW_BYTES = 16`, `BAR_RECORD_BYTES = 48`. Both are wire shapes. +`BAR_ROW_BYTES` (16) and `BAR_RECORD_BYTES` (48) come from the same module. `staging` is the board's `DdrStaging` — its input regions, output region, sizes, and alignment — passed in from the composed platform. It is a keyword argument diff --git a/docs/wiki/tutorial/First-Device.md b/docs/wiki/tutorial/First-Device.md index 9be959a..c34a3a0 100644 --- a/docs/wiki/tutorial/First-Device.md +++ b/docs/wiki/tutorial/First-Device.md @@ -86,10 +86,7 @@ transport.registers[int(RegisterOffset.BATCHES)] = 12 print("magic:", hex(device.read_register(RegisterOffset.MAGIC))) status = device.read_status() -print("link_ready: ", status.link_ready) -print("mig_ready: ", status.mig_ready) -print("operator_ready:", status.operator_ready) -print("reset_active: ", status.reset_active) +print("status:", bin(status.raw_status), "last error:", status.last_error) print(device.read_counters()) ``` @@ -98,16 +95,13 @@ print(device.read_counters()) ```text magic: 0x44415531 -link_ready: True -mig_ready: True -operator_ready: True -reset_active: False +status: 0b111 last error: 0 DeviceCounters(h2c_bytes=0, c2h_bytes=0, rows=1000000, batches=12, stalls=0) ``` `0x44415531` is the DAU magic word — it is how discovery tells a DAU shell from -any other XDMA bitstream. `DeviceStatus` decodes the status word into named bits, -and the counters are 64-bit values read as two 32-bit register accesses. +any other XDMA bitstream. `DeviceStatus` holds the status word as read, and the +counters are 64-bit values read as two 32-bit register accesses. ## Step 4 — Check the register path is healthy diff --git a/examples/ohlcv_pipeline_mm_job.py b/examples/ohlcv_pipeline_mm_job.py index cee21a9..b7bbf0c 100644 --- a/examples/ohlcv_pipeline_mm_job.py +++ b/examples/ohlcv_pipeline_mm_job.py @@ -28,13 +28,13 @@ grouped_aggregate_rows, map_rows, ) +from dau_core.records import GROUP_RECORD_BYTES from dau_core.registers import JobControlFlag, JobStatusFlag, RegisterOffset from dau_driver import discover_devices INPUT_ADDRESS = 0x0000_0000 OUTPUT_ADDRESS = 0x0010_0000 -GROUP_RECORD_BYTES = 40 DMA_ALIGN = 64 PREDICATE = FilterPredicate(op0=PredicateOp.GT, constant0=0)