# ----------------------------------------------------------------------------
# Description : IEEE488.2 interface
# Git repository : https://gitlab.com/qblox/packages/software/qblox_instruments.git
# Copyright (C) Qblox BV (2020)
# ----------------------------------------------------------------------------
# -- include -----------------------------------------------------------------
from __future__ import annotations
from asyncio import Lock
from contextlib import asynccontextmanager
from functools import wraps
from typing import TYPE_CHECKING, TypeVar
from qblox_instruments.ieee488_2.helpers import make_sync_wrapper
from qblox_instruments.types import DebugLevel
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Awaitable
from typing import Any, Callable
from qblox_instruments.ieee488_2 import Transport
T = TypeVar("T")
# -- function ----------------------------------------------------------------
def ieee488_2_error_check( # noqa: ANN201
minimal_check: bool | Callable[..., Any | None] = False,
):
"""
Factory function for a decorator that catches and checks for errors on an IEEE488.2 call.
Parameters
----------
minimal_check
If True, this decorator will always check for errors unless the debug
level is ``DebugLevel.NO_CHECK``. By default False.
Returns
-------
Callable
The decorator.
Raises
------
RuntimeError
An error was found in system error.
"""
def decorator(func: Callable[..., Any | None]) -> Callable[..., Any | None]:
@wraps(func)
def wrapper(self: Ieee488_2, *args: Any, **kwargs: Any) -> Any | None:
if self._debug in (DebugLevel.ERROR_CHECK, DebugLevel.VERSION_AND_ERROR_CHECK) or (
self._debug == DebugLevel.MINIMAL_CHECK and minimal_check is True
):
error = None
try:
return func(self, *args, **kwargs)
except OSError:
raise
except Exception as err:
error = err
finally:
self._check_error(error)
else:
return func(self, *args, **kwargs)
return wrapper
# if used without parentheses
if callable(minimal_check):
return decorator(minimal_check)
# if used with parentheses
return decorator
def ieee488_2_error_check_async( # noqa: ANN201
minimal_check: bool | Callable[..., Any | None] = False,
):
"""
Factory function for a decorator that catches and checks for errors on an IEEE488.2 call.
Parameters
----------
minimal_check
If True, this decorator will always check for errors unless the debug
level is ``DebugLevel.NO_CHECK``. By default False.
Returns
-------
Callable
The decorator.
Raises
------
RuntimeError
An error was found in system error.
"""
def decorator(func: Callable[..., Any | None]) -> Callable[..., Any | None]:
@wraps(func)
async def wrapper(self: Ieee488_2, *args: Any, **kwargs: Any) -> Any | None:
if self._debug in (DebugLevel.ERROR_CHECK, DebugLevel.VERSION_AND_ERROR_CHECK) or (
self._debug == DebugLevel.MINIMAL_CHECK and minimal_check is True
):
error = None
try:
return await func(self, *args, **kwargs)
except OSError:
raise
except Exception as err:
error = err
finally:
await self._check_error_async(error)
else:
return await func(self, *args, **kwargs)
return wrapper
# if used without parentheses
if callable(minimal_check):
return decorator(minimal_check)
# if used with parentheses
return decorator
# -- class -------------------------------------------------------------------
[docs]
class Ieee488_2: # noqa: N801
"""
Class that implements the IEEE488.2 interface.
"""
__slots__ = ("_debug", "_lock", "_transport")
# ------------------------------------------------------------------------
[docs]
def __init__(self, transport: Transport, debug: DebugLevel = DebugLevel.MINIMAL_CHECK) -> None:
"""
Creates IEEE488.2 interface object.
Parameters
----------
transport : Transport
Transport class responsible for the lowest level of communication
(e.g. ethernet).
debug : DebugLevel
Debug level. See :class:`~qblox_instruments.types.DebugLevel` for more
information. By default None, which means that for a connection to a dummy
cluster, `DebugLevel.ERROR_CHECK` will be used, and for a real cluster,
`DebugLevel.MINIMAL_CHECK`.
"""
self._transport = transport
self._lock = Lock()
self._debug = debug
# ------------------------------------------------------------------------
def _run_in_loop(self, fn: Awaitable[T]) -> T:
return self._transport._run_in_loop(fn)
# ------------------------------------------------------------------------
[docs]
def reconnect(self) -> None:
"""Close and re-open the connection to instrument."""
self._transport.reconnect()
# ------------------------------------------------------------------------
[docs]
async def close(self) -> None:
"""Close the connection to the instrument. It can not be used afterwards."""
if self._transport:
await self._transport.close()
self._transport = None
# ------------------------------------------------------------------------
@asynccontextmanager
async def _connection(self) -> AsyncGenerator[Ieee488_2]:
async with self._lock:
yield self
# ------------------------------------------------------------------------
async def _write_async(self, cmd_str: str) -> None:
"""
Write command to instrument.
Parameters
----------
cmd_str : str
Command string.
"""
await self._transport.write(cmd_str)
_write = make_sync_wrapper(_write_async)
# ------------------------------------------------------------------------
async def _write_bin_async(self, cmd_str: str, bin_block: bytes) -> None:
"""
Write command and binary data block to instrument.
Parameters
----------
cmd_str : str
Command string.
bin_block : bytes
Binary data array to send.
"""
await self._bin_block_write(cmd_str + " ", bin_block)
_write_bin = make_sync_wrapper(_write_bin_async)
# ------------------------------------------------------------------------
async def _read_async(self, cmd_str: str) -> str:
"""
Write command to instrument and read response. Using an empty command
string will skip the write and only read.
Parameters
----------
cmd_str : str
Command string.
Returns
-------
str
Command response string.
"""
await self._transport.write(cmd_str)
data = await self._transport.readline()
return data.rstrip() # Remove trailing white space, CR, LF
_read = make_sync_wrapper(_read_async)
# ------------------------------------------------------------------------
async def _read_bin_async(self, cmd_str: str, flush_line_end: bool = True) -> bytes:
"""
Write command to instrument and read binary data block. Using an empty
command string will skip the write and only read to allow reading
concatenated binary blocks.
Parameters
----------
cmd_str : str
Command string.
flush_line_end : bool
Flush end of line characters.
Returns
-------
bytes
Binary data array received.
"""
if cmd_str != "":
await self._transport.write(cmd_str)
return await self._bin_block_read(flush_line_end)
_read_bin = make_sync_wrapper(_read_bin_async)
# ------------------------------------------------------------------------
async def _write_read_bin_async(
self, cmd_str: str, bin_block: bytes, flush_line_end: bool = True
) -> bytes:
"""
Write command and binary data block to instrument and read binary data
block. Using an empty command string will skip the write and only read
to allow reading concatenated binary blocks.
Parameters
----------
cmd_str : str
Command string.
bin_block : bytes
Binary data array to send.
flush_line_end : bool
Flush end of line characters.
Returns
-------
bytes
Binary data array received.
"""
if cmd_str != "":
await self._write_bin_async(cmd_str, bin_block)
return await self._bin_block_read(flush_line_end)
_write_read_bin = make_sync_wrapper(_write_read_bin_async)
# ------------------------------------------------------------------------
async def _bin_block_write(self, cmd_str: str, bin_block: bytes) -> None:
"""
Write IEEE488.2 binary data block to instrument.
Parameters
----------
cmd_str : str
Command string.
bin_block : bytes
Binary data array to send.
"""
header = cmd_str + Ieee488_2._build_header_string(len(bin_block))
await self._transport.write_binary(header.encode(), bin_block)
await self._transport.write("") # Add a Line Terminator
# ------------------------------------------------------------------------
async def _bin_block_read(self, flush_line_end: bool = True) -> bytes:
"""
Read IEEE488.2 binary data block from instrument.
Parameters
----------
flush_line_end : bool
Flush end of line characters.
Returns
-------
bytes
Binary data array received.
Raises
------
RunTimeError
Header error.
"""
header_a = await self._transport.read_binary(2) # Read '#N'
header_a_str = header_a.decode()
# character ',' is a valid delimiter and we need to discard it
if header_a_str[0] == ",":
# read again to get the digit part of '#N'
digit = await self._transport.read_binary(1)
digit_str = digit.decode()
header_a_str = header_a_str[1] + digit_str # continue as before
if header_a_str[0] != "#":
s = f"Header error: received {header_a}"
raise RuntimeError(s)
digit_cnt = int(header_a_str[1])
header_b = await self._transport.read_binary(digit_cnt)
byte_cnt = int(header_b.decode())
bin_block = await self._transport.read_binary(byte_cnt)
if flush_line_end:
await self._flush_line_end_async()
return bin_block
# ------------------------------------------------------------------------
@staticmethod
def _build_header_string(byte_cnt: int) -> str:
"""
Generate IEEE488.2 binary data block header.
Parameters
----------
byte_cnt : int
Size of the binary data block in bytes.
Returns
-------
str
Header string.
"""
byte_cnt_str = str(byte_cnt)
digit_cnt_str = str(len(byte_cnt_str))
bin_header_str = "#" + digit_cnt_str + byte_cnt_str
return bin_header_str
# ------------------------------------------------------------------------
async def _flush_line_end_async(self) -> None:
"""
Flush end of line <CR><LF> characters.
"""
await self._transport.read_binary(2) # Consume <CR><LF>
_flush_line_end = make_sync_wrapper(_flush_line_end_async)
# -------------------------------------------------------------------------
async def _check_error_async(self, err: Exception | None = None) -> None:
"""
Check system error for errors. Empties and prints the complete error
queue.
Parameters
----------
err : Optional[Exception]
Exception to reraise.
Raises
------
Exception
An exception was passed as input argument.
"""
if err:
raise err from err
_check_error = make_sync_wrapper(_check_error_async)
# -------------------------------------------------------------------------
async def _fast_error_check_async(self, *raw_data: Any) -> None:
"""
Inspects raw SCPI responses for firmware error sentinels.
Triggers a full error queue check if a sentinel is detected.
Parameters
----------
*raw_data : Any
The raw response data returned from the SCPI read operation.
Accepts multiple arguments if a command returns multiple variables.
Raises
------
RuntimeError
An error was found in the system error queue after a
sentinel was detected.
"""
is_error = False
for data in raw_data:
if isinstance(data, (str, bytes)) and len(data) == 0:
is_error = True
break
if is_error:
await self._check_error_queue_async()
fast_error_check = make_sync_wrapper(_fast_error_check_async)
# ------------------------------------------------------------------------
# IEEE488.2 constants
# ------------------------------------------------------------------------
# fmt: off
# *ESR and *ESE bits
_ESR_OPERATION_COMPLETE = 0x01
_ESR_REQUEST_CONTROL = 0x02
_ESR_QUERY_ERROR = 0x04
_ESR_DEVICE_DEPENDENT_ERROR = 0x08
_ESR_EXECUTION_ERROR = 0x10
_ESR_COMMAND_ERROR = 0x20
_ESR_USER_REQUEST = 0x40
_ESR_POWER_ON = 0x80
# fmt: on
# ------------------------------------------------------------------------
# IEEE488.2 commands
# ------------------------------------------------------------------------
# -------------------------------------------------------------------------
async def _get_idn_async(self) -> str:
"""
Get device identity.
Parameters
----------
None
Returns
-------
str
Concatenated list of strings separated by the semicolon character.
The IDN consists of four strings.
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
result = await conn._read_async("*IDN?")
await self._fast_error_check_async(result)
return result
_get_idn = make_sync_wrapper(_get_idn_async)
# -------------------------------------------------------------------------
@ieee488_2_error_check_async
async def _reset_async(self) -> None:
"""
Reset device and clear all status and event registers.
Parameters
----------
None
Returns
-------
None
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
return await conn._write_async("*RST")
_reset = make_sync_wrapper(_reset_async)
# -------------------------------------------------------------------------
@ieee488_2_error_check_async
async def _clear_async(self) -> None:
"""
Clear all status and event registers (see `SCPI <https://www.ivifoundation.org/downloads/SCPI/scpi-99.pdf>`_).
Parameters
----------
None
Returns
-------
None
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
return await conn._write_async("*CLS")
clear = make_sync_wrapper(_clear_async)
# -------------------------------------------------------------------------
async def _get_status_byte_async(self) -> int:
"""
Get status byte register. Register is only cleared when feeding registers are cleared.
Parameters
----------
None
Returns
-------
int
Status byte register.
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
result = await conn._read_async("*STB?")
await self._fast_error_check_async(result)
return int(result)
get_status_byte = make_sync_wrapper(_get_status_byte_async)
# -------------------------------------------------------------------------
@ieee488_2_error_check_async
async def _set_service_request_enable_async(self, reg: int) -> None:
"""
Set service request enable register.
Parameters
----------
reg : int
Service request enable register.
Returns
-------
None
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
"""
if not isinstance(reg, int):
raise TypeError(
f"Unexpected type for input argument reg, expected int but got {type(reg).__qualname__}." # noqa: E501
)
async with self._connection() as conn:
await conn._write_async(f"*SRE {reg}")
set_service_request_enable = make_sync_wrapper(_set_service_request_enable_async)
# -------------------------------------------------------------------------
async def _get_service_request_enable_async(self) -> int:
"""
Get service request enable register. The register is cleared after reading it.
Parameters
----------
None
Returns
-------
int
Service request enable register.
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
result = await conn._read_async("*SRE?")
await self._fast_error_check_async(result)
return int(result)
get_service_request_enable = make_sync_wrapper(_get_service_request_enable_async)
# -------------------------------------------------------------------------
@ieee488_2_error_check_async
async def _set_standard_event_status_enable_async(self, reg: int) -> None:
"""
Set standard event status enable register.
Parameters
----------
reg : int
Standard event status enable register.
Returns
-------
None
Raises
------
Exception
Invalid input parameter type.
Exception
An error is reported in system error and debug <= 1.
"""
if not isinstance(reg, int):
raise TypeError(
f"Unexpected type for input argument reg, expected int but got {type(reg).__qualname__}." # noqa: E501
)
async with self._connection() as conn:
await conn._write_async(f"*ESE {reg}")
set_standard_event_status_enable = make_sync_wrapper(_set_standard_event_status_enable_async)
# -------------------------------------------------------------------------
async def _get_standard_event_status_enable_async(self) -> int:
"""
Get standard event status enable register. The register is cleared after reading it.
Parameters
----------
None
Returns
-------
int
Standard event status enable register.
Raises
------
Exception
An error is reported in system error and debug <= 1.
All errors are read from system error and listed in the exception.
"""
async with self._connection() as conn:
result = await conn._read_async("*ESE?")
await self._fast_error_check_async(result)
return int(result)
get_standard_event_status_enable = make_sync_wrapper(_get_standard_event_status_enable_async)
# -------------------------------------------------------------------------
async def _get_standard_event_status_async(self) -> int:
"""
Get standard event status register. The register is cleared after reading it.
Parameters
----------
None
Returns
-------
int
Standard event status register.
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
result = await conn._read_async("*ESR?")
await self._fast_error_check_async(result)
return int(result)
get_standard_event_status = make_sync_wrapper(_get_standard_event_status_async)
# -------------------------------------------------------------------------
@ieee488_2_error_check_async
async def _set_operation_complete_async(self) -> None:
"""
Set device in operation complete query active state.
Parameters
----------
None
Returns
-------
None
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
await conn._write_async("*OPC")
set_operation_complete = make_sync_wrapper(_set_operation_complete_async)
# -------------------------------------------------------------------------
async def _get_operation_complete_async(self) -> bool:
"""
Get operation complete state.
Parameters
----------
None
Returns
-------
bool
Operation complete state (False = running, True = completed).
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
result = await conn._read_async("*OPC?")
await self._fast_error_check_async(result)
return bool(int(result))
get_operation_complete = make_sync_wrapper(_get_operation_complete_async)
# -------------------------------------------------------------------------
async def _test_async(self) -> bool:
"""
Run self-test.
Parameters
----------
None
Returns
-------
bool
Test result (False = failed, True = success).
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
result = await conn._read_async("*TST?")
await self._fast_error_check_async(result)
return bool(int(result))
test = make_sync_wrapper(_test_async)
# -------------------------------------------------------------------------
@ieee488_2_error_check_async
async def _wait_async(self) -> None:
"""
Wait until operations completed before continuing.
Parameters
----------
None
Returns
-------
None
Raises
------
Exception
An error is reported in system error and debug <= 1.
"""
async with self._connection() as conn:
await conn._write_async("*WAI")
wait = make_sync_wrapper(_wait_async)
