kschappell/py-dvt-ate
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Refactor Transport from Protocol to ABC for explicit interface implementation

Browse Source
This commit is contained in:
Kai Chappell
2025-12-02 22:06:00 +00:00
parent 78a09e7bf0
commit b6edb1c0bb
2 changed files with 26 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/py_dvt_ate/instruments/transport/base.py
View File

@@ -1,31 +1,38 @@
"""Base transport protocol for instrument communication.""" """Base transport interface for instrument communication."""
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.
This protocol defines the interface that all transport implementations This abstract base class defines the interface that all transport
(TCP, VISA, etc.) must satisfy. It provides basic connection management implementations (TCP, VISA, etc.) must implement. It provides basic
and communication primitives for SCPI-based instruments. connection management and communication primitives for SCPI-based
instruments.
Implementations must inherit from this class and implement all abstract
methods.
""" """
@abstractmethod
def connect(self) -> None: def connect(self) -> None:
"""Establish connection to instrument. """Establish connection to instrument.
Raises: Raises:
ConnectionError: If connection fails. ConnectionError: If connection fails.
""" """
... pass
@abstractmethod
def disconnect(self) -> None: def disconnect(self) -> None:
"""Close connection to instrument. """Close connection to instrument.
Should be idempotent - safe to call multiple times. Should be idempotent - safe to call multiple times.
""" """
... pass
@abstractmethod
def write(self, command: str) -> None: def write(self, command: str) -> None:
"""Send command to instrument. """Send command to instrument.
@@ -36,8 +43,9 @@ class Transport(Protocol):
ConnectionError: If not connected. ConnectionError: If not connected.
IOError: If write fails. IOError: If write fails.
""" """
... pass
@abstractmethod
def read(self, timeout: float | None = None) -> str: def read(self, timeout: float | None = None) -> str:
"""Read response from instrument. """Read response from instrument.
@@ -52,8 +60,9 @@ class Transport(Protocol):
TimeoutError: If read times out. TimeoutError: If read times out.
IOError: If read fails. IOError: If read fails.
""" """
... 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. """Send command and read response.
@@ -71,13 +80,14 @@ class Transport(Protocol):
TimeoutError: If read times out. TimeoutError: If read times out.
IOError: If communication fails. IOError: If communication fails.
""" """
... pass
@property @property
@abstractmethod
def is_connected(self) -> bool: def is_connected(self) -> bool:
"""Check if connection is active. """Check if connection is active.
Returns: Returns:
True if connected, False otherwise. True if connected, False otherwise.
""" """
... pass

6
src/py_dvt_ate/instruments/transport/tcp.py
View File

@@ -3,11 +3,13 @@
import socket import socket
from typing import Any from typing import Any
from py_dvt_ate.instruments.transport.base import Transport
class TCPTransport:
class TCPTransport(Transport):
"""TCP socket transport implementation. """TCP socket transport implementation.
Implements the Transport protocol for communicating with SCPI Implements the Transport interface for communicating with SCPI
instruments over TCP/IP using newline-terminated messages. instruments over TCP/IP using newline-terminated messages.
Attributes: Attributes: