| # Copyright 2026 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Driver-shaped public Python programming model (Spec Sections 6.1, 9; Milestone H4). |
| |
| Provides ergonomic, capability-aware access to hardware resources on Fuchsia targets. |
| Preserves production FIDL method boundaries, MMIO access width and ordering, |
| asynchronous waits, and the distinction between semantic device protocols and |
| raw registers. |
| """ |
| |
| from __future__ import annotations |
| |
| import asyncio |
| import dataclasses |
| from collections.abc import Awaitable, Callable, Mapping, Sequence |
| from typing import Any |
| |
| from driver_lab.models import InterruptOutcome, ResourceKind |
| from driver_lab.transport import ( |
| Denial, |
| DirectDescription, |
| DirectSession, |
| FidlCallOutcome, |
| OperationDenied, |
| PollOutcome, |
| ProxyDescription, |
| ProxySession, |
| ResourceInfo, |
| SequenceItem, |
| SequenceOutcome, |
| SnapshotItem, |
| WriteOutcome, |
| ) |
| |
| # Aliases matching both spec and transport terminology |
| WriteResult = WriteOutcome |
| PollResult = PollOutcome |
| |
| |
| class UnsupportedCapabilityError(Exception): |
| """Raised when an operation requires capabilities not supported by the session mode.""" |
| |
| |
| @dataclasses.dataclass(frozen=True) |
| class TranslationMetadata: |
| """Documents C++ and Rust driver analogues for host Python operations (Spec Section 9.1).""" |
| |
| cpp_analogue: str |
| rust_analogue: str |
| directly_translatable: bool |
| differences: str |
| target_local_timing: bool |
| experiment_only: bool = False |
| |
| |
| @dataclasses.dataclass(frozen=True) |
| class SessionCapabilities: |
| """Explicitly reported session capabilities and degradation guarantees (Spec Section 6.2).""" |
| |
| mode: str # "proxy" or "direct" |
| target_policy: bool |
| target_audit: bool |
| target_local_timing: bool |
| fault_isolation: str # "driver_host" | "none" | "process" |
| production_driver_active: bool |
| restoration_required: bool = False |
| resources: tuple[str, ...] = () |
| protocols: tuple[str, ...] = () |
| |
| def check_support(self, feature: str) -> None: |
| """Fails closed if the requested feature is not supported by the current session mode.""" |
| if feature == "mmio" and self.mode not in ("proxy", "in-situ"): |
| raise UnsupportedCapabilityError( |
| f"MMIO register operations are only supported in proxy or in-situ mode (current mode: {self.mode})" |
| ) |
| if feature == "sequence" and not self.target_local_timing: |
| raise UnsupportedCapabilityError( |
| "Target-local sequence execution is not supported in direct mode" |
| ) |
| if feature == "target_audit" and not self.target_audit: |
| raise UnsupportedCapabilityError( |
| "Target audit logging is not supported in direct mode" |
| ) |
| if feature == "target_policy" and not self.target_policy: |
| raise UnsupportedCapabilityError( |
| "Target policy enforcement is not supported in direct mode" |
| ) |
| |
| |
| @dataclasses.dataclass(frozen=True) |
| class AccessRequirements: |
| """Declared requirements for connecting or attaching to a hardware node (Spec Section 7.3).""" |
| |
| needs_mmio: bool = False |
| needs_sequence: bool = False |
| needs_target_timing: bool = False |
| needs_target_policy: bool = False |
| needs_target_audit: bool = False |
| is_mutating: bool = False |
| protocol: str | None = None |
| |
| |
| class MmioRegion: |
| """Driver-shaped interface to a memory-mapped I/O register block (Spec Sections 6.1, 9). |
| |
| Provides typed 32-bit register reads, masked writes with preconditions and readback, |
| polling, and bounded snapshots. |
| """ |
| |
| read32_metadata = TranslationMetadata( |
| cpp_analogue="mmio.Read32(offset) / fdf::MmioBuffer::Read32(offset)", |
| rust_analogue="mmio.read32(offset)", |
| directly_translatable=True, |
| differences="Executed over FIDL proxy channel with host round-trip latency and target audit logging.", |
| target_local_timing=False, |
| experiment_only=False, |
| ) |
| |
| write32_metadata = TranslationMetadata( |
| cpp_analogue="mmio.Write32(value, offset) / mmio.ModifyBits32(value, mask, offset)", |
| rust_analogue="mmio.write32(offset, value) / mmio.modify32(offset, ...)", |
| directly_translatable=True, |
| differences="Executed over FIDL proxy channel with host round trip, optional precondition check, and automatic readback.", |
| target_local_timing=False, |
| experiment_only=False, |
| ) |
| |
| poll32_metadata = TranslationMetadata( |
| cpp_analogue="hwreg::RegisterAddr<...>::ReadFrom(&mmio).Poll(...) or loop with zx::nanosleep", |
| rust_analogue="polling loop with fuchsia_async::Timer", |
| directly_translatable=False, |
| differences="Target-local polling in proxy driver dispatcher without host round trips per attempt.", |
| target_local_timing=True, |
| experiment_only=False, |
| ) |
| |
| snapshot32_metadata = TranslationMetadata( |
| cpp_analogue="bounded read loop", |
| rust_analogue="bounded read loop", |
| directly_translatable=False, |
| differences="Convenience batched read operation for experiment snapshotting.", |
| target_local_timing=False, |
| experiment_only=True, |
| ) |
| |
| def __init__( |
| self, |
| resource: ResourceInfo, |
| session: ProxySession, |
| audit_drainer: Callable[[], Awaitable[None]] | None = None, |
| ) -> None: |
| self._resource = resource |
| self._session = session |
| self._audit_drainer = audit_drainer |
| |
| @property |
| def name(self) -> str: |
| """Name of the logical resource.""" |
| return self._resource.name |
| |
| @property |
| def id(self) -> int: |
| """Numeric ID of the resource.""" |
| return self._resource.id |
| |
| @property |
| def logical_size(self) -> int: |
| """Logical size in bytes.""" |
| return self._resource.logical_size |
| |
| @property |
| def digest(self) -> str: |
| """Resource digest.""" |
| return self._resource.digest |
| |
| def _validate_offset(self, offset: int) -> None: |
| if offset < 0 or offset + 4 > self._resource.logical_size: |
| raise ValueError( |
| f"Offset {hex(offset)} out of logical bounds [0, {hex(self._resource.logical_size)}) for {self.name}" |
| ) |
| if offset % 4 != 0: |
| raise ValueError( |
| f"Misaligned 32-bit register access at offset {hex(offset)} (must be 4-byte aligned)" |
| ) |
| |
| async def read32(self, offset: int) -> int: |
| """Reads a 32-bit register at byte offset.""" |
| self._validate_offset(offset) |
| outcome = await self._session.read32(self.id, offset) |
| return outcome.value |
| |
| async def write32( |
| self, |
| offset: int, |
| value: int, |
| *, |
| mask: int = 0xFFFF_FFFF, |
| expected_before: int | None = None, |
| expected_mask: int | None = 0xFFFF_FFFF, |
| require_readback: bool = True, |
| ) -> WriteOutcome: |
| """Writes a 32-bit register with optional mask, precondition, and readback.""" |
| self._validate_offset(offset) |
| precondition_tuple: tuple[int, int] | None = None |
| if expected_before is not None: |
| precondition_tuple = ( |
| expected_before, |
| expected_mask if expected_mask is not None else 0xFFFF_FFFF, |
| ) |
| outcome = await self._session.write32( |
| resource=self.id, |
| offset=offset, |
| value=value, |
| write_mask=mask, |
| precondition=precondition_tuple, |
| readback=require_readback, |
| ) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return outcome |
| |
| async def poll32( |
| self, |
| offset: int, |
| *, |
| expected: int, |
| mask: int = 0xFFFF_FFFF, |
| interval_s: float = 0.001, |
| timeout_s: float = 1.0, |
| ) -> PollOutcome: |
| """Polls a 32-bit register until (value & mask) == (expected & mask) or timeout.""" |
| self._validate_offset(offset) |
| interval_ns = max(1, int(interval_s * 1e9)) |
| timeout_ns = max(1, int(timeout_s * 1e9)) |
| outcome = await self._session.poll32( |
| resource=self.id, |
| offset=offset, |
| expected=expected, |
| mask=mask, |
| interval_ns=interval_ns, |
| timeout_ns=timeout_ns, |
| ) |
| return outcome |
| |
| async def snapshot32(self, offsets: Sequence[int]) -> list[int]: |
| """Reads multiple 32-bit registers in a single bounded batch.""" |
| for off in offsets: |
| self._validate_offset(off) |
| items = [SnapshotItem(resource=self.id, offset=off) for off in offsets] |
| outcome = await self._session.snapshot(items) |
| if not outcome.complete: |
| raise OperationDenied(Denial.BACKEND_FAULT) |
| return [r.value for r in outcome.results if r.ok] |
| |
| |
| class StateHandle: |
| """Driver-shaped interface to a software state, runtime knob, and trigger bank (Phase 3 / CS32). |
| |
| Wraps an underlying 32-bit `StateBank` resource (`MmioRegion`) with semantic methods for: |
| - `.read32(offset)` for reading state slots and read probes |
| - `.poll32(offset, expected, mask)` for polling software state transitions |
| - `.write_knob(offset, value)` for tuning runtime parameters with readback |
| - `.trigger(offset, arg=1)` for invoking deterministic self-tests and reading back their result |
| """ |
| |
| read32_metadata = TranslationMetadata( |
| cpp_analogue="state_slot.load(std::memory_order_seq_cst)", |
| rust_analogue="state_slot.load()", |
| directly_translatable=False, |
| differences="In-situ host observation of a driver StateBank slot or read probe over FIDL.", |
| target_local_timing=False, |
| experiment_only=True, |
| ) |
| |
| write_knob_metadata = TranslationMetadata( |
| cpp_analogue="knob_slot.store(value, std::memory_order_seq_cst)", |
| rust_analogue="knob_slot.store(value)", |
| directly_translatable=False, |
| differences="In-situ host write to a driver StateBank runtime tuning knob with readback.", |
| target_local_timing=False, |
| experiment_only=True, |
| ) |
| |
| trigger_metadata = TranslationMetadata( |
| cpp_analogue="trigger_callback(arg)", |
| rust_analogue="trigger_callback(arg) -> Result<u32, BackendError>", |
| directly_translatable=False, |
| differences="In-situ host invocation of a synchronous driver StateBank trigger callback returning its status via readback.", |
| target_local_timing=True, |
| experiment_only=True, |
| ) |
| |
| def __init__(self, region: MmioRegion) -> None: |
| self._region = region |
| |
| @property |
| def name(self) -> str: |
| """Name of the logical state bank resource.""" |
| return self._region.name |
| |
| @property |
| def id(self) -> int: |
| """Numeric ID of the state bank resource.""" |
| return self._region.id |
| |
| @property |
| def logical_size(self) -> int: |
| """Logical size in bytes.""" |
| return self._region.logical_size |
| |
| @property |
| def digest(self) -> str: |
| """Resource digest.""" |
| return self._region.digest |
| |
| async def read32(self, offset: int) -> int: |
| """Reads a 32-bit software state slot or read probe at byte offset.""" |
| return await self._region.read32(offset) |
| |
| async def poll32( |
| self, |
| offset: int, |
| expected: int = 0, |
| mask: int = 0xFFFF_FFFF, |
| *, |
| interval_s: float = 0.001, |
| timeout_s: float = 1.0, |
| ) -> PollOutcome: |
| """Polls a 32-bit software state slot until (value & mask) == (expected & mask) or timeout.""" |
| return await self._region.poll32( |
| offset, |
| expected=expected, |
| mask=mask, |
| interval_s=interval_s, |
| timeout_s=timeout_s, |
| ) |
| |
| async def write_knob( |
| self, |
| offset: int, |
| value: int, |
| *, |
| mask: int = 0xFFFF_FFFF, |
| expected_before: int | None = None, |
| expected_mask: int | None = 0xFFFF_FFFF, |
| require_readback: bool = True, |
| ) -> WriteOutcome: |
| """Writes a 32-bit runtime tuning knob with automatic readback.""" |
| return await self._region.write32( |
| offset, |
| value, |
| mask=mask, |
| expected_before=expected_before, |
| expected_mask=expected_mask, |
| require_readback=require_readback, |
| ) |
| |
| async def trigger( |
| self, |
| offset: int, |
| arg: int = 1, |
| *, |
| mask: int = 0xFFFF_FFFF, |
| expected_before: int | None = None, |
| expected_mask: int | None = 0xFFFF_FFFF, |
| require_readback: bool = False, |
| ) -> WriteOutcome: |
| """Invokes a deterministic trigger slot at `offset` with `arg` without performing an extra readback.""" |
| return await self._region.write32( |
| offset, |
| arg, |
| mask=mask, |
| expected_before=expected_before, |
| expected_mask=expected_mask, |
| require_readback=require_readback, |
| ) |
| |
| async def snapshot32(self, offsets: Sequence[int]) -> list[int]: |
| """Reads multiple 32-bit state slots in a single bounded batch.""" |
| return await self._region.snapshot32(offsets) |
| |
| |
| class ProtocolProxy: |
| """Direct published FIDL protocol client adapter.""" |
| |
| def __init__(self, session: DirectSession, protocol_name: str) -> None: |
| self._session = session |
| self._protocol_name = protocol_name |
| |
| @property |
| def protocol_name(self) -> str: |
| return self._protocol_name |
| |
| async def call( |
| self, method: str, args: Mapping[str, object] | None = None |
| ) -> FidlCallOutcome: |
| """Invokes a method on the published protocol.""" |
| return await self._session.call_fidl(method, args) |
| |
| def __getattr__(self, name: str) -> Any: |
| async def _caller(**kwargs: object) -> FidlCallOutcome: |
| return await self.call(name, kwargs) |
| |
| return _caller |
| |
| |
| class Gpio: |
| """Thin shape-preserving adapter for GPIO pin protocol (Spec Sections 8.1, 9.5, 12).""" |
| |
| read_metadata = TranslationMetadata( |
| cpp_analogue="gpio.Read() / ddk::GpioProtocolClient::Read() / fuchsia::hardware::pin::Pin::Read()", |
| rust_analogue="gpio.read().await / fuchsia_hardware_pin::PinProxy::read()", |
| directly_translatable=True, |
| differences="Target-side GPIO read mediated by proxy driver with audit log entry and policy check.", |
| target_local_timing=False, |
| experiment_only=False, |
| ) |
| |
| write_metadata = TranslationMetadata( |
| cpp_analogue="gpio.Write(value) / ddk::GpioProtocolClient::Write(value) / fuchsia::hardware::pin::Pin::SetBufferMode()", |
| rust_analogue="gpio.write(value).await / fuchsia_hardware_pin::PinProxy::set_buffer_mode()", |
| directly_translatable=True, |
| differences="Target-side GPIO write mediated by proxy driver with audit log entry and policy check.", |
| target_local_timing=False, |
| experiment_only=False, |
| ) |
| |
| def __init__( |
| self, |
| session: DirectSession | ProxySession, |
| resource_name: str = "gpio", |
| resource_info: ResourceInfo | None = None, |
| audit_drainer: Callable[[], Awaitable[None]] | None = None, |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| self._resource_info = resource_info |
| self._audit_drainer = audit_drainer |
| |
| @property |
| def name(self) -> str: |
| return self._resource_name |
| |
| @property |
| def id(self) -> int | None: |
| return ( |
| self._resource_info.id if self._resource_info is not None else None |
| ) |
| |
| async def read(self) -> bool: |
| if self._resource_info is not None and hasattr( |
| self._session, "gpio_read" |
| ): |
| outcome = await self._session.gpio_read(self._resource_info.id) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return outcome.value |
| if hasattr(self._session, "call_fidl"): |
| outcome = await self._session.call_fidl( |
| "read", {"resource": self._resource_name} |
| ) |
| return bool(outcome.response.get("value", False)) |
| raise UnsupportedCapabilityError("Session does not support GPIO read") |
| |
| async def write(self, value: bool) -> None: |
| if self._resource_info is not None and hasattr( |
| self._session, "gpio_write" |
| ): |
| await self._session.gpio_write(self._resource_info.id, value) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return |
| if hasattr(self._session, "call_fidl"): |
| await self._session.call_fidl( |
| "write", {"resource": self._resource_name, "value": value} |
| ) |
| return |
| raise UnsupportedCapabilityError("Session does not support GPIO write") |
| |
| async def set_direction(self, direction: str) -> None: |
| if hasattr(self._session, "call_fidl"): |
| await self._session.call_fidl( |
| "set_direction", |
| {"resource": self._resource_name, "direction": direction}, |
| ) |
| return |
| raise UnsupportedCapabilityError( |
| "GPIO set_direction is only supported in direct mode" |
| ) |
| |
| |
| class I2c: |
| """Thin shape-preserving adapter for I2C bus protocol (Spec Sections 8.1, 9.5, 12).""" |
| |
| transfer_metadata = TranslationMetadata( |
| cpp_analogue="i2c.Transfer(...) / fuchsia::hardware::i2c::Device::Transfer()", |
| rust_analogue="i2c.transfer(...).await / fuchsia_hardware_i2c::DeviceProxy::transfer()", |
| directly_translatable=True, |
| differences="I2C transaction executed by target proxy driver via underlying FIDL/DDK protocol with bounded payload and audit logging.", |
| target_local_timing=False, |
| experiment_only=False, |
| ) |
| |
| def __init__( |
| self, |
| session: DirectSession | ProxySession, |
| resource_name: str = "i2c", |
| resource_info: ResourceInfo | None = None, |
| audit_drainer: Callable[[], Awaitable[None]] | None = None, |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| self._resource_info = resource_info |
| self._audit_drainer = audit_drainer |
| |
| @property |
| def name(self) -> str: |
| return self._resource_name |
| |
| @property |
| def id(self) -> int | None: |
| return ( |
| self._resource_info.id if self._resource_info is not None else None |
| ) |
| |
| async def transfer(self, write_data: bytes, read_length: int = 0) -> bytes: |
| if self._resource_info is not None and hasattr( |
| self._session, "i2c_transfer" |
| ): |
| outcome = await self._session.i2c_transfer( |
| self._resource_info.id, |
| write_data=write_data, |
| read_length=read_length, |
| ) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return outcome.read_data |
| if hasattr(self._session, "call_fidl"): |
| outcome = await self._session.call_fidl( |
| "transfer", |
| { |
| "resource": self._resource_name, |
| "write_data": list(write_data), |
| "read_length": read_length, |
| }, |
| ) |
| data = outcome.response.get("read_data", b"") |
| if isinstance(data, (bytes, bytearray)): |
| return bytes(data) |
| if isinstance(data, list): |
| return bytes(data) |
| return b"" |
| raise UnsupportedCapabilityError( |
| "Session does not support I2C transfer" |
| ) |
| |
| |
| class Spi: |
| """Thin shape-preserving adapter for SPI bus protocol (Spec Sections 8.1, 9.5, 12).""" |
| |
| transmit_metadata = TranslationMetadata( |
| cpp_analogue="spi.Transmit(...) / fuchsia::hardware::spi::Device::Transmit()", |
| rust_analogue="spi.transmit(...).await / fuchsia_hardware_spi::DeviceProxy::transmit()", |
| directly_translatable=True, |
| differences="SPI full-duplex / transmit transaction executed by target proxy driver with bounded payload and audit logging.", |
| target_local_timing=False, |
| experiment_only=False, |
| ) |
| |
| def __init__( |
| self, |
| session: DirectSession | ProxySession, |
| resource_name: str = "spi", |
| resource_info: ResourceInfo | None = None, |
| audit_drainer: Callable[[], Awaitable[None]] | None = None, |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| self._resource_info = resource_info |
| self._audit_drainer = audit_drainer |
| |
| @property |
| def name(self) -> str: |
| return self._resource_name |
| |
| @property |
| def id(self) -> int | None: |
| return ( |
| self._resource_info.id if self._resource_info is not None else None |
| ) |
| |
| async def transmit(self, tx_data: bytes) -> bytes: |
| if self._resource_info is not None and hasattr( |
| self._session, "spi_transmit" |
| ): |
| outcome = await self._session.spi_transmit( |
| self._resource_info.id, |
| tx_data=tx_data, |
| ) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return outcome.rx_data |
| if hasattr(self._session, "call_fidl"): |
| outcome = await self._session.call_fidl( |
| "transmit", |
| { |
| "resource": self._resource_name, |
| "tx_data": list(tx_data), |
| }, |
| ) |
| data = outcome.response.get("rx_data", b"") |
| if isinstance(data, (bytes, bytearray)): |
| return bytes(data) |
| if isinstance(data, list): |
| return bytes(data) |
| return b"" |
| raise UnsupportedCapabilityError( |
| "Session does not support SPI transmit" |
| ) |
| |
| |
| class Serial: |
| """Thin shape-preserving adapter for serial stream protocol.""" |
| |
| def __init__( |
| self, session: DirectSession, resource_name: str = "serial" |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| |
| async def write(self, data: bytes) -> int: |
| outcome = await self._session.call_fidl( |
| "write", |
| {"resource": self._resource_name, "data": list(data)}, |
| ) |
| val = outcome.response.get("bytes_written") |
| if isinstance(val, int): |
| return val |
| return len(data) |
| |
| async def read(self, max_bytes: int) -> bytes: |
| outcome = await self._session.call_fidl( |
| "read", |
| {"resource": self._resource_name, "max_bytes": max_bytes}, |
| ) |
| data = outcome.response.get("data", b"") |
| if isinstance(data, (bytes, bytearray)): |
| return bytes(data) |
| if isinstance(data, list): |
| return bytes(data) |
| return b"" |
| |
| |
| class Clock: |
| """Thin shape-preserving adapter for clock control protocol.""" |
| |
| def __init__( |
| self, session: DirectSession, resource_name: str = "clock" |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| |
| async def enable(self) -> None: |
| await self._session.call_fidl( |
| "enable", {"resource": self._resource_name} |
| ) |
| |
| async def disable(self) -> None: |
| await self._session.call_fidl( |
| "disable", {"resource": self._resource_name} |
| ) |
| |
| |
| class Reset: |
| """Thin shape-preserving adapter for reset control protocol.""" |
| |
| def __init__( |
| self, session: DirectSession, resource_name: str = "reset" |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| |
| async def assert_reset(self) -> None: |
| await self._session.call_fidl( |
| "assert_reset", {"resource": self._resource_name} |
| ) |
| |
| async def deassert_reset(self) -> None: |
| await self._session.call_fidl( |
| "deassert_reset", {"resource": self._resource_name} |
| ) |
| |
| |
| class Interrupt: |
| """Thin shape-preserving adapter for interrupt observation (Spec Sections 11.7, 17).""" |
| |
| wait_metadata = TranslationMetadata( |
| cpp_analogue="zx::interrupt::wait(×tamp) / fdf::WireAsyncEventHandler", |
| rust_analogue="fuchsia_async::OnSignals::new(&interrupt, zx::Signals::INTERRUPT).await", |
| directly_translatable=True, |
| differences="Bounded hanging request to target proxy with sequence tracking, coalescing metrics, and target monotonic timestamping.", |
| target_local_timing=True, |
| experiment_only=False, |
| ) |
| |
| def __init__( |
| self, |
| session: DirectSession | ProxySession, |
| resource_name: str = "interrupt", |
| resource_info: ResourceInfo | None = None, |
| audit_drainer: Callable[[], Awaitable[None]] | None = None, |
| ) -> None: |
| self._session = session |
| self._resource_name = resource_name |
| self._resource_info = resource_info |
| self._audit_drainer = audit_drainer |
| |
| @property |
| def name(self) -> str: |
| return self._resource_name |
| |
| @property |
| def id(self) -> int | None: |
| return ( |
| self._resource_info.id if self._resource_info is not None else None |
| ) |
| |
| async def wait( |
| self, |
| timeout_s: float = 1.0, |
| after_sequence: int = 0, |
| ) -> InterruptOutcome: |
| """Observes an interrupt event (Spec Sections 11.7, 17).""" |
| if self._resource_info is not None and isinstance( |
| self._session, ProxySession |
| ): |
| outcome = await self._session.wait_for_interrupt( |
| self._resource_info.id, |
| after_sequence=after_sequence, |
| timeout_s=timeout_s, |
| ) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return outcome |
| |
| if isinstance(self._session, DirectSession): |
| outcome_fidl = await self._session.call_fidl( |
| "wait", |
| { |
| "resource": self._resource_name, |
| "timeout_ns": int(timeout_s * 1e9), |
| }, |
| ) |
| val = outcome_fidl.response.get("timestamp_ns") |
| ts = val if isinstance(val, int) else 0 |
| return InterruptOutcome( |
| resource=self._resource_info.id if self._resource_info else 0, |
| sequence=1, |
| count=1, |
| timestamp_ns=ts, |
| coalesced_count=0, |
| ) |
| raise RuntimeError("Session does not support wait_for_interrupt") |
| |
| |
| class HardwareSession: |
| """Unified driver-shaped session facade for hardware exploration (Spec Sections 6.1, 9).""" |
| |
| def __init__( |
| self, |
| *, |
| capabilities: SessionCapabilities, |
| proxy_session: ProxySession | None = None, |
| direct_session: DirectSession | None = None, |
| proxy_description: ProxyDescription | None = None, |
| direct_description: DirectDescription | None = None, |
| audit_drainer: Callable[[], Awaitable[None]] | None = None, |
| ) -> None: |
| self._capabilities = capabilities |
| self._proxy_session = proxy_session |
| self._direct_session = direct_session |
| self._proxy_description = proxy_description |
| self._direct_description = direct_description |
| self._audit_drainer = audit_drainer |
| self._closed = False |
| |
| @property |
| def capabilities(self) -> SessionCapabilities: |
| """The explicit capabilities and degradation guarantees of this session.""" |
| return self._capabilities |
| |
| @property |
| def mode(self) -> str: |
| """Session mode ('proxy' or 'direct').""" |
| return self._capabilities.mode |
| |
| @property |
| def is_closed(self) -> bool: |
| """Whether this session has been closed.""" |
| return self._closed |
| |
| def _check_not_closed(self) -> None: |
| if self._closed: |
| raise UnsupportedCapabilityError("HardwareSession is closed") |
| |
| async def mmio(self, resource: str) -> MmioRegion: |
| """Acquires a named MMIO region. Fails closed in direct mode.""" |
| self._check_not_closed() |
| self._capabilities.check_support("mmio") |
| if self._proxy_session is None or self._proxy_description is None: |
| raise UnsupportedCapabilityError("Proxy session is not available") |
| info = self._proxy_description.resource_named(resource) |
| if info is None: |
| available = [r.name for r in self._proxy_description.resources] |
| raise ValueError( |
| f"Unknown MMIO resource '{resource}'. Available: {available}" |
| ) |
| return MmioRegion( |
| resource=info, |
| session=self._proxy_session, |
| audit_drainer=self._audit_drainer, |
| ) |
| |
| async def state( |
| self, name: str = "state0", *, resource: str | None = None |
| ) -> StateHandle: |
| """Acquires a named StateBank resource handle (`StateHandle`). Fails closed in direct mode.""" |
| target_name = resource if resource is not None else name |
| region = await self.mmio(target_name) |
| return StateHandle(region) |
| |
| async def protocol(self, protocol_name: str) -> ProtocolProxy: |
| """Connects to a published FIDL protocol. Fails in proxy mode.""" |
| self._check_not_closed() |
| if self._direct_session is None: |
| raise UnsupportedCapabilityError( |
| "Direct published FIDL protocols are only supported in direct mode" |
| ) |
| return ProtocolProxy(self._direct_session, protocol_name) |
| |
| async def gpio(self, resource: str = "gpio") -> Gpio: |
| """Acquires a GPIO protocol adapter in proxy or direct mode.""" |
| self._check_not_closed() |
| if ( |
| self._proxy_session is not None |
| and self._proxy_description is not None |
| ): |
| info = self._proxy_description.resource_named(resource) |
| if info is None: |
| gpio_res = [ |
| r |
| for r in self._proxy_description.resources |
| if r.kind == ResourceKind.GPIO |
| ] |
| if resource == "gpio" and gpio_res: |
| info = gpio_res[0] |
| else: |
| available = [ |
| r.name for r in self._proxy_description.resources |
| ] |
| raise ValueError( |
| f"Unknown GPIO resource '{resource}'. Available: {available}" |
| ) |
| return Gpio( |
| self._proxy_session, |
| resource_name=info.name, |
| resource_info=info, |
| audit_drainer=self._audit_drainer, |
| ) |
| if self._direct_session is not None: |
| return Gpio(self._direct_session, resource) |
| raise UnsupportedCapabilityError( |
| "GPIO protocol is not supported in current session" |
| ) |
| |
| async def i2c(self, resource: str = "i2c") -> I2c: |
| """Acquires an I2C protocol adapter in proxy or direct mode.""" |
| self._check_not_closed() |
| if ( |
| self._proxy_session is not None |
| and self._proxy_description is not None |
| ): |
| info = self._proxy_description.resource_named(resource) |
| if info is None: |
| i2c_res = [ |
| r |
| for r in self._proxy_description.resources |
| if r.kind == ResourceKind.I2C |
| ] |
| if resource == "i2c" and i2c_res: |
| info = i2c_res[0] |
| else: |
| available = [ |
| r.name for r in self._proxy_description.resources |
| ] |
| raise ValueError( |
| f"Unknown I2C resource '{resource}'. Available: {available}" |
| ) |
| return I2c( |
| self._proxy_session, |
| resource_name=info.name, |
| resource_info=info, |
| audit_drainer=self._audit_drainer, |
| ) |
| if self._direct_session is not None: |
| return I2c(self._direct_session, resource) |
| raise UnsupportedCapabilityError( |
| "I2C protocol is not supported in current session" |
| ) |
| |
| async def spi(self, resource: str = "spi") -> Spi: |
| """Acquires a SPI protocol adapter in proxy or direct mode.""" |
| self._check_not_closed() |
| if ( |
| self._proxy_session is not None |
| and self._proxy_description is not None |
| ): |
| info = self._proxy_description.resource_named(resource) |
| if info is None: |
| spi_res = [ |
| r |
| for r in self._proxy_description.resources |
| if r.kind == ResourceKind.SPI |
| ] |
| if resource == "spi" and spi_res: |
| info = spi_res[0] |
| else: |
| available = [ |
| r.name for r in self._proxy_description.resources |
| ] |
| raise ValueError( |
| f"Unknown SPI resource '{resource}'. Available: {available}" |
| ) |
| return Spi( |
| self._proxy_session, |
| resource_name=info.name, |
| resource_info=info, |
| audit_drainer=self._audit_drainer, |
| ) |
| if self._direct_session is not None: |
| return Spi(self._direct_session, resource) |
| raise UnsupportedCapabilityError( |
| "SPI protocol is not supported in current session" |
| ) |
| |
| async def serial(self, resource: str = "serial") -> Serial: |
| """Acquires a Serial protocol adapter in direct mode.""" |
| self._check_not_closed() |
| if self._direct_session is None: |
| raise UnsupportedCapabilityError( |
| "Serial protocol is only supported in direct mode" |
| ) |
| return Serial(self._direct_session, resource) |
| |
| async def clock(self, resource: str = "clock") -> Clock: |
| """Acquires a Clock protocol adapter in direct mode.""" |
| self._check_not_closed() |
| if self._direct_session is None: |
| raise UnsupportedCapabilityError( |
| "Clock protocol is only supported in direct mode" |
| ) |
| return Clock(self._direct_session, resource) |
| |
| async def reset(self, resource: str = "reset") -> Reset: |
| """Acquires a Reset protocol adapter in direct mode.""" |
| self._check_not_closed() |
| if self._direct_session is None: |
| raise UnsupportedCapabilityError( |
| "Reset protocol is only supported in direct mode" |
| ) |
| return Reset(self._direct_session, resource) |
| |
| async def interrupt(self, resource: str = "interrupt") -> Interrupt: |
| """Acquires an Interrupt adapter in proxy or direct mode (Spec Sections 11.7, 17).""" |
| self._check_not_closed() |
| if self._proxy_session is not None: |
| if self._proxy_description is None: |
| raise UnsupportedCapabilityError("Missing proxy description") |
| info = None |
| for r in self._proxy_description.resources: |
| if ( |
| r.name == resource or str(r.id) == resource |
| ) and r.kind == ResourceKind.INTERRUPT: |
| info = r |
| break |
| if info is None: |
| irq_res = [ |
| r |
| for r in self._proxy_description.resources |
| if r.kind == ResourceKind.INTERRUPT |
| ] |
| if resource == "interrupt" and irq_res: |
| info = irq_res[0] |
| else: |
| available = [ |
| r.name for r in self._proxy_description.resources |
| ] |
| raise ValueError( |
| f"Unknown interrupt resource '{resource}'. Available: {available}" |
| ) |
| return Interrupt( |
| self._proxy_session, |
| resource_name=info.name, |
| resource_info=info, |
| audit_drainer=self._audit_drainer, |
| ) |
| if self._direct_session is not None: |
| return Interrupt(self._direct_session, resource) |
| raise UnsupportedCapabilityError( |
| "Interrupt observation is not supported in current session" |
| ) |
| |
| async def sequence( |
| self, operations: Sequence[SequenceItem] |
| ) -> SequenceOutcome: |
| """Executes a target-local sequence of operations (Spec Section 9).""" |
| self._check_not_closed() |
| self._capabilities.check_support("sequence") |
| if self._proxy_session is None: |
| raise UnsupportedCapabilityError("Proxy session is not available") |
| outcome = await self._proxy_session.execute_sequence(tuple(operations)) |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| return outcome |
| |
| async def close(self) -> None: |
| """Closes the session, drains remaining audit, and releases target resources.""" |
| if self._closed: |
| return |
| self._closed = True |
| try: |
| if self._audit_drainer is not None: |
| await self._audit_drainer() |
| except (asyncio.CancelledError, Exception): |
| pass |
| finally: |
| if self._proxy_session is not None: |
| await self._proxy_session.close() |
| if self._direct_session is not None: |
| await self._direct_session.close() |
| |
| async def __aenter__(self) -> "HardwareSession": |
| self._check_not_closed() |
| return self |
| |
| async def __aexit__( |
| self, |
| exc_type: type[BaseException] | None, |
| exc_val: BaseException | None, |
| exc_tb: Any, |
| ) -> None: |
| await self.close() |
| |
| |
| __all__ = [ |
| "AccessRequirements", |
| "Clock", |
| "Gpio", |
| "HardwareSession", |
| "I2c", |
| "Interrupt", |
| "MmioRegion", |
| "PollResult", |
| "ProtocolProxy", |
| "Reset", |
| "SequenceOutcome", |
| "Serial", |
| "SessionCapabilities", |
| "Spi", |
| "StateHandle", |
| "TranslationMetadata", |
| "UnsupportedCapabilityError", |
| "WriteResult", |
| ] |
