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 ec164596a8
commit fcb4f68a53

View File

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