Add async TCP server foundation

Create InstrumentServer class with asyncio for hosting virtual SCPI
instruments over TCP. Supports registering instruments on specific
ports with port-to-instrument mapping.
This commit is contained in:
2025-05-22 21:32:38 +00:00
parent 9fedc1f851
commit 0f0672dd03
2 changed files with 94 additions and 0 deletions

View File

@@ -3,3 +3,7 @@
Provides virtual instruments backed by a coupled thermal-electrical
physics engine. Used for development and testing without real hardware.
"""
from py_dvt_ate.simulation.tcp_server import InstrumentServer
__all__ = ["InstrumentServer"]

View File

@@ -0,0 +1,90 @@
"""Async TCP server for exposing virtual instruments over network.
This module provides the InstrumentServer class that hosts virtual SCPI
instruments over TCP, allowing client applications to communicate using
standard SCPI commands over a network connection.
"""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from py_dvt_ate.simulation.virtual.base import BaseInstrument
logger = logging.getLogger(__name__)
class InstrumentServer:
"""Async TCP server hosting virtual SCPI instruments.
Each instrument is assigned a port. Clients connect via TCP and send
SCPI commands as newline-terminated strings. Responses are also
newline-terminated.
Attributes:
host: Host address to bind to.
"""
def __init__(self, host: str = "127.0.0.1") -> None:
"""Initialise the instrument server.
Args:
host: Host address to bind to. Defaults to localhost.
"""
self._host = host
self._instruments: dict[int, BaseInstrument] = {}
self._servers: list[asyncio.Server] = []
self._running = False
@property
def host(self) -> str:
"""Get the host address."""
return self._host
@property
def is_running(self) -> bool:
"""Check if server is currently running."""
return self._running
def register_instrument(self, port: int, instrument: BaseInstrument) -> None:
"""Register an instrument to be served on a specific port.
Args:
port: TCP port number to serve the instrument on.
instrument: Virtual instrument to serve.
Raises:
ValueError: If port is already registered.
RuntimeError: If server is already running.
"""
if self._running:
raise RuntimeError("Cannot register instruments while server is running")
if port in self._instruments:
raise ValueError(f"Port {port} is already registered")
self._instruments[port] = instrument
logger.info(
"Registered %s on port %d",
instrument.__class__.__name__,
port,
)
def get_instrument(self, port: int) -> BaseInstrument | None:
"""Get the instrument registered on a port.
Args:
port: Port number to look up.
Returns:
Registered instrument, or None if port not registered.
"""
return self._instruments.get(port)
@property
def registered_ports(self) -> list[int]:
"""Get list of registered port numbers."""
return list(self._instruments.keys())