Update specification to mandate ABC over Protocol for maximum type safety

This commit is contained in:
2025-06-24 23:59:34 +00:00
parent 4778b275d5
commit 11a100832f

View File

@@ -450,109 +450,123 @@ High-level call → Driver → SCPI command → Transport → Instrument
```python ```python
# py_dvt_ate/instruments/interfaces.py # py_dvt_ate/instruments/interfaces.py
from typing import Protocol, runtime_checkable from abc import ABC, abstractmethod
@runtime_checkable @runtime_checkable
class IThermalChamber(Protocol): class IThermalChamber(ABC):
"""Hardware abstraction for thermal chambers.""" """Hardware abstraction for thermal chambers."""
@abstractmethod
def set_temperature(self, setpoint: float) -> None: def set_temperature(self, setpoint: float) -> None:
"""Set target temperature in degrees Celsius.""" """[docstring]"""
... pass
@abstractmethod
def get_temperature(self) -> float: def get_temperature(self) -> float:
"""Get current actual temperature in degrees Celsius.""" """[docstring]"""
... pass
@abstractmethod
def get_setpoint(self) -> float: def get_setpoint(self) -> float:
"""Get current temperature setpoint.""" """[docstring]"""
... pass
@abstractmethod
def is_stable(self) -> bool: def is_stable(self) -> bool:
"""Check if temperature has stabilised at setpoint.""" """[docstring]"""
... pass
@abstractmethod
def wait_until_stable( def wait_until_stable(
self, self,
timeout: float = 300.0, timeout: float = 300.0,
poll_interval: float = 1.0 poll_interval: float = 1.0
) -> bool: ) -> bool:
""" """[docstring]"""
Block until temperature stabilises or timeout. pass
Returns:
True if stable, False if timeout
"""
...
@abstractmethod
def set_ramp_rate(self, rate: float) -> None: def set_ramp_rate(self, rate: float) -> None:
"""Set temperature ramp rate in degrees C per minute.""" """[docstring]"""
... pass
@runtime_checkable @runtime_checkable
class IPowerSupply(Protocol): class IPowerSupply(ABC):
"""Hardware abstraction for programmable power supplies.""" """Hardware abstraction for programmable power supplies."""
@abstractmethod
def set_voltage(self, channel: int, voltage: float) -> None: def set_voltage(self, channel: int, voltage: float) -> None:
"""Set output voltage for specified channel.""" """[docstring]"""
... pass
@abstractmethod
def get_voltage(self, channel: int) -> float: def get_voltage(self, channel: int) -> float:
"""Get voltage setpoint for specified channel.""" """[docstring]"""
... pass
@abstractmethod
def set_current_limit(self, channel: int, current: float) -> None: def set_current_limit(self, channel: int, current: float) -> None:
"""Set current limit for specified channel.""" """[docstring]"""
... pass
@abstractmethod
def get_current_limit(self, channel: int) -> float: def get_current_limit(self, channel: int) -> float:
"""Get current limit for specified channel.""" """[docstring]"""
... pass
@abstractmethod
def measure_voltage(self, channel: int) -> float: def measure_voltage(self, channel: int) -> float:
"""Measure actual output voltage.""" """[docstring]"""
... pass
@abstractmethod
def measure_current(self, channel: int) -> float: def measure_current(self, channel: int) -> float:
"""Measure actual output current.""" """[docstring]"""
... pass
@abstractmethod
def enable_output(self, channel: int, enable: bool) -> None: def enable_output(self, channel: int, enable: bool) -> None:
"""Enable or disable channel output.""" """[docstring]"""
... pass
@abstractmethod
def is_output_enabled(self, channel: int) -> bool: def is_output_enabled(self, channel: int) -> bool:
"""Check if channel output is enabled.""" """[docstring]"""
... pass
@runtime_checkable @runtime_checkable
class IMultimeter(Protocol): class IMultimeter(ABC):
"""Hardware abstraction for digital multimeters.""" """Hardware abstraction for digital multimeters."""
@abstractmethod
def measure_dc_voltage(self, range: str = "AUTO") -> float: def measure_dc_voltage(self, range: str = "AUTO") -> float:
"""Measure DC voltage. Range: AUTO, 0.1, 1, 10, 100, 1000.""" """[docstring]"""
... pass
@abstractmethod
def measure_dc_current(self, range: str = "AUTO") -> float: def measure_dc_current(self, range: str = "AUTO") -> float:
"""Measure DC current.""" """[docstring]"""
... pass
@abstractmethod
def measure_resistance(self, range: str = "AUTO") -> float: def measure_resistance(self, range: str = "AUTO") -> float:
"""Measure resistance.""" """[docstring]"""
... pass
@abstractmethod
def set_integration_time(self, nplc: float) -> None: def set_integration_time(self, nplc: float) -> None:
"""Set integration time in power line cycles (0.1 to 100).""" """[docstring]"""
... pass
@runtime_checkable @runtime_checkable
class ITestLogger(Protocol): class ITestLogger(ABC):
"""Abstraction for test data logging.""" """Abstraction for test data logging."""
@abstractmethod
def log_measurement( def log_measurement(
self, self,
parameter: str, parameter: str,
@@ -560,9 +574,10 @@ class ITestLogger(Protocol):
unit: str, unit: str,
conditions: dict[str, float] | None = None conditions: dict[str, float] | None = None
) -> None: ) -> None:
"""Log a single measurement.""" """[docstring]"""
... pass
@abstractmethod
def log_result( def log_result(
self, self,
parameter: str, parameter: str,
@@ -571,12 +586,13 @@ class ITestLogger(Protocol):
lower_limit: float | None = None, lower_limit: float | None = None,
upper_limit: float | None = None upper_limit: float | None = None
) -> None: ) -> None:
"""Log a test result with optional limits.""" """[docstring]"""
... pass
@abstractmethod
def log_event(self, message: str, level: str = "INFO") -> None: def log_event(self, message: str, level: str = "INFO") -> None:
"""Log a test event or message.""" """[docstring]"""
... pass
``` ```
### 4.2 Transport Interface ### 4.2 Transport Interface
@@ -584,36 +600,42 @@ class ITestLogger(Protocol):
```python ```python
# py_dvt_ate/instruments/transport/base.py # py_dvt_ate/instruments/transport/base.py
from typing import Protocol from abc import ABC, abstractmethod
class Transport(Protocol): class Transport(ABC):
"""Abstract transport interface for instrument communication.""" """Abstract transport interface for instrument communication."""
@abstractmethod
def connect(self) -> None: def connect(self) -> None:
"""Establish connection to instrument.""" """[docstring]"""
... pass
@abstractmethod
def disconnect(self) -> None: def disconnect(self) -> None:
"""Close connection to instrument.""" """[docstring]"""
... pass
@abstractmethod
def write(self, command: str) -> None: def write(self, command: str) -> None:
"""Send command to instrument.""" """[docstring]"""
... pass
@abstractmethod
def read(self, timeout: float | None = None) -> str: def read(self, timeout: float | None = None) -> str:
"""Read response from instrument.""" """[docstring]"""
... pass
@abstractmethod
def query(self, command: str, timeout: float | None = None) -> str: def query(self, command: str, timeout: float | None = None) -> str:
"""Send command and read response.""" """[docstring]"""
... pass
@property @property
@abstractmethod
def is_connected(self) -> bool: def is_connected(self) -> bool:
"""Check if connection is active.""" """[docstring]"""
... pass
``` ```
### 4.3 Test Interface ### 4.3 Test Interface
@@ -624,7 +646,7 @@ class Transport(Protocol):
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
from typing import Protocol from abc import ABC, abstractmethod
from uuid import UUID from uuid import UUID
@@ -675,22 +697,25 @@ class TestContext:
config: dict config: dict
class ITest(Protocol): class ITest(ABC):
"""Interface for test implementations.""" """Interface for test implementations."""
@property @property
@abstractmethod
def name(self) -> str: def name(self) -> str:
"""Test name identifier.""" """[docstring]"""
... pass
@property @property
@abstractmethod
def description(self) -> str: def description(self) -> str:
"""Human-readable test description.""" """[docstring]"""
... pass
@abstractmethod
def execute(self, context: TestContext) -> TestStatus: def execute(self, context: TestContext) -> TestStatus:
"""Execute the test, return status.""" """[docstring]"""
... pass
``` ```
### 4.4 Factory Interface ### 4.4 Factory Interface
@@ -1173,30 +1198,34 @@ Schema:
```python ```python
# py_dvt_ate/data/repository.py (interface) # py_dvt_ate/data/repository.py (interface)
from typing import Protocol from abc import ABC, abstractmethod
from uuid import UUID from uuid import UUID
class ITestRepository(Protocol): class ITestRepository(ABC):
"""Repository interface for test data.""" """Repository interface for test data."""
@abstractmethod
def create_run( def create_run(
self, self,
test_name: str, test_name: str,
config: dict, config: dict,
operator: str | None = None operator: str | None = None
) -> UUID: ) -> UUID:
"""Create a new test run, return its ID.""" """[docstring]"""
... pass
@abstractmethod
def update_run_status(self, run_id: UUID, status: str) -> None: def update_run_status(self, run_id: UUID, status: str) -> None:
"""Update test run status.""" """[docstring]"""
... pass
@abstractmethod
def complete_run(self, run_id: UUID, status: str) -> None: def complete_run(self, run_id: UUID, status: str) -> None:
"""Mark test run as complete with final status.""" """[docstring]"""
... pass
@abstractmethod
def save_result( def save_result(
self, self,
run_id: UUID, run_id: UUID,
@@ -1206,28 +1235,32 @@ class ITestRepository(Protocol):
lower_limit: float | None = None, lower_limit: float | None = None,
upper_limit: float | None = None upper_limit: float | None = None
) -> None: ) -> None:
"""Save a test result.""" """[docstring]"""
... pass
@abstractmethod
def save_measurements( def save_measurements(
self, self,
run_id: UUID, run_id: UUID,
measurements: list["Measurement"] measurements: list["Measurement"]
) -> None: ) -> None:
"""Save batch of measurements to Parquet.""" """[docstring]"""
... pass
@abstractmethod
def get_run(self, run_id: UUID) -> "TestRun": def get_run(self, run_id: UUID) -> "TestRun":
"""Get test run by ID.""" """[docstring]"""
... pass
@abstractmethod
def get_results(self, run_id: UUID) -> list["TestResult"]: def get_results(self, run_id: UUID) -> list["TestResult"]:
"""Get all results for a test run.""" """[docstring]"""
... pass
@abstractmethod
def get_measurements_dataframe(self, run_id: UUID): def get_measurements_dataframe(self, run_id: UUID):
"""Get measurements as pandas DataFrame.""" """[docstring]"""
... pass
``` ```
--- ---