API

This page contains the auto-generated documentation for the socs package.

socs.agents

socs.agents.scpi_psu.drivers

class socs.agents.scpi_psu.drivers.PsuInterface(ip_address, gpibAddr, verbose=False, **kwargs)

Bases: PrologixInterface

enable(ch)

Enables output for channel (1,2,3) but does not turn it on. Depending on state of power supply, it might need to be called before the output is set.

disable(ch)

disabled output from a channel (1,2,3). once called, enable must be called to turn on the channel again

set_chan(ch)

set_output(ch, out)

set status of power supply channel ch - channel (1,2,3) to set status out - ON: True|1|’ON’ OFF: False|0|’OFF’

Calls enable to ensure a channel can be turned on. We might want to make them separate (and let us use disable as a safety feature) but for now I am thinking we just want to thing to turn on when we tell it to turn on.

get_output(ch)

check if the output of a channel (1,2,3) is on (True) or off (False)

set_volt(ch, volt)

set_curr(ch, curr)

get_volt(ch)

get_curr(ch)

socs.agents.tektronix3021c.drivers

Michael Randall mrandall@ucsd.edu

class socs.agents.tektronix3021c.drivers.TektronixInterface(ip_address, gpibAddr, **kwargs)

Bases: PrologixInterface

set_freq(freq)

set_amp(amp)

set_output(state)

socs.agents.vantagepro2.drivers

socs.agents.vantagepro2.drivers.calc_crc(data)

Calculates CRC.

socs.agents.vantagepro2.drivers.F_to_C(temp)

Function to convert fahrenheit measurement to celsius

socs.agents.vantagepro2.drivers.wind_chill(temp, wind)

Function to calculate wind chill temperature. Only valid if temp < 50F. Taken from https://www.calculator.net/wind-chill-calculator.html If temp > 50F, need to use heat index instead…

Temp: Temperature in Fahrenheit wind: Speed in miles per hour

class socs.agents.vantagepro2.drivers.VantagePro2(path, baud=19200, timeout=1)

Bases: object

Allows communication to Vantage Pro 2 Weather Monitor Module. Contains commands to be issued and member variables that store collected data.

startup()

Wakeup vantage pro 2 console.

crc_check(data)

Checks received data. If received data has a CRC value of 0, then correct message received!

close()

conditions_screen()

Move console to conditions screen (where loop command can succesfully be sent.)

msg(msg)

Send general command or query to module. If command received and acknowledged, continue else retry 2 more times.

interrupt_daq()

Interrupts loop command…if sent before loop command finishes.

receive_data()

Reads weather data from console and returns loop_data{}. User should read Vantage Pro Serial Communication Reference Manual for the format of the loop data packet. Specifically for units of data! Available fields:

• Barometer trend: the current 3 hour barometer trend. Possible values:

  -60 = Falling rapidly
  -20 = Falling slowly
    0 = Steady
   20 = Rising slowly
   60 = Rising rapidly
   80 = No trend available
  
  Any other value = The VantagePro2 does not have the 3 hours
  of data needed to determine the barometer trend.
  

• Barometer: Current barometer reading (Hg/1000)

• Inside Temperature: Temperatue in Celsius (10th of a degree)

• Inside Humidity: Relative humidity in percent

• Outside Temperature: Temperature in Celsius (10th of a degree)

• Wind Speed: Wind speed in miles per hour

• 10 min average wind speed: 10 minute average wind speed in mph

• Wind Direction: From 1-360 degrees. Possible values:

    0 = No wind direction data
   90 = East
  180 = South
  270 = West
  360 = North
  

• Extra Temperatures: Temperature from up to 7 extra temperature stations.

• Soil Temperatures: Four soil temperature sensors, in the same format as the extra temperatures format listed above.

• Leaf Temperatures: Four leaf temperature sensors, in the same format as the extra temperatures format listed above.

• Outside Humidity: Relativie humidity in percent

• Extra Humidities: Realtive humidity in percent for 7 humidity stations

• Rain Rate: Number of rain clicks (0.1in or 0.2mm) per hour

• UV: “Unit is in UV index”

• Solar Radiation: Units in watt/meter^2

• Storm Rain: Stored in 100th of an inch

• Start Date of Current storm: Gives month, date, and year (offset by 2000)

• Day Rain: Number of rain clicks (0.1in or 0.2mm)/hour in the past day

• Month Rain: Number of rain clicks (0.1in or 0.2mm)/hour in the past month

• Year Rain: Number of rain clicks (0.1in or 0.2mm)/hour in the past year

• Day ET: 1000th of an inch

• Month ET: 1000th of an inch

• Year ET: 1000th of an inch

• Soil Moistures: In centibar, supports 4 soil sensors

• Leaf Wetnesses: Scale from 0-15. Supports 4 leaf sensors. Possible values:

   0 = Very dry
  15 = Very wet
  

• Inside Alarms: Currently active inside alarms

• Rain Alarms: Currently active rain alarms

• Outside Alarms: Currently active outside alarms

• Outside Humidity Alarms: Currently active humidity alarms

• Extra Temp/Hum Alarms: Currently active extra temperature/humidity alarms

• Soil & Leaf Alarms: Currently active soil/leaf alarms

• Console Battery Voltage: Voltage

• Time of Sunrise: Time is stored as hour x 100 + min

• Time of Sunset: Time is stored as hour x 100 + min

weather_daq()

Issues “LOOP 1”(which samples weather data) command to weather station, and unpacks the data.

print_data(data)

Prints contents of data collected from LOOP <loops> command. Loop: loop# Field : Value format.

socs.common

The common/ directory contains driver code that is used by multiple socs Agents.

socs.common.moxa_serial

class socs.common.moxa_serial.Serial_TCPServer(port, timeout=1.0)

Bases: object

Class to speak with the moxa serial / Ethernet converter. Set up the moxa box ports according to the specifications of the device hooked into each serial port.

A typical sequence of messages for dealing with a device. Create the socket once:

>>> moxa = moxa_serial.Serial_TCPServer(('IP',port),timeout=1.0)

Then do this sequence, complicated in some way by an individual device’s hand shaking needs:

>>> moxa.flushInput()
>>> moxa.write(msg)
>>> moxa.readexactly(n)

I write a “cmd” methods that handle the proper sequence with checksums etc. Most devices require a certain delay between commands, which is left to the user. If using multithreading, wrap your delays in mutexes.

Parameters:

• port (tuple) – (IP addr, TCP port)

• timeout (float) – Timeout for reading from the moxa box

readexactly(n)

Tries to read exactly n bytes within the timeout.

Parameters:

n – Number of bytes to read.

Returns:

Returned message if n bytes were read. Empty string if len(message) != n.

Return type:

str

readbuf_slow(n)

Reads whatever is in the buffer right now, but is O(N) in buffer size.

Parameters:

n – Number of bytes to read.

readbuf(n)

Returns whatever is currently in the buffer. Suitable for large buffers.

Parameters:

n – Number of bytes to read.

readpacket(n)

Like read(), but may not return everything if the moxa box flushes too soon.

Will probably read whatever arrives in the buffer, up to n or the timeout. Use read() for certainty.

read(n)

Like readexactly(), but returns whatever is in the buffer if it can’t fill up.

This replicates the behavior of the read method in pyserial. I feel that readexactly() has better behavior for most applications though.

Parameters:

n – Number of bytes to read. Will read at most n bytes.

Returns:

Returned message of up to n bytes.

Return type:

str

readline(term='\n')

readall()

write(msg)

Sends message to the moxa box.

Parameters:

msg (str) – Message to send, including terminator (i.e. \r\n) if needed.

writeread(msg)

flushInput()

Erases the input buffer at this moment.

Before I ask for new info from a device, I flush my receive buffer to make sure I don’t get any garbage in front.

settimeout(timeout)

Sets the socket in timeout mode.

gettimeout()

property timeout

Communication timeout. Only use timeout mode with timeout > 0.0.

socs.common.pmx

class socs.common.pmx.PMX(rtu_port=None, tcp_ip=None, tcp_port=None, timeout=None)

Bases: object

The PMX object is for communicating with the Kikusui PMX power supplies.

Parameters:

• rtu_port (str) – Serial RTU port

• tcp_ip (str) – TCP IP address

• tcp_port (int) – TCP port

• timeout (int) – Connection timeout

check_connect()

Check the connection.

check_voltage()

Check the voltage.

check_current()

Check the current.

check_voltage_current()

Check both the voltage and current.

check_voltage_setting()

Check the voltage setting

check_current_setting()

Check the current setting

check_voltage_current_setting()

Check both the voltage and current setting

check_output()

Return the output status.

set_voltage(val, silent=False)

Set the PMX voltage.

set_current(val, silent=False)

Set the PMX on.

use_external_voltage()

Set PMX to use external voltage.

ign_external_voltage()

Set PMX to ignore external voltage.

set_voltage_limit(val, silent=False)

Set the PMX voltage limit.

set_current_limit(val, silent=False)

Set the PMX current limit.

turn_on()

Turn the PMX on.

turn_off()

Turn the PMX off.

wait()

Sleep.

clean_serial()

Flush the serial buffer.

class socs.common.pmx.Command(input_PMX)

Bases: object

The Command object is used to command the PMX.

Parameters:

PMX (PMX) – PMX object

get_help()

Print possible commands.

user_input(arg)

Take user input and execute PMX command.

socs.common.prologix_interface

class socs.common.prologix_interface.PrologixInterface(ip_address, gpibAddr, **kwargs)

Bases: object

conn_socket()

configure()

write(msg)

read()

version()

identify()

socs.db

The db/ directory contains code that supports databases used within socs.

socs.db.suprsync

class socs.db.suprsync.TimecodeDir(**kwargs)

Bases: Base

Table for information about ‘timecode’ directories. These are directories in particular archives such as ‘smurf’ that we care about tracking whether or not they’ve completed syncing.

Timecode directories must start with a 5-digit time-code.

timecode

Timecode for directory. Must be 5 digits, and will be roughly 1 a day.

Type:

int

archive_name

Archive the directory is in.

Type:

str

completed

True if we expect no more files to be added to this directory.

Type:

bool

synced

True if all files in this directory have been synced to the remote.

Type:

bool

finalized

True if the ‘finalization’ file has been written and added to the db.

Type:

bool

finalize_file_id

ID for the SupRsyncFile object that is the finalization file for this timecode dir.

Type:

int

id

timecode

archive_name

completed

synced

finalized

finalize_file_id

class socs.db.suprsync.SupRsyncFile(**kwargs)

Bases: Base

Files table utilized by the SupRsync agent.

local_path

Absolute path of the local file to be copied

Type:

String

local_md5sum

locally calculated checksum

Type:

String

archive_name

Name of the archive, i.e. timestreams or smurf. Each archive is managed by its own SupRsync instance, so they can be copied to different base-dirs or hosts.

Type:

String

remote_path

Path of the file on the remote server relative to the base-dir. specified in the SupRsync agent config.

Type:

String

remote_md5sum

Md5sum calculated on remote machine

Type:

String, optional

timestamp

Timestamp that file was added to db

Type:

Float

copied

Time at which file was transfered

Type:

Float, optional

removed

Time at which file was removed from local server.

Type:

Float, optional

failed_copy_attempts

Number of failed copy attempts

Type:

Int

deletable

Whether file should be deleted after copying

Type:

Bool

ignore

If true, file will be ignored by SupRsync agent and not included in finalized_until.

Type:

Bool

id

local_path

local_md5sum

archive_name

remote_path

timestamp

remote_md5sum

copied

removed

failed_copy_attempts

deletable

ignore

socs.db.suprsync.split_path(path)

Splits path into a list where each element is a subdirectory

socs.db.suprsync.check_timecode(file: SupRsyncFile)

Tries to extract timecode from the remote path. If it fails, returns None.

socs.db.suprsync.create_file(local_path, remote_path, archive_name, local_md5sum=None, timestamp=None, deletable=True)

Creates SupRsyncFiles object.

Parameters:

• local_path (String or Path) – Absolute path of the local file to be copied

• remote_path (String) – Path of the file on the remote server relative to the base-dir. specified in the SupRsync agent config.

• archive_name (String) – Name of the archive, i.e. timestreams or smurf. Each archive is managed by its own SupRsync instance, so they can be copied to different base-dirs or hosts.

• local_md5sum (String, optional) – locally calculated checksum. If not specified, will calculate md5sum automatically.

• timestamp – Timestamp of file. If None is specified, will use the current time.

• deletable (bool) – If true, can be deleted by suprsync agent

class socs.db.suprsync.SupRsyncFilesManager(db_path, create_all=True, echo=False)

Bases: object

Helper class for accessing and adding entries to the SupRsync files database.

Parameters:

• db_path (path) – path to sqlite db

• create_all (bool) – Create table if it hasn’t been generated yet.

• echo (bool) – If true, writes sql statements to stdout

get_archive_stats(archive_name, session=None)

get_finalized_until(archive_name, session=None)

Returns a timetamp for which all files preceding are either successfully copied, or ignored. If all files are copied, returns the current time.

Parameters:

• archive_name (String) – Archive name to get finalized_until for

• session (sqlalchemy session) – SQLAlchemy session to use. If none is passed, will create a new session

add_file(local_path, remote_path, archive_name, local_md5sum=None, timestamp=None, session=None, deletable=True)

Adds file to the SupRsyncFiles table.

Parameters:

• local_path (String) – Absolute path of the local file to be copied

• remote_path (String) – Path of the file on the remote server relative to the base-dir. specified in the SupRsync agent config.

• archive_name (String) – Name of the archive, i.e. timestreams or smurf. Each archive is managed by its own SupRsync instance, so they can be copied to different base-dirs or hosts.

• local_md5sum (String, optional) – locally calculated checksum. If not specified, will calculate md5sum automatically.

• session (sqlalchemy session) – Session to use to add the SupRsyncFile. If None, will create a new session and commit afterwards.

• deletable (bool) – If true, can be deleted by suprsync agent

get_copyable_files(archive_name, session=None, max_copy_attempts=None, num_files=None)

Gets all SupRsyncFiles that are copyable, meaning they satisfy:

• local and remote md5sums do not match

• Failed copy attempts is below the max number of attempts

Parameters:

• archive_name (string) – Name of archive to get files from

• session (sqlalchemy session) – Session to use to get files. If none is specified, one will be created. You need to specify this if you wish to change file data and commit afterwards.

• max_copy_attempts (int) – Max number of failed copy atempts

• num_files (int) – Number of files to return

get_deletable_files(archive_name, delete_after, session=None)

Gets all files that are deletable, meaning that the local and remote md5sums match, and they have existed longer than delete_after seconds.

Parameters:

• archive_name (str) – Name of archive to pull files from

• delete_after (float) – Time since creation (in seconds) for which it’s ok to delete files.

• session (sqlalchemy session) – Session to use to query files.

get_known_files(archive_name, session=None, min_ctime=None)

Gets all files. This can be used to help avoid double-registering files.

Parameters:

• archive_name (str) – Name of archive to pull files from

• session (sqlalchemy session) – Session to use to query files.

• min_ctime (float, optional) – minimum ctime to use when querying files.

create_all_timecode_dirs(archive_name, min_ctime=None)

update_all_timecode_dirs(archive_name, file_root, sync_id)

class socs.db.suprsync.SupRsyncFileHandler(file_manager, archive_name, remote_basedir, ssh_host=None, ssh_key=None, cmd_timeout=None, copy_timeout=None, compression=None, bwlimit=None)

Bases: object

Helper class to handle files in the suprsync db and copy them to their dest / delete them if enough time has passed.

run_on_remote(cmd, timeout=None)

Runs a command on the remote server or locally if ssh_host is None.

Parameters:

cmd (list) – Command to be run

copy_files(max_copy_attempts=None, num_files=None)

Copies a batch of files, and computes remote md5sums.

Parameters:

• max_copy_attempts (int) – Max number of failed copy atempts

• num_files (int) – Number of files to return

Returns:

copy_attempts – Each entry of the list provides the path to the copied file, and a bool indicating wheter the remote md5sum matched.

Return type:

list of (str, bool)

delete_files(delete_after)

Gets deletable files, deletes them, and updates file info

Parameters:

delete_after (float) – Time since creation (in seconds) for which it’s ok to delete files.

socs.Lakeshore

The Lakeshore/ directory contains code that supports the various Lakeshore devices used in socs. For details, see the “Supporting APIs” section of the relevant Agent page.

socs.smurf

socs.smurf.MultiSlotController

class socs.smurf.MultiSlotController.MultiSlotController(client_args=[])

Bases: object

Simple client controller useful for managing all pysmurf-controller agents that exist on a network. This can dispatch operations and check operation status for any or all pysmurf-controllers on a network.

Parameters:

client_args (list) – List of command line arguments to pass onto the OCSClient objects.

smurfs

Dictionary containing OCSClient objects for each pysmurf-controller on the network. This assumes pysmurf controller agents have instance-id’s of the form pysmurf-controller-<id> where id can be the uxm-id or something like crate1slot2. The keys of this dict will be <id>. This is populated from the registry agent on init or using the funciton get_active_smurfs.

Type:

dict

Example

>>> msc = MultiSlotController()
>>> print(msc.keys())
dict_keys(['c1s2', 'c1s3', 'c1s4', 'c1s5', 'c1s6', 'c1s7'])
>>> msc.start('uxm_setup')  # Run UXM Setup on all slots
>>> # Waits for all slots to finish setup. You should monitor logs or check
>>> # returned session-data to make sure this ran correctly
>>> msc.wait('uxm_setup')
>>> # Streams for 30 seconds on slots 2 and 3
>>> msc.start('stream', ids=['c1s2', 'c1s3'], duration=30)
>>> msc.wait('stream')  # Waits for controllers to finish streaming

get_active_smurfs()

Obtains a list of active pysmurf-controller agents on a network from the registry agent. Uses this to populate the smurfs dictionary, which contains an OCSClient object for each.

start(op_name, ids=None, **kwargs)

Starts an operation on any or all pysmurf controllers.

Parameters:

• op_name (str) – Name of the operation to start

• ids (list, optional) – List of pysmurf-controller id’s. This defaults to running on all controllers.

• **kwargs – Any additional keyword arguments are passed onto the start operation as params

stop(op_name, ids=None, **kwargs)

Stops an operation on any or all pysmurf controllers.

Parameters:

• op_name (str) – Name of the operation to stop

• ids (list, optional) – List of pysmurf-controller id’s. This defaults to running on all controllers.

• **kwargs – Any additional keyword arguments are passed onto the stop operation as params

status(op_name, ids=None)

Checks the status of an operation on any or all pysmurf controllers

Parameters:

• op_name (str) – Name of the operation to stop

• ids (list, optional) – List of pysmurf-controller id’s. This defaults to running on all controllers.

wait(op_name, ids=None)

Waits for an operation to finish on any or all pysmurf controllers. This will block until the operation sessions of all speicified ID’s have been completed.

Parameters:

• op_name (str) – Name of the operation to stop

• ids (list, optional) – List of pysmurf-controller id’s. This defaults to running on all controllers.

socs.snmp

class socs.snmp.SNMPTwister(address, port=161)

Bases: object

Helper class for handling SNMP communication with twisted.

More information can be found in the pySNMP documentation: PySNMP Examples

Parameters:

• address (str) – Address of the SNMP Agent to send GET/SET requests to

• port (int) – Associated port for SNMP communication. Default is 161

snmp_engine

PySNMP engine

Type:

pysnmp.entity.engine.SnmpEngine

address

Address of the SNMP Agent to send GET/SET requests to

Type:

str

udp_transport

UDP transport for UDP over IPv4

Type:

pysnmp.hlapi.twisted.transport.UdpTransportTarget

log

txaio logger object

Type:

txaio.tx.Logger

get(oid_list, version)

Issue a getCmd to get SNMP OID states.

Example

>>> snmp = SNMPTwister('localhost', 161)
>>> snmp.get([ObjectType(ObjectIdentity('MBG-SNMP-LTNG-MIB',
                                        'mbgLtNgRefclockState',
                                        1)),
              ObjectType(ObjectIdentity('MBG-SNMP-LTNG-MIB',
                                        'mbgLtNgRefclockLeapSecondDate',
                                        1))])

>>> snmp = SNMPTwister('localhost', 161)
>>> result = snmp.get([('MBG-SNMP-LTNG-MIB', 'mbgLtNgRefclockState', 1),
                       ('MBG-SNMP-LTNG-MIB', 'mbgLtNgRefclockLeapSecondDate', 1)])
>>> # Simply printing the returned object shows a nice string
>>> print(result[0])
MBG-SNMP-LTNG-MIB::mbgLtNgRefclockState.1 = notSynchronized
>>> # The corresponding integer value is hidden within the returned object
>>> print(result[0][1]._value)
2

Parameters:

• oid_list (list) –

  List of high-level MIB Object OIDs. The list elements should either be ObjectType, or tuples which define the OIDs, as shown in the example above. See Specifying MIB object for more info.

• version (int) –

  SNMP version for communicaton (1, 2, or 3). All versions supported here without auth or privacy. If using v3 the configured username on the SNMP device should be ‘ocs’. For details on version implementation in pysnmp see SNMP Versions.

Returns:

A Deferred which will callback with the var_binds list from self._success. If successful, this will contain a list of ObjectType class instances representing MIB variables returned in SNMP response.

Return type:

twisted.internet.defer.Deferred

set(oid_list, version, setvalue, community_name='private')

Issue a setCmd to set SNMP OID states. See Modifying MIB variables for more info on setting OID states.

Parameters:

• oid_list (list) – List of high-level MIB Object OIDs. The list elements should either be ObjectType, or tuples which define the OIDs.

• version (int) – SNMP version for communicaton (1, 2, or 3). All versions supported here without auth or privacy. If using v3 the configured username on the SNMP device should be ‘ocs’.

• setvalue (int) – Integer to set OID. For example, 0 is off and 1 is on for outletControl on the iBootPDU.

Returns:

A Deferred which will callback with the var_binds list from self._success. If successful, this will contain a list of ObjectType class instances representing MIB variables returned in SNMP response.

Return type:

twisted.internet.defer.Deferred

socs.testing

The testing/ directory contains code that supports testing Agents.

socs.testing.device_emulator

socs.testing.device_emulator.create_device_emulator(responses, relay_type, port=9001, encoding='utf-8', reconnect=False)

Create a device emulator fixture.

This provides a device emulator that can be used to mock a device during testing.

Parameters:

• responses (dict) – Dictionary with commands as keys, and responses as values. See DeviceEmulator for details.

• relay_type (str) – Communication relay type. Either ‘serial’ or ‘tcp’.

• port (int) – Port for the TCP relay to listen for connections on. Defaults to 9001. Only used if relay_type is ‘tcp’.

• encoding (str) – Encoding for the messages and responses. See socs.testing.device_emulator.DeviceEmulator() for more details.

• reconnect (bool) – If True, on TCP client disconnect, the emulator will listen for new incoming connections instead of quitting

Returns:

A pytest fixture that creates a Device emulator of the specified type.

Return type:

function

class socs.testing.device_emulator.DeviceEmulator(responses, encoding='utf-8', reconnect=False)

Bases: object

A mocked device which knows how to respond on various communication channels.

Parameters:

• responses (dict) – Initial responses, any response required by Agent startup, if any.

• encoding (str) – Encoding for the messages and responses. DeviceEmulator will try to encode and decode messages with the given encoding. No encoding is used if set to None. That can be useful if you need to use raw data from your hardware. Defaults to ‘utf-8’.

• reconnect (bool) – If True, on TCP client disconnect, the emulator will listen for new incoming connections instead of quitting

responses

Current set of responses the DeviceEmulator would give. Should all be strings, not bytes-like.

Type:

dict

default_response

Default response to send if a command is unrecognized. No response is sent and an error message is logged if a command is unrecognized and the default response is set to None. Defaults to None.

Type:

str

encoding

Encoding for the messages and responses, set by the encoding argument.

Type:

str

_type

Relay type, either ‘serial’ or ‘tcp’.

Type:

str

_read

Used to stop the background reading of data recieved on the relay.

Type:

bool

_conn

TCP connection for use in ‘tcp’ relay.

Type:

socket.socket

create_serial_relay()

Create the serial relay, emulating a hardware device connected over serial.

This first uses socat to setup a relay. It then connects to the “internal” end of the relay, ready to receive communications sent to the “responder” end of the relay. This end of the relay is located at ./responder. You will need to configure your Agent to use that path for communication.

Next it creates a thread to read commands sent to the serial relay in the background. This allows responses to be defined within a test using DeviceEmulator.define_responses() after instantiation of the DeviceEmulator object within a given test.

get_response(msg)

Determine the response to a given message.

Parameters:

msg (str) – Command string to get the response for.

Returns:

Response string. Will return None if a valid response is not

found.

Return type:

str

shutdown()

Shutdown communication on the configured relay. This will stop any attempt to read communication on the relay, as well as shutdown the relay itself.

create_tcp_relay(port)

Create the TCP relay, emulating a hardware device connected over TCP.

Creates a thread to read commands sent to the TCP relay in the background. This allows responses to be defined within a test using DeviceEmulator.define_responses() after instantiation of the DeviceEmulator object within a given test.

Parameters:

port (int) – Port for the TCP relay to listen for connections on.

Notes

This will not return until the socket is properly bound to the given port. If this setup is not working it is likely another device emulator instance is not yet finished or has not been properly shutdown.

update_responses(responses: Dict)

Updates the current responses. See define_responses for more detail.

Parameters:

responses (dict) – Dict of commands to use to update the current responses.

define_responses(responses, default_response=None)

Define what responses are available to reply with on the configured communication relay.

Parameters:

• responses (dict) – Dictionary of commands: response. Values can be a list, in which case the responses in the list are popped and given in order until depleted.

• default_response (str) – Default response to send if a command is unrecognized. No response is sent and an error message is logged if a command is unrecognized and the default response is set to None. Defaults to None.

Examples

The given responses might look like:

>>> responses = {'KRDG? 1': '+1.7E+03'}
>>> responses = {'*IDN?': 'LSCI,MODEL425,4250022,1.0',
                 'RDGFIELD?': ['+1.0E-01', '+1.2E-01', '+1.4E-01']}

Notes

The DeviceEmulator will handle encoding/decoding. The responses defined should all be strings, not bytes-like, unless you set encoding=None.

socs.util

socs.util.get_md5sum(filename)