Source code for rsmxo.rsmxo

from typing import ClassVar, List

from .Internal.Core import Core
from .Internal.InstrumentErrors import RsInstrException
from .Internal.CommandsGroup import CommandsGroup
from .Internal.VisaSession import VisaSession
from datetime import datetime, timedelta


# noinspection PyPep8Naming,PyAttributeOutsideInit,SpellCheckingInspection
[docs] class RsMxo: """ | Commands in total: 3535 | Subgroups: 48 | Direct child commands: 1 """ _driver_options = "SupportedInstrModels = MXO/ScopeStudio, SupportedIdnPatterns = MXO/ScopeStudio, SimulationIdnString = 'Rohde&Schwarz,MXO,100001,2.9.0.82'" # noinspection PyClassVar _global_logging_relative_timestamp: ClassVar[datetime] = None _global_logging_target_stream: ClassVar = None def __init__(self, resource_name: str, id_query: bool= True, reset: bool=False, options: str=None, direct_session: object=None): r""" Initializes new RsMxo 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`` - prefers usage of RohdeSchwarz VISA - ``SelectVisa=ivi`` - prefers usage of National Instruments VISA - ``SelectVisa=pyvisa-py`` - prefers usage of python VISA backend pyvisa-py - ``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-synchronized 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`` - ``ScpiQuotes = double'. - for SCPI commands, you can define how strings are quoted. With single or double quotes. Possible values: single | double | {char}. Default: ``single`` - ``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'`` - ``LogToGlobalTarget = True`` - Sets the logging target to the class-property previously set with RsMxo.set_global_logging_target() Default: ``False`` - ``LoggingToConsole = True`` - Immediately starts logging to the console. Default: False - ``LoggingToUdp = True`` - Immediately starts logging to the UDP port. Default: False - ``LoggingUdpPort = 49200`` - UDP port to log to. Default: 49200 :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. More tokens are separated by comma. :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, RsMxo._driver_options, options, direct_session) self._core.driver_version = '2.9.0.0082' self._options = options self._add_all_global_repcaps() self._custom_properties_init() self.utilities.default_instrument_setup() # noinspection PyTypeChecker self._cmd_group = CommandsGroup("ROOT", self._core, None)
[docs] @classmethod def from_existing_session(cls, session: object, options: str=None) -> 'RsMxo': """ Creates a new RsMxo object with the entered 'session' reused. \n :param session: Can be another driver or a direct pyvisa session. :param options: String tokens alternating the driver settings. More tokens are separated by comma. """ # noinspection PyTypeChecker resource_name = None if hasattr(session, 'resource_name'): resource_name = getattr(session, 'resource_name') return cls(resource_name, False, False, options, session)
[docs] @classmethod def set_global_logging_target(cls, target) -> None: """ Sets global common target stream that each instance can use. To use it, call the following: io.utilities.logger.set_logging_target_global(). If an instance uses global logging target, it automatically uses the global relative timestamp (if set). You can set the target to None to invalidate it. """ cls._global_logging_target_stream = target
[docs] @classmethod def get_global_logging_target(cls): """ Returns global common target stream. """ return cls._global_logging_target_stream
[docs] @classmethod def set_global_logging_relative_timestamp(cls, timestamp: datetime) -> None: """ Sets global common relative timestamp for log entries. To use it, call the following: io.utilities.logger.set_relative_timestamp_global() """ cls._global_logging_relative_timestamp = timestamp
[docs] @classmethod def set_global_logging_relative_timestamp_now(cls) -> None: """ Sets global common relative timestamp for log entries to this moment. To use it, call the following: io.utilities.logger.set_relative_timestamp_global(). """ cls._global_logging_relative_timestamp = datetime.now()
[docs] @classmethod def clear_global_logging_relative_timestamp(cls) -> None: """ Clears the global relative timestamp. After this, all the instances using the global relative timestamp continue logging with the absolute timestamps. """ # noinspection PyTypeChecker cls._global_logging_relative_timestamp = None
[docs] @classmethod def get_global_logging_relative_timestamp(cls) -> datetime | None: """ Returns global common relative timestamp for log entries. """ return cls._global_logging_relative_timestamp
def __str__(self) -> str: if self._core.io: return f"RsMxo session '{self._core.io.resource_name}'" else: return f"RsMxo with session closed"
[docs] def get_total_execution_time(self) -> timedelta: """ Returns total time spent by the library on communicating with the instrument. This time is always shorter than get_total_time(), since it does not include gaps between the communication. You can reset this counter with reset_time_statistics(). """ return self._core.io.total_execution_time
[docs] def get_total_time(self) -> timedelta: """ Returns total time spent by the library on communicating with the instrument. This time is always shorter than get_total_time(), since it does not include gaps between the communication. You can reset this counter with reset_time_statistics(). """ return datetime.now() - self._core.io.total_time_startpoint
[docs] def reset_time_statistics(self) -> None: """ Resets all execution and total time counters. Affects the results of get_total_time() and get_total_execution_time() """ self._core.io.reset_time_statistics()
[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.9.0.0082'.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 RsMxo version failed. Current version: '2.9.0.0082', 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 RsMxo 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) -> None: """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) from .CustomFiles.ivi_utility import IviUtility self.ivi_utility = IviUtility(self._core) from .CustomFiles.ivi_direct_io import IviDirectIo self.ivi_direct_io = IviDirectIo(self._core) def _sync_to_custom_properties(self, cloned: 'RsMxo') -> None: """Synchronises the state of all the custom properties to the entered object.""" cloned.utilities.sync_from(self.utilities) cloned.events.sync_from(self.events) cloned.ivi_utility.sync_from(self.ivi_utility) cloned.ivi_direct_io.sync_from(self.ivi_direct_io) @property def display(self): """ | Commands in total: 59 | Subgroups: 10 | Direct child commands: 1 """ if not hasattr(self, '_display'): from .Implementations.Display import DisplayCls self._display = DisplayCls(self._core, self._cmd_group) return self._display @property def franalysis(self): """ | Commands in total: 76 | Subgroups: 15 | Direct child commands: 5 """ if not hasattr(self, '_franalysis'): from .Implementations.Franalysis import FranalysisCls self._franalysis = FranalysisCls(self._core, self._cmd_group) return self._franalysis @property def hardCopy(self): """ | Commands in total: 9 | Subgroups: 2 | Direct child commands: 5 """ if not hasattr(self, '_hardCopy'): from .Implementations.HardCopy import HardCopyCls self._hardCopy = HardCopyCls(self._core, self._cmd_group) return self._hardCopy @property def service(self): """ | Commands in total: 1 | Subgroups: 1 | Direct child commands: 0 """ if not hasattr(self, '_service'): from .Implementations.Service import ServiceCls self._service = ServiceCls(self._core, self._cmd_group) return self._service @property def userDefined(self): """ | Commands in total: 4 | Subgroups: 1 | Direct child commands: 0 """ if not hasattr(self, '_userDefined'): from .Implementations.UserDefined import UserDefinedCls self._userDefined = UserDefinedCls(self._core, self._cmd_group) return self._userDefined @property def power(self): """ | Commands in total: 408 | Subgroups: 9 | Direct child commands: 0 """ if not hasattr(self, '_power'): from .Implementations.Power import PowerCls self._power = PowerCls(self._core, self._cmd_group) return self._power @property def wgenerator(self): """ | Commands in total: 66 | Subgroups: 13 | Direct child commands: 1 """ if not hasattr(self, '_wgenerator'): from .Implementations.Wgenerator import WgeneratorCls self._wgenerator = WgeneratorCls(self._core, self._cmd_group) return self._wgenerator @property def system(self): """ | Commands in total: 12 | Subgroups: 7 | Direct child commands: 2 """ if not hasattr(self, '_system'): from .Implementations.System import SystemCls self._system = SystemCls(self._core, self._cmd_group) return self._system @property def run(self): """ | Commands in total: 2 | Subgroups: 0 | Direct child commands: 2 """ if not hasattr(self, '_run'): from .Implementations.Run import RunCls self._run = RunCls(self._core, self._cmd_group) return self._run @property def massMemory(self): """ | Commands in total: 32 | Subgroups: 8 | Direct child commands: 10 """ if not hasattr(self, '_massMemory'): from .Implementations.MassMemory import MassMemoryCls self._massMemory = MassMemoryCls(self._core, self._cmd_group) return self._massMemory @property def channel(self): """ | Commands in total: 43 | Subgroups: 20 | Direct child commands: 0 """ if not hasattr(self, '_channel'): from .Implementations.Channel import ChannelCls self._channel = ChannelCls(self._core, self._cmd_group) return self._channel @property def calculate(self): """ | Commands in total: 73 | Subgroups: 2 | Direct child commands: 0 """ if not hasattr(self, '_calculate'): from .Implementations.Calculate import CalculateCls self._calculate = CalculateCls(self._core, self._cmd_group) return self._calculate @property def digital(self): """ | Commands in total: 12 | Subgroups: 10 | Direct child commands: 0 """ if not hasattr(self, '_digital'): from .Implementations.Digital import DigitalCls self._digital = DigitalCls(self._core, self._cmd_group) return self._digital @property def trfs(self): """ | Commands in total: 1 | Subgroups: 0 | Direct child commands: 1 """ if not hasattr(self, '_trfs'): from .Implementations.Trfs import TrfsCls self._trfs = TrfsCls(self._core, self._cmd_group) return self._trfs @property def treference(self): """ | Commands in total: 21 | Subgroups: 10 | Direct child commands: 0 """ if not hasattr(self, '_treference'): from .Implementations.Treference import TreferenceCls self._treference = TreferenceCls(self._core, self._cmd_group) return self._treference @property def mdefaults(self): """ | Commands in total: 7 | Subgroups: 1 | Direct child commands: 1 """ if not hasattr(self, '_mdefaults'): from .Implementations.Mdefaults import MdefaultsCls self._mdefaults = MdefaultsCls(self._core, self._cmd_group) return self._mdefaults @property def sbus(self): """ | Commands in total: 1351 | Subgroups: 30 | Direct child commands: 0 """ if not hasattr(self, '_sbus'): from .Implementations.Sbus import SbusCls self._sbus = SbusCls(self._core, self._cmd_group) return self._sbus @property def trigger(self): """ | Commands in total: 658 | Subgroups: 11 | Direct child commands: 5 """ if not hasattr(self, '_trigger'): from .Implementations.Trigger import TriggerCls self._trigger = TriggerCls(self._core, self._cmd_group) return self._trigger @property def export(self): """ | Commands in total: 27 | Subgroups: 3 | Direct child commands: 0 """ if not hasattr(self, '_export'): from .Implementations.Export import ExportCls self._export = ExportCls(self._core, self._cmd_group) return self._export @property def eye(self): """ | Commands in total: 20 | Subgroups: 10 | Direct child commands: 0 """ if not hasattr(self, '_eye'): from .Implementations.Eye import EyeCls self._eye = EyeCls(self._core, self._cmd_group) return self._eye @property def measurement(self): """ | Commands in total: 81 | Subgroups: 21 | Direct child commands: 1 """ if not hasattr(self, '_measurement'): from .Implementations.Measurement import MeasurementCls self._measurement = MeasurementCls(self._core, self._cmd_group) return self._measurement @property def formatPy(self): """ | Commands in total: 3 | Subgroups: 1 | Direct child commands: 2 """ if not hasattr(self, '_formatPy'): from .Implementations.FormatPy import FormatPyCls self._formatPy = FormatPyCls(self._core, self._cmd_group) return self._formatPy @property def saveset(self): """ | Commands in total: 2 | Subgroups: 1 | Direct child commands: 0 """ if not hasattr(self, '_saveset'): from .Implementations.Saveset import SavesetCls self._saveset = SavesetCls(self._core, self._cmd_group) return self._saveset @property def sessions(self): """ | Commands in total: 6 | Subgroups: 2 | Direct child commands: 1 """ if not hasattr(self, '_sessions'): from .Implementations.Sessions import SessionsCls self._sessions = SessionsCls(self._core, self._cmd_group) return self._sessions @property def meter(self): """ | Commands in total: 9 | Subgroups: 1 | Direct child commands: 2 """ if not hasattr(self, '_meter'): from .Implementations.Meter import MeterCls self._meter = MeterCls(self._core, self._cmd_group) return self._meter @property def p3Phase(self): """ | Commands in total: 133 | Subgroups: 4 | Direct child commands: 0 """ if not hasattr(self, '_p3Phase'): from .Implementations.P3Phase import P3PhaseCls self._p3Phase = P3PhaseCls(self._core, self._cmd_group) return self._p3Phase @property def generator(self): """ | Commands in total: 3 | Subgroups: 1 | Direct child commands: 2 """ if not hasattr(self, '_generator'): from .Implementations.Generator import GeneratorCls self._generator = GeneratorCls(self._core, self._cmd_group) return self._generator @property def status(self): """ | Commands in total: 51 | Subgroups: 2 | Direct child commands: 1 """ if not hasattr(self, '_status'): from .Implementations.Status import StatusCls self._status = StatusCls(self._core, self._cmd_group) return self._status @property def acquire(self): """ | Commands in total: 35 | Subgroups: 5 | Direct child commands: 11 """ if not hasattr(self, '_acquire'): from .Implementations.Acquire import AcquireCls self._acquire = AcquireCls(self._core, self._cmd_group) return self._acquire @property def calibration(self): """ | Commands in total: 4 | Subgroups: 1 | Direct child commands: 3 """ if not hasattr(self, '_calibration'): from .Implementations.Calibration import CalibrationCls self._calibration = CalibrationCls(self._core, self._cmd_group) return self._calibration @property def cursor(self): """ | Commands in total: 35 | Subgroups: 29 | Direct child commands: 0 """ if not hasattr(self, '_cursor'): from .Implementations.Cursor import CursorCls self._cursor = CursorCls(self._core, self._cmd_group) return self._cursor @property def histogram(self): """ | Commands in total: 26 | Subgroups: 7 | Direct child commands: 1 """ if not hasattr(self, '_histogram'): from .Implementations.Histogram import HistogramCls self._histogram = HistogramCls(self._core, self._cmd_group) return self._histogram @property def mtest(self): """ | Commands in total: 30 | Subgroups: 11 | Direct child commands: 0 """ if not hasattr(self, '_mtest'): from .Implementations.Mtest import MtestCls self._mtest = MtestCls(self._core, self._cmd_group) return self._mtest @property def zone(self): """ | Commands in total: 20 | Subgroups: 8 | Direct child commands: 0 """ if not hasattr(self, '_zone'): from .Implementations.Zone import ZoneCls self._zone = ZoneCls(self._core, self._cmd_group) return self._zone @property def gate(self): """ | Commands in total: 10 | Subgroups: 8 | Direct child commands: 0 """ if not hasattr(self, '_gate'): from .Implementations.Gate import GateCls self._gate = GateCls(self._core, self._cmd_group) return self._gate @property def layout(self): """ | Commands in total: 45 | Subgroups: 10 | Direct child commands: 0 """ if not hasattr(self, '_layout'): from .Implementations.Layout import LayoutCls self._layout = LayoutCls(self._core, self._cmd_group) return self._layout @property def probe(self): """ | Commands in total: 59 | Subgroups: 3 | Direct child commands: 0 """ if not hasattr(self, '_probe'): from .Implementations.Probe import ProbeCls self._probe = ProbeCls(self._core, self._cmd_group) return self._probe @property def trProbe(self): """ | Commands in total: 5 | Subgroups: 1 | Direct child commands: 0 """ if not hasattr(self, '_trProbe'): from .Implementations.TrProbe import TrProbeCls self._trProbe = TrProbeCls(self._core, self._cmd_group) return self._trProbe @property def refCurve(self): """ | Commands in total: 25 | Subgroups: 14 | Direct child commands: 4 """ if not hasattr(self, '_refCurve'): from .Implementations.RefCurve import RefCurveCls self._refCurve = RefCurveCls(self._core, self._cmd_group) return self._refCurve @property def refLevel(self): """ | Commands in total: 11 | Subgroups: 4 | Direct child commands: 0 """ if not hasattr(self, '_refLevel'): from .Implementations.RefLevel import RefLevelCls self._refLevel = RefLevelCls(self._core, self._cmd_group) return self._refLevel @property def timebase(self): """ | Commands in total: 8 | Subgroups: 2 | Direct child commands: 4 """ if not hasattr(self, '_timebase'): from .Implementations.Timebase import TimebaseCls self._timebase = TimebaseCls(self._core, self._cmd_group) return self._timebase @property def xy(self): """ | Commands in total: 4 | Subgroups: 4 | Direct child commands: 0 """ if not hasattr(self, '_xy'): from .Implementations.Xy import XyCls self._xy = XyCls(self._core, self._cmd_group) return self._xy @property def sense(self): """ | Commands in total: 2 | Subgroups: 1 | Direct child commands: 0 """ if not hasattr(self, '_sense'): from .Implementations.Sense import SenseCls self._sense = SenseCls(self._core, self._cmd_group) return self._sense @property def hdefinition(self): """ | Commands in total: 4 | Subgroups: 0 | Direct child commands: 4 """ if not hasattr(self, '_hdefinition'): from .Implementations.Hdefinition import HdefinitionCls self._hdefinition = HdefinitionCls(self._core, self._cmd_group) return self._hdefinition @property def pbus(self): """ | Commands in total: 25 | Subgroups: 16 | Direct child commands: 1 """ if not hasattr(self, '_pbus'): from .Implementations.Pbus import PbusCls self._pbus = PbusCls(self._core, self._cmd_group) return self._pbus @property def synchronize(self): """ | Commands in total: 14 | Subgroups: 3 | Direct child commands: 0 """ if not hasattr(self, '_synchronize'): from .Implementations.Synchronize import SynchronizeCls self._synchronize = SynchronizeCls(self._core, self._cmd_group) return self._synchronize @property def autoScale(self): """ | Commands in total: 1 | Subgroups: 0 | Direct child commands: 1 """ if not hasattr(self, '_autoScale'): from .Implementations.AutoScale import AutoScaleCls self._autoScale = AutoScaleCls(self._core, self._cmd_group) return self._autoScale @property def triggerInvoke(self): """ | Commands in total: 1 | Subgroups: 0 | Direct child commands: 1 """ if not hasattr(self, '_triggerInvoke'): from .Implementations.TriggerInvoke import TriggerInvokeCls self._triggerInvoke = TriggerInvokeCls(self._core, self._cmd_group) return self._triggerInvoke
[docs] def stop(self) -> None: """ ``STOP`` \n Snippet: ``driver.stop()`` \n Stops the running acquisition. """ self._core.io.write(f'STOP')
[docs] def stop_and_wait(self, opc_timeout_ms: int = -1) -> None: """ ``STOP`` \n Snippet: ``driver.stop_and_wait()`` \n Stops the running acquisition. Same as stop, but waits for the operation to complete before continuing further. Use the RsMxo.utilities.opc_timeout_set() to set the timeout value. :param opc_timeout_ms: Maximum time to wait in milliseconds, valid only for this call. """ self._core.io.write_with_opc(f'STOP', opc_timeout_ms)
def clone(self) -> 'RsMxo': """ Creates a deep copy of the RsMxo object. Also copies: - All the existing Global repeated capability values - All the default group repeated capabilities setting \n Does not check the *IDN? response, and does not perform Reset. After cloning, you can set all the repeated capabilities settings independentely from the original group. Calling close() on the new object does not close the original VISA session """ cloned = RsMxo.from_existing_session(self.get_session_handle(), self._options) self._cmd_group.synchronize_repcaps(cloned) self._sync_to_custom_properties(cloned) return cloned
[docs] def restore_all_repcaps_to_default(self) -> None: """ Sets all the Group and Global repcaps to their initial values. """ self._cmd_group.restore_repcaps()