SimpleMockController¶

class lsst.ts.hexrotcomm.SimpleMockController(log: Logger, port: int = 6210, host: str = '127.0.0.1', initial_state: IntEnum = ControllerState.STANDBY)¶

Bases: BaseMockController

Simple mock controller for unit testing BaseMockController.

The MOVE command sets cmd_position and curr_position, then the controller slowly increments curr_position.

Parameters:

loglogging.Logger: Logger.
portint: Port for the TCP/IP server. Specify 0 to choose random values for both ports; this is recommended for unit tests, to avoid collision with other tests.
initial_statelsst.ts.xml.enums.MTHexapod.ControllerState (optional): Initial state of mock controller.
hoststr or None, optional: IP address for the TCP/IP server. Typically “127.0.0.1” (the default) for an IPV4 server and “::” for an IPV6 server. If None then bind to all network interfaces and run both IPV4 and IPV6 servers. Do not specify None with port=0 (see lsst.ts.tcpip.OneClientServer for details).

Notes

The MOVE command is rejected if the new position is not within the configured limits.

Attributes Summary

`connected`	Return True if self._reader and self._writer are connected.
`enabled_substate`
`state`
`telemetry_interval`

Methods Summary

`assert_state`(state[, enabled_substate])	Check the state and, optionally, the substate.
`assert_stationary`()
`basic_close_client`()	Close the connected client socket, if any.
`call_connect_callback`()	A client has connected or disconnected.
`close`()	Close socket server and client socket, and set done_task done.
`close_client`(**kwargs)	Close the connected client (if any) and stop background tasks.
`connect_callback`(server)	Called when the server connection state changes.
`do_clear_error`(command)
`do_enable`(command)
`do_enable_drives`(command)
`do_position_set`(command)
`do_standby`(command)
`end_run_command`(command, cmd_method)	Called when run_command is done.
`get_command_key`(command)	Return the key to command_table.
`read`(n)	Read up to n bytes.
`read_and_dispatch`()	Read and execute one command.
`read_into`(struct)	Read binary data from a stream reader into a `ctypes.Structure`.
`read_json`()	Read JSON data.
`read_loop`()	Read incoming data and handle them.
`read_str`()	Read and decode a terminated str; strip the terminator.
`readexactly`(n)	Read exactly n bytes.
`readline`()	Read a sequence of bytes ending with `\n`.
`readuntil`([separator])	Read one line, where “line” is a sequence of bytes ending with `separator`.
`run_command`(command)	Run a command and wait for the reply.
`set_state`(state)	Set the current state and substates.
`start`(**kwargs)	Start the TCP/IP server.
`telemetry_loop`()	Write configuration once, then telemetry at regular intervals.
`update_and_get_header`(frame_id)	Update the config or telemetry header and return it and the time.
`update_telemetry`(curr_tai)	Update self.client.telemetry.
`write`(data)	Write data and call `drain`.
`write_command_status`(counter, status[, ...])	Write a command status.
`write_config`()	Write the current configuration.
`write_from`(*structs)	Write binary data from one or more `ctypes.Structure`s.
`write_json`(data)	Write data in JSON format.
`write_str`(line)	Encode, terminate, and write a str.
`writelines`(lines)	Write an iterable of bytes and call `drain`.

Attributes Documentation

connected¶

Return True if self._reader and self._writer are connected.

Note: if the other end drops the connection and if you are not trying to read data (e.g. in a background loop), then it takes the operating system awhile to realize the connection is lost. So this can return true for some unknown time after the connection has been dropped.

enabled_substate¶

state¶

telemetry_interval = 0.1¶

Methods Documentation

assert_state(state: int, enabled_substate: int | None = None) → None¶

Check the state and, optionally, the substate.

Parameters:

stateint: Required state.
enabled_substateint or None, optional: Required enabled substate, or None to not check.

Raises:

CommandError: If the state is not as expected.

assert_stationary() → None¶

async basic_close_client() → None¶

Close the connected client socket, if any.

Also:

Reset self.connected_task to a new Future.
Call connect_callback, if a client was connected.

Unlike close_client, this does not touch self.should_be_connected.

Always safe to call.

async call_connect_callback() → None¶: A client has connected or disconnected.

async close() → None¶

Close socket server and client socket, and set done_task done.

Call connect_callback if a client was connected.

Always safe to call.

async close_client(**kwargs: dict[str, Any]) → None¶: Close the connected client (if any) and stop background tasks.

async connect_callback(server: OneClientReadLoopServer) → None¶

Called when the server connection state changes.

If connected: start the command and telemetry loops. If not connected: stop the command and telemetry loops.

async do_clear_error(command: Command) → None¶

async do_enable(command: Command) → None¶

async do_enable_drives(command: Command) → None¶

async do_position_set(command: Command) → None¶

async do_standby(command: Command) → None¶

async end_run_command(command: Command, cmd_method: Coroutine) → None¶

Called when run_command is done.

Can be used to clear the set position.

get_command_key(command: Command) → Any¶: Return the key to command_table.

async read(n: int) → bytes¶

Read up to n bytes.

Parameters:

nint: The number of bytes to read. If -1 then block until the other end closes its writer, then return all data seen.

Raises:

ConnectionError: If the connection is lost before, or while, reading.

async read_and_dispatch() → None¶: Read and execute one command.

async read_into(struct: Structure) → None¶

Read binary data from a stream reader into a ctypes.Structure.

Parameters:

structctypes.Structure: Structure to set.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before n bytes can be read. Use the IncompleteReadError.partial attribute to get the partially read data.

async read_json() → Any¶

Read JSON data.

Read the data with read_str and return the json-decoded result.

Returns:

datatyping.Any: Data decoded from JSON.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.
TypeError: If the data are of a type that cannot be decoded from JSON.
json.JSONDecodeError: If the data cannot be decoded from JSON.

async read_loop() → None¶

Read incoming data and handle them.

The actual reading is deferred to the read_and_dispatch method and needs to be implemented by subclasses.

async read_str() → str¶

Read and decode a terminated str; strip the terminator.

Read until self.terminator, strip the terminator, and decode the data as self.encoding with strict error handling.

Returns:

linestr: Line of data, as a str with the terminator stripped.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.
UnicodeError: If decoding fails.

async readexactly(n: int) → bytes¶

Read exactly n bytes.

Parameters:

nint: The number of bytes to read.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before n bytes can be read. Use the IncompleteReadError.partial attribute to get the partially read data.

async readline() → bytes¶

Read a sequence of bytes ending with \n.

If EOF is received and \n was not found, the method returns partially read data.

Raises:

ConnectionError: If the connection is lost before, or while, reading.

async readuntil(separator: bytes = b'\n') → bytes¶

Read one line, where “line” is a sequence of bytes ending with separator.

Read data from the stream until separator is found.

On success, the data and separator will be removed from the internal buffer (consumed). Returned data will include the separator at the end.

See also read_str, which is more convenient for most use cases.

Parameters:

separatorbytes: The desired separator. The default matches the standard library, rather than using terminator.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.

async run_command(command: Command) → float | None¶

Run a command and wait for the reply.

Parameters:

commandCommand: The command to run. This method sets the commander and counter fields.

Returns:

durationfloat or None: Estimated duration (seconds) until the command is finished. 0 if the command is already done or almost already done.

Raises:

CommandError: If the command fails.

set_state(state: IntEnum) → None¶

Set the current state and substates.

Parameters:

statelsst.ts.xml.enums.MTHexapod.ControllerState or int: New state.

Notes

Sets the substates as follows:

lsst.ts.xml.enums.MTHexapod.EnabledSubstate.STATIONARY if state == lsst.ts.xml.enums.MTHexapod.ControllerState.ENABLED

async start(**kwargs: Any) → None¶

Start the TCP/IP server.

This is called automatically by the constructor, and is not intended to be called by the user. It is a public method so that subclasses can override it.

Parameters:

**kwargsdict [str, typing.Any]: Additional keyword arguments for asyncio.start_server, beyond host and port.

Raises:

RuntimeError: If start has already been called and has successfully constructed a server.

async telemetry_loop() → None¶: Write configuration once, then telemetry at regular intervals.

update_and_get_header(frame_id: FrameId) → tuple[lsst.ts.hexrotcomm.structs.Header, float]¶

Update the config or telemetry header and return it and the time.

Call this prior to writing configuration or telemetry.

Parameters:

frame_idFrameId: Frame ID of header to write.

Returns:

headerstructs.Header: The header.
curr_taifloat: Current time in header timestamp (TAI, unix seconds).

async update_telemetry(curr_tai: float) → None¶

Update self.client.telemetry.

Parameters:

curr_taifloat: Time at which to compute telemetry (TAI, unix seconds). This is the time in the header, which is (approximately) the current time.

async write(data: bytes) → None¶

Write data and call drain.

Parameters:

databytes: The data to write.

Raises:

ConnectionError: If self.connected false before writing begins.

async write_command_status(counter: int, status: CommandStatusCode, duration: float | None = 0.0, reason: str = '') → None¶

Write a command status.

Parameters:

counterint: counter field of command being acknowledged.
statusCommandStatusCode: Command status code.
durationfloat or None, optional: Estimated duration. None is treated as 0.
reasonstr, optional: Reason for failure. Should be non-blank if and only if the command failed.

Raises:

ConnectionError: If not connected.

async write_config() → None¶

Write the current configuration.

Raises:

RuntimeError: If not connected.

async write_from(*structs: Structure) → None¶

Write binary data from one or more ctypes.Structures.

Parameters:

structslist [ctypes.Structure]: Structures to write.

Raises:

ConnectionError: If self.connected false before writing begins.

async write_json(data: Any) → None¶

Write data in JSON format.

Encode the data as json and write the result with write_str.

Parameters:

dataany: The data to be written. Typically a dict, but any json-encodable data is acceptable.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
UnicodeError: If encoding fails.
json.JSONEncodeError: If the data cannot be json-encoded.

async write_str(line: str) → None¶

Encode, terminate, and write a str.

Encode the str as self.encoding with strict error handling, and append self.terminator.

Parameters:

linestr: The line of data to be written.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
UnicodeError: If encoding fails.

async writelines(lines: Iterable) → None¶

Write an iterable of bytes and call drain.

Parameters:

linescollections.abc.Iterable [bytes]: The data to write, as an iterable collection of bytes.

Raises:

ConnectionError: If self.connected false before writing begins.

Navigation

SimpleMockController¶