PneumaticsSimulator

class lsst.ts.atpneumaticssimulator.PneumaticsSimulator(host: str, cmd_evt_port: int, telemetry_port: int)

Bases: AtSimulator

Simulate the ATPneumatics system.

Attributes:

hoststr

The simulator host.

cmd_evt_portint

The command and events port.

telemetry_portint

The telemetry port.

Methods Summary

cmd_evt_connect_callback(server)	Callback function for when a cmd/evt client connects or disconnects.
cmd_evt_dispatch_callback(data)	Asynchronous function to call when data is read and dispatched.
configure([m1_covers_close_time, ...])	Set configuration.
disable(*, sequence_id)	Switch to sal_enums.State.DISABLED.
do_close_instrument_air_valve(sequence_id)
do_close_m1_cell_vents(sequence_id)
do_close_m1_cover(sequence_id)
do_close_master_air_supply(sequence_id)
do_m1_close_air_valve(sequence_id)
do_m1_open_air_valve(sequence_id)
do_m1_set_pressure(sequence_id, pressure)
do_m2_close_air_valve(sequence_id)
do_m2_open_air_valve(sequence_id)
do_m2_set_pressure(sequence_id, pressure)
do_open_instrument_air_valve(sequence_id)
do_open_m1_cell_vents(sequence_id)
do_open_m1_cover(sequence_id)
do_open_master_air_supply(sequence_id)
enable(*, sequence_id)	Switch to sal_enums.State.ENABLED.
fault()	Switch to sal_enums.State.FAULT.
initialize()	Initialize events and telemetry.
load_schemas()	Load the JSON schemas needed to validate the incoming and outgoing JSON messages.
send_detailed_state_events()	Send detailed state events.
set_cell_vents_events(closed, opened)	Set m1VentsLimitSwitches, m1VentsPosition and cellVentsState events.
set_m1_cover_events(closed, opened)	Set m1CoverLimitSwitches and m1CoverState events.
standby(*, sequence_id)	Switch to sal_enums.State.STANDBY.
telemetry_connect_callback(server)	Callback function for when a tel client connects or disconnects.
telemetry_dispatch_callback(data)	Asynchronous function to call when data are read and dispatched.
telemetry_loop()	Output telemetry and events that have changed
update_telemetry()	Output all telemetry data messages.
verify_data(data)	Verify the format and values of the data.
write_ack_response(sequence_id)	Write an ACK response.
write_fail_response(sequence_id)	Write a FAIL response.
write_noack_response(sequence_id)	Write a NOACK response.
write_success_response(sequence_id)	Write a SUCCESS response.

Methods Documentation

async cmd_evt_connect_callback(server: OneClientServer) → None

Callback function for when a cmd/evt client connects or disconnects.

When a cmd/evt client connects, background tasks are started and events are sent. When the cmd/evt client disconnects, all background tasks get stopped.

async cmd_evt_dispatch_callback(data: Any) → None

Asynchronous function to call when data is read and dispatched.

The received data is validated and then dispatched to the corresponding function as listed in self.dispatch_dict. If the data is invalid then a noack response is sent back to the client and the data is not dispatched.

Parameters:

datadict`[`str, typing.Any]

The data sent to the server.

async configure(m1_covers_close_time: float = 20.0, m1_covers_open_time: float = 20.0, cell_vents_close_time: float = 5.0, cell_vents_open_time: float = 1.0, m1_pressure: float = 5.0, m2_pressure: float = 6.0, main_pressure: float = 10.0, cell_load: float = 100.0) → None

Set configuration.

Parameters:

m1_covers_close_timefloat

Time to close M1 mirror covers [sec].

m1_covers_open_timefloat

Time to open M1 mirror covers [sec].

cell_vents_close_timefloat

Time to close cell vents [sec].

cell_vents_open_timefloat

Time to open cell vents [sec].

m1_pressurefloat

Initial M1 air pressure [Pa].

m2_pressurefloat

Initial M2 air pressure [Pa].

main_pressurefloat

Initial main air pressure [Pa].

cell_loadfloat

Initial cell load [kg].

async disable(*, sequence_id: int) → None

Switch to sal_enums.State.DISABLED.

async do_close_instrument_air_valve(sequence_id: int) → None

async do_close_m1_cell_vents(sequence_id: int) → None

async do_close_m1_cover(sequence_id: int) → None

async do_close_master_air_supply(sequence_id: int) → None

async do_m1_close_air_valve(sequence_id: int) → None

async do_m1_open_air_valve(sequence_id: int) → None

async do_m1_set_pressure(sequence_id: int, pressure: float) → None

async do_m2_close_air_valve(sequence_id: int) → None

async do_m2_open_air_valve(sequence_id: int) → None

async do_m2_set_pressure(sequence_id: int, pressure: float) → None

async do_open_instrument_air_valve(sequence_id: int) → None

async do_open_m1_cell_vents(sequence_id: int) → None

async do_open_m1_cover(sequence_id: int) → None

async do_open_master_air_supply(sequence_id: int) → None

async enable(*, sequence_id: int) → None

Switch to sal_enums.State.ENABLED.

async fault() → None

Switch to sal_enums.State.FAULT.

async initialize() → None

Initialize events and telemetry.

• instrumentState

• m1CoverLimitSwitches

• m1CoverState

• m1SetPressure

• m1State

• cellVentsState (should be named m1VentsState)

• m1VentsLimitSwitches

• m1VentsPosition

• m2SetPressure

• m2State

• mainValveState

• powerStatus

load_schemas() → None

Load the JSON schemas needed to validate the incoming and outgoing JSON messages.

async send_detailed_state_events() → None

Send detailed state events.

This method is called when going to FAULT state. Subclasses may implement this such that the device specific behavior is mimicked.

async set_cell_vents_events(closed: bool, opened: bool) → None

Set m1VentsLimitSwitches, m1VentsPosition and cellVentsState events.

Output any changes.

Parameters:

closedbool

Are the closed switches active?

openedbool

Are the opened switches active?

async set_m1_cover_events(closed: bool, opened: bool) → None

Set m1CoverLimitSwitches and m1CoverState events.

Output any changes.

Parameters:

closedbool

Are the closed switches active?

openedbool

Are the opened switches active?

async standby(*, sequence_id: int) → None

Switch to sal_enums.State.STANDBY.

async telemetry_connect_callback(server: OneClientServer) → None

Callback function for when a tel client connects or disconnects.

When a tel client connects, the telemetry loop is started. When the tel client disconnects, the telemetry loop is stopped.

async telemetry_dispatch_callback(data: Any) → None

Asynchronous function to call when data are read and dispatched.

The received data is ignored since the telemetry server is only supposed to send data and not to receive any.

Parameters:

datadict`[`str, typing.Any]

The data sent to the server.

async telemetry_loop() → None

Output telemetry and events that have changed

Notes

Here are the telemetry topics that are output:

• m1AirPressure

• m2AirPressure

• mainAirSourcePressure

• loadCell

See update_events for the events that are output.

async update_telemetry() → None

Output all telemetry data messages.

async verify_data(data: dict[str, Any]) → bool

Verify the format and values of the data.

The format of the data is described at https://github.com/lsst-ts/ts_labview_tcp_json as well as in the JSON schemas in the schemas directory.

Parameters:

datadict of any

The dict to be verified.

Returns:

bool:

Whether the data follows the correct format and has the correct contents or not.

async write_ack_response(sequence_id: int) → None

Write an ACK response.

Parameters:

sequence_idint

The command sequence id.

async write_fail_response(sequence_id: int) → None

Write a FAIL response.

Parameters:

sequence_idint

The command sequence id.

async write_noack_response(sequence_id: int) → None

Write a NOACK response.

Parameters:

sequence_idint

The command sequence id.

async write_success_response(sequence_id: int) → None

Write a SUCCESS response.

Parameters:

sequence_idint

The command sequence id.