openpectus.aggregator.aggregator

Attributes

logger

EngineDataMap

Classes

FromEngine

FromFrontend

Aggregator

Module Contents

openpectus.aggregator.aggregator.logger
openpectus.aggregator.aggregator.EngineDataMap
class openpectus.aggregator.aggregator.FromEngine(engine_data_map, publisher, webpush_publisher)
Parameters:
_engine_data_map
publisher
webpush_publisher
__str__()
Return type:

str

register_engine_data(engine_data)
Parameters:

engine_data (openpectus.aggregator.models.EngineData)

_try_restore_reconnected_engine_data(engine_data)
Parameters:

engine_data (openpectus.aggregator.models.EngineData)

engine_connected(engine_id)
Parameters:

engine_id (str)

engine_disconnected(engine_id)
Parameters:

engine_id (str)

run_started(msg)
Parameters:

msg (openpectus.protocol.engine_messages.RunStartedMsg)

run_stopped(msg)
Parameters:

msg (openpectus.protocol.engine_messages.RunStoppedMsg)

uod_info_changed(engine_id, readings, commands, uod_definition, plot_configuration, hardware_str, required_roles, data_log_interval_seconds)
Parameters:

  • engine_id (str)

  • readings (list[openpectus.aggregator.models.ReadingInfo])

  • commands (list[openpectus.aggregator.models.CommandInfo])

  • uod_definition (openpectus.aggregator.models.UodDefinition)

  • plot_configuration (openpectus.aggregator.models.PlotConfiguration)

  • hardware_str (str)

  • required_roles (set[str])

  • data_log_interval_seconds (float)

tag_values_changed(engine_id, changed_tag_values)
Parameters:

  • engine_id (str)

  • changed_tag_values (list[openpectus.aggregator.models.TagValue])

abstract __run_id_changed(plot_log_repo, recent_run_repo, engine_data, run_id_tag)

Handles persistance related to start and end of a run

Parameters:
_persist_tag_values(engine_data, plot_log_repo)
Parameters:
runlog_changed(engine_id, run_id, runlog)
Parameters:

  • engine_id (str)

  • run_id (str)

  • runlog (openpectus.aggregator.models.RunLog)

control_state_changed(engine_id, control_state)
Parameters:

  • engine_id (str)

  • control_state (openpectus.aggregator.models.ControlState)

method_state_changed(engine_id, method_state)
Parameters:

  • engine_id (str)

  • method_state (openpectus.aggregator.models.MethodState)

error_log_changed(engine_id, error_log)
Parameters:

  • engine_id (str)

  • error_log (openpectus.aggregator.models.ErrorLog)

publish_engine_disconnected_notification(engine_id)
Parameters:

engine_id (str)

class openpectus.aggregator.aggregator.FromFrontend(engine_data_map, dispatcher, publisher, webpush_publisher)
Parameters:
_engine_data_map
dispatcher
publisher
webpush_publisher
dead_man_switch_user_ids: dict[str, str]
__str__()
Return type:

str

async save_method(engine_id, method, user)
Parameters:
Return type:

int

async request_cancel(engine_id, line_id, user)
Parameters:
Return type:

bool

async request_force(engine_id, line_id, user)
Parameters:
Return type:

bool

async user_subscribed_pubsub(subscriber_id, topics)
Parameters:

  • subscriber_id (str)

  • topics (list[str])

async on_ws_disconnect(subscriber_id)
Parameters:

subscriber_id (str)

async register_active_user(engine_id, user_id, user_name)
Parameters:

  • engine_id (str)

  • user_id (str)

  • user_name (str)

async unregister_active_user(engine_id, user_id)
Parameters:

  • engine_id (str)

  • user_id (str)

webpush_user_subscribed(subscription, user_id)
Parameters:

  • subscription (webpush.WebPushSubscription)

  • user_id (str | None)

Return type:

bool

webpush_notification_preferences_requested(user_id, user_roles)
Parameters:

  • user_id (str | None)

  • user_roles (set[str])

Return type:

openpectus.aggregator.models.WebPushNotificationPreferences

webpush_notification_preferences_posted(preferences)
Parameters:

preferences (openpectus.aggregator.models.WebPushNotificationPreferences)

async add_contributor(unit_id, user_id, user_name)
Parameters:

  • unit_id (str)

  • user_id (str | None)

  • user_name (str)

async excute_command(unit_id, user_id, user_name, msg)
Parameters:
async excute_control_button_command(unit_id, user_id, user_name, msg)
Parameters:
publish_notification_test(user_id)
Parameters:

user_id (None | str)

publish_new_contributor_notification(engine_id, contributor)
Parameters:
class openpectus.aggregator.aggregator.Aggregator(dispatcher, publisher, webpush_publisher, secret='')
Parameters:
_engine_data_map: EngineDataMap

all client data except channels, indexed by engine_id

dispatcher
from_frontend
from_engine
secret = ''
webpush_publisher
__str__()
Return type:

str

shutdown()
create_engine_id(register_engine_msg)

Defines the generation of the engine_id that is uniquely assigned to each engine.

TODO: Considerations:

  • engine name should be machine name

  • uod hash should probably be included

Implications of the registration process - historical data; we should not corrupt historical data by accidentially reusing engine_id - number of cards shown; we should not show many irrelevant cards in frontend of superseeded engine_ids

Parameters:

register_engine_msg (openpectus.protocol.engine_messages.RegisterEngineMsg)

get_registered_engine_data(engine_id)
Parameters:

engine_id (str)

get_all_registered_engine_data()
Return type:

list[openpectus.aggregator.models.EngineData]

has_registered_engine_id(engine_id)
Parameters:

engine_id (str)

Return type:

bool