Add transport protocol definition

This commit is contained in:
2025-06-14 20:48:34 +00:00
parent b26943bdf8
commit 729da8184b
2 changed files with 85 additions and 1 deletions

View File

@@ -6,6 +6,7 @@ Provides connection abstractions for different backends:
- PyVISA (for real instruments)
"""
from py_dvt_ate.instruments.transport.base import Transport
from py_dvt_ate.instruments.transport.server import InstrumentServer, SCPIDevice
__all__ = ["InstrumentServer", "SCPIDevice"]
__all__ = ["Transport", "InstrumentServer", "SCPIDevice"]

View File

@@ -0,0 +1,83 @@
"""Base transport protocol for instrument communication."""
from typing import Protocol
class Transport(Protocol):
"""Abstract transport interface for instrument communication.
This protocol defines the interface that all transport implementations
(TCP, VISA, etc.) must satisfy. It provides basic connection management
and communication primitives for SCPI-based instruments.
"""
def connect(self) -> None:
"""Establish connection to instrument.
Raises:
ConnectionError: If connection fails.
"""
...
def disconnect(self) -> None:
"""Close connection to instrument.
Should be idempotent - safe to call multiple times.
"""
...
def write(self, command: str) -> None:
"""Send command to instrument.
Args:
command: SCPI command string to send (without terminator).
Raises:
ConnectionError: If not connected.
IOError: If write fails.
"""
...
def read(self, timeout: float | None = None) -> str:
"""Read response from instrument.
Args:
timeout: Read timeout in seconds. None uses default.
Returns:
Response string from instrument (without terminator).
Raises:
ConnectionError: If not connected.
TimeoutError: If read times out.
IOError: If read fails.
"""
...
def query(self, command: str, timeout: float | None = None) -> str:
"""Send command and read response.
Convenience method combining write() and read().
Args:
command: SCPI command string to send.
timeout: Read timeout in seconds. None uses default.
Returns:
Response string from instrument.
Raises:
ConnectionError: If not connected.
TimeoutError: If read times out.
IOError: If communication fails.
"""
...
@property
def is_connected(self) -> bool:
"""Check if connection is active.
Returns:
True if connected, False otherwise.
"""
...