coordinator module

The coordinator provides an interface between the process threads, and facilitates inter-process communication. It is a wrapper around xml-rpc, which allowed us to use defined data-structures such as SensorValues.

Submodules

coordinator

Classes:

CoordinatorBase([sim_mode])
CoordinatorLocal([sim_mode])	param sim_mode
CoordinatorRemote([sim_mode])

Functions:

get_coordinator([single_process, sim_mode])

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

Bases: object

Methods:

get_sensors()
get_alarms()
set_control(control_setting)
get_control(control_setting_name)
set_breath_detection(breath_detection)
get_breath_detection()
start()
is_running()
kill()
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_breath_detection() → bool

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_sensors()
get_alarms()
set_control(control_setting)
get_control(control_setting_name)
set_breath_detection(breath_detection)
get_breath_detection()
start()	Start the coordinator.
is_running()	Test whether the whole system is running
stop()	Stop the coordinator.
kill()

__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_breath_detection() → bool

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_sensors()
get_alarms()
set_control(control_setting)
get_control(control_setting_name)
set_breath_detection(breath_detection)
get_breath_detection()
start()	Start the coordinator.
is_running()	Test whether the whole system is running
stop()	Stop the coordinator.
kill()	Stop the coordinator and end the whole program

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_breath_detection() → bool

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

Functions:

get_sensors()
get_alarms()
set_control(control_setting)
get_control(control_setting_name)
set_breath_detection(breath_detection)
get_breath_detection()
rpc_server_main(sim_mode, serve_event[, ...])
get_rpc_client()

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

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:

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

start_process()

try_stop_process()

restart_process()

heartbeat(timestamp)