from typing import List
from .Internal.Core import Core
from .Internal.InstrumentErrors import RsInstrException
from .Internal.CommandsGroup import CommandsGroup
from .Internal.VisaSession import VisaSession
# noinspection PyPep8Naming,PyAttributeOutsideInit,SpellCheckingInspection
[docs]class RsOsp:
"""63 total commands, 5 Sub-groups, 0 group commands"""
_driver_options = "SupportedInstrModels = OSP2xx/OSP3xx, SupportedIdnPatterns = OSP, SimulationIdnString = 'Rohde&Schwarz,OSP320,100001,2.10.17.0075'"
def __init__(self, resource_name: str, id_query: bool = True, reset: bool = False, options: str = None, direct_session: object = None):
"""Initializes new RsOsp session. \n
Parameter options tokens examples:
- ``Simulate=True`` - starts the session in simulation mode. Default: ``False``
- ``SelectVisa=socket`` - uses no VISA implementation for socket connections - you do not need any VISA-C installation
- ``SelectVisa=rs`` - forces usage of RohdeSchwarz Visa
- ``SelectVisa=ivi`` - forces usage of National Instruments Visa
- ``QueryInstrumentStatus = False`` - same as ``driver.utilities.instrument_status_checking = False``. Default: ``True``
- ``WriteDelay = 20, ReadDelay = 5`` - Introduces delay of 20ms before each write and 5ms before each read. Default: ``0ms`` for both
- ``OpcWaitMode = OpcQuery`` - mode for all the opc-synchronised write/reads. Other modes: StbPolling, StbPollingSlow, StbPollingSuperSlow. Default: ``StbPolling``
- ``AddTermCharToWriteBinBLock = True`` - Adds one additional LF to the end of the binary data (some instruments require that). Default: ``False``
- ``AssureWriteWithTermChar = True`` - Makes sure each command/query is terminated with termination character. Default: Interface dependent
- ``TerminationCharacter = "\\r"`` - Sets the termination character for reading. Default: ``\\n`` (LineFeed or LF)
- ``DataChunkSize = 10E3`` - Maximum size of one write/read segment. If transferred data is bigger, it is split to more segments. Default: ``1E6`` bytes
- ``OpcTimeout = 10000`` - same as driver.utilities.opc_timeout = 10000. Default: ``30000ms``
- ``VisaTimeout = 5000`` - same as driver.utilities.visa_timeout = 5000. Default: ``10000ms``
- ``ViClearExeMode = Disabled`` - viClear() execution mode. Default: ``execute_on_all``
- ``OpcQueryAfterWrite = True`` - same as driver.utilities.opc_query_after_write = True. Default: ``False``
- ``StbInErrorCheck = False`` - if true, the driver checks errors with *STB? If false, it uses SYST:ERR?. Default: ``True``
- ``LoggingMode = LoggingMode.On`` - Sets the logging status right from the start. Default: ``Off``
- ``LoggingName = 'MyDevice'`` - Sets the name to represent the session in the log entries. Default: ``'resource_name'``
:param resource_name: VISA resource name, e.g. 'TCPIP::192.168.2.1::INSTR'
:param id_query: if True: the instrument's model name is verified against the models supported by the driver and eventually throws an exception.
:param reset: Resets the instrument (sends *RST command) and clears its status sybsystem
:param options: string tokens alternating the driver settings.
:param direct_session: Another driver object or pyVisa object to reuse the session instead of opening a new session."""
self._core = Core(resource_name, id_query, reset, RsOsp._driver_options, options, direct_session)
self._core.driver_version = '2.10.17.0075'
self._options = options
self._add_all_global_repcaps()
self._custom_properties_init()
# noinspection PyTypeChecker
self._base = CommandsGroup("ROOT", self._core, None)
[docs] @classmethod
def from_existing_session(cls, session: object, options: str = None) -> 'RsOsp':
"""Creates a new RsOsp object with the entered 'session' reused. \n
:param session: can be an another driver or a direct pyvisa session.
:param options: string tokens alternating the driver settings."""
# noinspection PyTypeChecker
return cls(None, False, False, options, session)
def __str__(self) -> str:
if self._core.io:
return f"RsOsp session '{self._core.io.resource_name}'"
else:
return f"RsOsp with session closed"
[docs] @staticmethod
def assert_minimum_version(min_version: str) -> None:
"""Asserts that the driver version fulfills the minimum required version you have entered.
This way you make sure your installed driver is of the entered version or newer."""
min_version_list = min_version.split('.')
curr_version_list = '2.10.17.0075'.split('.')
count_min = len(min_version_list)
count_curr = len(curr_version_list)
count = count_min if count_min < count_curr else count_curr
for i in range(count):
minimum = int(min_version_list[i])
curr = int(curr_version_list[i])
if curr > minimum:
break
if curr < minimum:
raise RsInstrException(f"Assertion for minimum RsOsp version failed. Current version: '2.10.17.0075', minimum required version: '{min_version}'")
[docs] @staticmethod
def list_resources(expression: str = '?*::INSTR', visa_select: str = None) -> List[str]:
"""Finds all the resources defined by the expression
- '?*' - matches all the available instruments
- 'USB::?*' - matches all the USB instruments
- "TCPIP::192?*' - matches all the LAN instruments with the IP address starting with 192
:param expression: see the examples in the function
:param visa_select: optional parameter selecting a specific VISA. Examples: '@ivi', '@rs'
"""
rm = VisaSession.get_resource_manager(visa_select)
resources = rm.list_resources(expression)
rm.close()
# noinspection PyTypeChecker
return resources
[docs] def close(self) -> None:
"""Closes the active RsOsp session."""
self._core.io.close()
[docs] def get_session_handle(self) -> object:
"""Returns the underlying session handle."""
return self._core.get_session_handle()
def _add_all_global_repcaps(self) -> None:
"""Adds all the repcaps defined as global to the instrument's global repcaps dictionary."""
def _custom_properties_init(self):
"""Adds all the interfaces that are custom for the driver."""
from .CustomFiles.utilities import Utilities
self.utilities = Utilities(self._core)
from .CustomFiles.events import Events
self.events = Events(self._core)
@property
def configure(self):
"""configure commands group. 8 Sub-classes, 0 commands."""
if not hasattr(self, '_configure'):
from .Implementations.Configure import Configure
self._configure = Configure(self._core, self._base)
return self._configure
@property
def read(self):
"""read commands group. 3 Sub-classes, 0 commands."""
if not hasattr(self, '_read'):
from .Implementations.Read import Read
self._read = Read(self._core, self._base)
return self._read
@property
def route(self):
"""route commands group. 4 Sub-classes, 0 commands."""
if not hasattr(self, '_route'):
from .Implementations.Route import Route
self._route = Route(self._core, self._base)
return self._route
@property
def diagnostic(self):
"""diagnostic commands group. 1 Sub-classes, 0 commands."""
if not hasattr(self, '_diagnostic'):
from .Implementations.Diagnostic import Diagnostic
self._diagnostic = Diagnostic(self._core, self._base)
return self._diagnostic
@property
def trigger(self):
"""trigger commands group. 5 Sub-classes, 4 commands."""
if not hasattr(self, '_trigger'):
from .Implementations.Trigger import Trigger
self._trigger = Trigger(self._core, self._base)
return self._trigger