Source code for qblox_instruments.ieee488_2.transport
# ----------------------------------------------------------------------------
# Description : Transport layer (abstract, IP, file, dummy)
# Git repository : https://gitlab.com/qblox/packages/software/qblox_instruments.git
# Copyright (C) Qblox BV (2020)
# ----------------------------------------------------------------------------
# -- include -----------------------------------------------------------------
import asyncio
from abc import ABCMeta, abstractmethod
from collections.abc import Awaitable
from threading import Event, Thread
from typing import Optional, TypeVar
from qblox_instruments.ieee488_2.helpers import LoopError
# -- helpers ------------------------------------------------------------------
class TransportEventLoopThread(Thread):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.loop = None
self.loop_ready = Event()
def run(self) -> None:
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self.loop_ready.set()
try:
self.loop.run_forever()
finally:
self.loop.close()
self.loop_ready.clear()
TRANSPORT_THREAD: Optional[TransportEventLoopThread] = None
def get_threaded_event_loop() -> asyncio.BaseEventLoop:
global TRANSPORT_THREAD # noqa: PLW0603, discourage globals
if TRANSPORT_THREAD is None:
TRANSPORT_THREAD = TransportEventLoopThread(daemon=True)
TRANSPORT_THREAD.start()
TRANSPORT_THREAD.loop_ready.wait()
return TRANSPORT_THREAD.loop
# -- class --------------------------------------------------------------------
T = TypeVar("T")
[docs]
class Transport(metaclass=ABCMeta):
"""
Abstract base class for data transport to instruments.
"""
[docs]
def __init__(
self, loop_from: Optional["Transport"] = None, use_running_loop: bool = False
) -> None:
"""
Create transport instance.
Parameters
----------
loop_from: Transport | None = None
Transport to use event loop from: bypasses `use_running_loop` checks.
use_running_loop : bool = False
Whether to use an existing event loop if it is already running;
use this only if you are supplying your own event loop and only invoking
asynchronous functions, or if the event loop is not running when invoking
synchronous command functions. Default to `False`.
"""
if loop_from is not None:
self._loop = loop_from._loop
self._loop_threaded = loop_from._loop_threaded
else:
self._loop_threaded = False
try:
self._loop = asyncio.get_running_loop()
except RuntimeError:
self._loop = asyncio.new_event_loop()
if self._loop.is_running() and not use_running_loop:
self._loop = get_threaded_event_loop()
self._loop_threaded = True
# ------------------------------------------------------------------------
def _run_in_loop(self, fn: Awaitable[T]) -> T:
# NOTE: We need to manually cancel the coroutine when it isn't run.
# In some cases, it might be destroyed by the destructor of the coroutine
# without having been awaited, which leads to RuntimeWarning being emitted
# and polluting the logs.
if self._loop_threaded:
return asyncio.run_coroutine_threadsafe(fn, self._loop).result()
elif self._loop.is_running():
fn.close() # Cancel the coroutine if it can't be run.
raise LoopError(f"An event loop is already running. Cannot run {fn}.")
else:
try:
return self._loop.run_until_complete(fn)
except RuntimeError as e:
fn.close() # Cancel the coroutine if it failed to run
if str(e) == "Event loop is closed":
# Raise a more specific error if the event loop is already closed,
# so it can be easily caught and suppressed by the caller.
raise LoopError(str(e)) from None
raise
# ------------------------------------------------------------------------
@property
def loop(self) -> asyncio.BaseEventLoop:
"""
Event loop for this transport.
"""
return self._loop
# ------------------------------------------------------------------------
@property
@abstractmethod
def timeout(self) -> float:
"""
Timeout for this transport.
"""
pass
# ------------------------------------------------------------------------
@timeout.setter
@abstractmethod
def timeout(self, value: float) -> None:
"""
Set timeout for this transport.
"""
pass
# ------------------------------------------------------------------------
[docs]
@abstractmethod
async def close(self) -> None:
"""
Abstract method to close instrument.
"""
pass
# ------------------------------------------------------------------------
[docs]
@abstractmethod
def reconnect(self) -> None:
"""
Close and re-open connection to instrument.
"""
# ------------------------------------------------------------------------
[docs]
@abstractmethod
async def write(self, cmd_str: str) -> None:
"""
Abstract method to write command to instrument.
Parameters
----------
cmd_str : str
Command
"""
pass
# ------------------------------------------------------------------------
[docs]
@abstractmethod
async def write_binary(self, *data: bytes) -> None:
"""
Abstract method to write binary data to instrument.
Parameters
----------
*data : bytes
Binary data
"""
pass
# ------------------------------------------------------------------------
[docs]
@abstractmethod
async def read_binary(self, size: int) -> bytes:
"""
Abstract method to read binary data from instrument.
Parameters
----------
size : int
Number of bytes
Returns
-------
bytes
Binary data array of length "size".
"""
pass
# ------------------------------------------------------------------------
[docs]
@abstractmethod
async def readline(self) -> str:
"""
Abstract method to read data from instrument.
Returns
-------
str
String with data.
"""
pass
# ------------------------------------------------------------------------
def __enter__(self) -> "Transport":
"""
Context manager entry.
Returns
-------
Any
Returns self
"""
return self._run_in_loop(self.__aenter__())
# ------------------------------------------------------------------------
def __exit__(self, exc_type, exc_value, traceback) -> Optional[bool]:
"""
Context manager exit. Closes the transport.
"""
return self._run_in_loop(self.__aexit__(exc_type, exc_value, traceback))
# ------------------------------------------------------------------------
async def __aenter__(self) -> "Transport":
"""
Asynchronous context manager entry.
Returns
-------
Any
Returns self
"""
return self
# ------------------------------------------------------------------------
async def __aexit__(self, exc_type, exc_value, traceback) -> Optional[bool]:
"""
Context manager exit. Closes the transport.
"""
await self.close()
