MockMTRotatorController

class lsst.ts.mtrotator.MockMTRotatorController(log: Logger, port: int = 0, initial_state: ControllerState = ControllerState.STANDBY)

Bases: BaseMockController

Mock MT rotator controller that talks over TCP/IP.

Parameters:
loglogging.Logger

Logger.

portint (optional)

Port for telemetry and configuration; if nonzero then the command port will be one larger. Specify 0 to choose random values for both ports; this is recommended for unit tests, to avoid collision with other tests. Do not specify 0 with host=None (see Raises section).

initial_stateControllerState (optional)

Initial state of mock controller.

Raises:
ValueError

If host=None and port=0. See CommandTelemetryServer for details.

Notes

To start the mock controller:

ctrl = MockRotatorController(…) await ctrl.connect_task

To stop the server:

await ctrl.stop()

Known Limitations

  • Constant-velocity motion is not supported.

  • The odometer resets to zero each time the mock controller is constructed.

  • The motor current and torque are scaled versions of current acceleration, with a wild guess as to scale.

  • Motor current and torque exactly match for motors A and B.

  • Many telemetry fields are not set at all.

Attributes Summary

connected

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

current_per_acceleration

enabled_substate

state

telemetry_interval

torque_per_acceleration

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()

Kill command and telemetry tasks and close the connections.

close_client(**kwargs)

Close the connected client (if any) and stop background tasks.

connect_callback(server)

Called when the server connection state changes.

do_clearError(data)

do_clear_error(command)

do_config_accel(command)

do_config_accel_emergency(command)

do_config_jerk(command)

do_config_jerk_emergency(command)

do_config_vel(command)

do_constant_velocity(command)

do_enable(command)

do_enable_drives(command)

do_fault(command)

do_move_point_to_point(command)

do_position_set(command)

do_set_constant_vel(command)

do_standby(command)

do_stop(command)

do_track(command)

do_track_vel_cmd(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.

tracking_timer()

If this times out then go into a FAULT state.

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.Structures.

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.

current_per_acceleration = 5
enabled_substate
state
telemetry_interval = 0.1
torque_per_acceleration = 0.002

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

Kill command and telemetry tasks and close the connections.

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_clearError(data: Command) None
async do_clear_error(command: Command) None
async do_config_accel(command: Command) None
async do_config_accel_emergency(command: Command) None
async do_config_jerk(command: Command) None
async do_config_jerk_emergency(command: Command) None
async do_config_vel(command: Command) None
async do_constant_velocity(command: Command) None
async do_enable(command: Command) None
async do_enable_drives(command: Command) None
async do_fault(command: Command) None
async do_move_point_to_point(command: Command) None
async do_position_set(command: Command) None
async do_set_constant_vel(command: Command) None
async do_standby(command: Command) None
async do_stop(command: Command) None
async do_track(command: Command) None
async do_track_vel_cmd(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: Command) 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.

async tracking_timer() None

If this times out then go into a FAULT state.

Used to make sure TRACK commands arrive often enough.

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.