Add transport protocol definition
This commit is contained in:
@@ -6,6 +6,7 @@ Provides connection abstractions for different backends:
|
|||||||
- PyVISA (for real instruments)
|
- PyVISA (for real instruments)
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
from py_dvt_ate.instruments.transport.base import Transport
|
||||||
from py_dvt_ate.instruments.transport.server import InstrumentServer, SCPIDevice
|
from py_dvt_ate.instruments.transport.server import InstrumentServer, SCPIDevice
|
||||||
|
|
||||||
__all__ = ["InstrumentServer", "SCPIDevice"]
|
__all__ = ["Transport", "InstrumentServer", "SCPIDevice"]
|
||||||
|
|||||||
83
src/py_dvt_ate/instruments/transport/base.py
Normal file
83
src/py_dvt_ate/instruments/transport/base.py
Normal 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.
|
||||||
|
"""
|
||||||
|
...
|
||||||
Reference in New Issue
Block a user