coordinator module

Submodules

coordinator

Classes:

Alarm(alarm_type, severity, start_time, …)	Representation of alarm status and parameters
ControlSetting(name, value, min_value, …)	Message containing ventilation control parameters.
CoordinatorBase([sim_mode])
CoordinatorLocal([sim_mode])	param sim_mode
CoordinatorRemote([sim_mode])
ProcessManager(sim_mode[, startCommandLine, …])
SensorValues([timestamp, loop_counter, …])	Structured class for communicating sensor readings throughout PVP.
ValueName(value)	Canonical names of all values used in PVP.

Data:

Dict	The central part of internal API.
List	The central part of internal API.

Functions:

get_coordinator([single_process, sim_mode])
get_rpc_client()
init_logger(module_name[, log_level, …])	Initialize a logger for logging events.

class pvp.coordinator.coordinator.CoordinatorBase(sim_mode=False)

Bases: object

Methods:

get_alarms()
get_control(control_setting_name)
get_sensors()
get_target_waveform()
is_running()
kill()
set_breath_detection(breath_detection)
set_control(control_setting)
start()
stop()

get_sensors() → pvp.common.message.SensorValues

get_alarms() → Union[None, Tuple[pvp.alarm.alarm.Alarm]]

set_control(control_setting: pvp.common.message.ControlSetting)

get_control(control_setting_name: pvp.common.values.ValueName) → pvp.common.message.ControlSetting

set_breath_detection(breath_detection: bool)

get_target_waveform()

start()

is_running() → bool

kill()

stop()

class pvp.coordinator.coordinator.CoordinatorLocal(sim_mode=False)

Bases: pvp.coordinator.coordinator.CoordinatorBase

Parameters

sim_mode –

_is_running

.set() when thread should stop

Type

threading.Event

Methods:

__init__([sim_mode])	param sim_mode
get_alarms()
get_control(control_setting_name)
get_sensors()
get_target_waveform()
is_running()	Test whether the whole system is running
kill()
set_breath_detection(breath_detection)
set_control(control_setting)
start()	Start the coordinator.
stop()	Stop the coordinator.

__init__(sim_mode=False)

Parameters

sim_mode –

_is_running

.set() when thread should stop

Type

threading.Event

get_sensors() → pvp.common.message.SensorValues

get_alarms() → Union[None, Tuple[pvp.alarm.alarm.Alarm]]

set_control(control_setting: pvp.common.message.ControlSetting)

get_control(control_setting_name: pvp.common.values.ValueName) → pvp.common.message.ControlSetting

set_breath_detection(breath_detection: bool)

get_target_waveform()

start()

Start the coordinator. This does a soft start (not allocating a process).

is_running() → bool

Test whether the whole system is running

stop()

Stop the coordinator. This does a soft stop (not kill a process)

kill()

class pvp.coordinator.coordinator.CoordinatorRemote(sim_mode=False)

Bases: pvp.coordinator.coordinator.CoordinatorBase

Methods:

get_alarms()
get_control(control_setting_name)
get_sensors()
get_target_waveform()
is_running()	Test whether the whole system is running
kill()	Stop the coordinator and end the whole program
set_breath_detection(breath_detection)
set_control(control_setting)
start()	Start the coordinator.
stop()	Stop the coordinator.

get_sensors() → pvp.common.message.SensorValues

get_alarms() → Union[None, Tuple[pvp.alarm.alarm.Alarm]]

set_control(control_setting: pvp.common.message.ControlSetting)

get_control(control_setting_name: pvp.common.values.ValueName) → pvp.common.message.ControlSetting

set_breath_detection(breath_detection: bool)

get_target_waveform()

start()

Start the coordinator. This does a soft start (not allocating a process).

is_running() → bool

Test whether the whole system is running

stop()

Stop the coordinator. This does a soft stop (not kill a process)

kill()

Stop the coordinator and end the whole program

pvp.coordinator.coordinator.get_coordinator(single_process=False, sim_mode=False) → pvp.coordinator.coordinator.CoordinatorBase

ipc

Classes:

SimpleXMLRPCServer(addr[, requestHandler, …])

Simple XML-RPC server.

Functions:

get_alarms()
get_control(control_setting_name)
get_rpc_client()
get_sensors()
get_target_waveform()
init_logger(module_name[, log_level, …])	Initialize a logger for logging events.
rpc_server_main(sim_mode, serve_event[, …])
set_breath_detection(breath_detection)
set_control(control_setting)

pvp.coordinator.rpc.get_sensors()

pvp.coordinator.rpc.get_alarms()

pvp.coordinator.rpc.set_control(control_setting)

pvp.coordinator.rpc.get_control(control_setting_name)

pvp.coordinator.rpc.set_breath_detection(breath_detection)

pvp.coordinator.rpc.get_target_waveform()

pvp.coordinator.rpc.rpc_server_main(sim_mode, serve_event, addr='localhost', port=9533)

pvp.coordinator.rpc.get_rpc_client()

process_manager

Classes:

ProcessManager(sim_mode[, startCommandLine, …])

class pvp.coordinator.process_manager.ProcessManager(sim_mode, startCommandLine=None, maxHeartbeatInterval=None)

Bases: object

Methods:

heartbeat(timestamp)
restart_process()
start_process()
try_stop_process()

start_process()

try_stop_process()

restart_process()

heartbeat(timestamp)