ska_oso_oet

Reading ska_oso_oet.ini file value and initializing constant of feature toggle with enabling event based polling/pubsub

ska_oso_oet.main

class ska_oso_oet.main.ActivityServiceWorker(name: str, startup_event: Event, shutdown_event: Event, event_q: MPQueue, work_q: MPQueue, mp_context: BaseContext, *args, **kwargs)

ActivityServiceWorker listens for user request messages, calling the appropriate ActivityService function and broadcasting its response.

__init__(name: str, startup_event: Event, shutdown_event: Event, event_q: MPQueue, work_q: MPQueue, mp_context: BaseContext, *args, **kwargs)

Create a new QueueProcWorker.

The events and MPQueues passed to this constructor should be created and managed within the scope of a MainContext context manager and shared with other ProcWorkers, so that the communication queues are shared correctly between Python processes and there is a common event that can be set to notify all processes when shutdown is required.

Parameters:

• name – name of this worker

• startup_event – event to trigger when startup is complete

• shutdown_event – event to monitor for shutdown

• event_q – outbox for posting messages to main context

• work_q – inbox message queue for work messages

• args – captures other anonymous arguments

• kwargs – captures other keyword arguments

shutdown() → None

Disconnect republishing function from pypubsub

startup() → None

Connect republishing function to pypubsub.

class ska_oso_oet.main.EventBusWorker(name: str, startup_event: Event, shutdown_event: Event, event_q: MPQueue, work_q: MPQueue, *args, **kwargs)

EventBusWorker converts external inter-process pub/sub messages to and from local intra-process pubsub messages.

EventBusWorker uses the QueueProcWorker’s ‘work queue’ as an inbox for pub/sub EventMessages sent by other ProcWorkers. EventMessages received on this queue are rebroadcast locally as pypubsub messages. Likewise, the EventBusWorker listens to all pypubsub messages broadcast locally, converts them to pub/sub EventQueue messages, and puts them on the ‘main’ queue for transmission to other EventBusWorkers.

main_func(evt: EventMessage) → None

Republish external pub/sub message locally.

QueueProcWorker ensures that main_func is called for every item in the work queue. This function takes that work item - the external pub/sub EventMessage - and rebroadcasts it locally as a pypubsub message.

Parameters:

evt – pub/sub EventMessage to broadcast locally

republish(topic: pubsub.pub.Topic = pubsub.pub.AUTO_TOPIC, **kwargs) → None

Republish a local event over the inter-process event bus.

Parameters:

• topic – message topic, set automatically by pypubsub

• kwargs – any metadata associated with pypubsub message

Returns:

shutdown() → None

Disconnect republishing function from pypubsub

startup() → None

Connect republishing function to pypubsub.

class ska_oso_oet.main.FastAPIWorker(name: str, startup_event: Event, shutdown_event: Event, event_q: MPQueue, work_q: MPQueue, *args, **kwargs)

FastAPIWorker is an EventBusWorker that runs a FastAPI app.

By extending EventBusWorker, FastAPI functions can use pypubsub to subscribe to and publish messages, and these messages will put on the main queue to be broadcast to other EventBusWorkers.

shutdown() → None

Disconnect republishing function from pypubsub

startup() → None

Connect republishing function to pypubsub.

class ska_oso_oet.main.ScriptExecutionServiceWorker(name: str, startup_event: Event, shutdown_event: Event, event_q: MPQueue, work_q: MPQueue, mp_context: BaseContext, *args, **kwargs)

ScriptExecutionService listens for user request messages, calling the appropriate ScriptExecutionService function and broadcasting its response.

Actions that occur in the user request domain (‘user clicked start observation’, ‘user aborted observation using the CLI’, etc.) are broadcast as events. ScriptExecutionServiceWorker listens for events on these topics and triggers the required action in the script execution domain (‘start a script’, ‘abort a script’, etc.).

Currently, the result of the action that occurred in the script execution domain (=the return object from the ScriptExecutionService) is broadcast to the world by the ScriptExecutionServiceWorker. This could change so that the ScriptExecutionService itself sends the message.

__init__(name: str, startup_event: Event, shutdown_event: Event, event_q: MPQueue, work_q: MPQueue, mp_context: BaseContext, *args, **kwargs)

Create a new QueueProcWorker.

The events and MPQueues passed to this constructor should be created and managed within the scope of a MainContext context manager and shared with other ProcWorkers, so that the communication queues are shared correctly between Python processes and there is a common event that can be set to notify all processes when shutdown is required.

Parameters:

• name – name of this worker

• startup_event – event to trigger when startup is complete

• shutdown_event – event to monitor for shutdown

• event_q – outbox for posting messages to main context

• work_q – inbox message queue for work messages

• args – captures other anonymous arguments

• kwargs – captures other keyword arguments

shutdown() → None

Disconnect republishing function from pypubsub

startup() → None

Connect republishing function to pypubsub.

ska_oso_oet.main.main(mp_ctx: BaseContext)

Create the OET components and start an event loop that dispatches messages between them.

Parameters:

logging_config

ska_oso_oet.main.main_loop(main_ctx: MainContext, event_bus_queues: List[MPQueue])

Main message parsing and routing loop, extracted from main() to increase testability.

Parameters:

• main_ctx

• event_bus_queues

Returns:

ska_oso_oet.ui

The ska_oso_oet.ui package contains code that present the OET interface to the outside world. In practical terms, this means the OET application’s REST interface

class ska_oso_oet.ui.Message(*args: Any, **kwargs: Any)

Data that is published as a server-sent event.

ska_oso_oet.ui.messages(shutdown_event: Event) → Generator[Message, None, None]

A generator of Message objects created from received pubsub events

async ska_oso_oet.ui.timeout_handler(_: fastapi.Request, err: werkzeug.exceptions.GatewayTimeout) → fastapi.HTTPException

A custom handler function to deal with GatewayTimeout and return the correct HTTP response.