API

Configuration Database

SDP State

High-level SDP state API.

class sip_config_db.states.sdp_state.SDPState[source]

SDP state data object.

allowed_state_transitions

Get dictionary of allowed state transitions.

allowed_states

Get list of allowed object states.

allowed_target_states

Get dictionary of allowed target states / commands.

current_state

Get the current state.

current_timestamp

Get the current state timestamp.

get_event_queue(subscriber: str)

Get an event queue for the specified subscriber.

static get_subscribers() → List[str]

Get the list of subscribers to state events.

Returns:

List[str], list of subscriber names.

id

Return the object id.

is_target_state_allowed(value)

Test if a transition is allowed.

Args:

value (str): New value for target state

Returns:

bool, transition is allowed

publish(event_type: str, event_data: dict = None)

Publish an state event.

Args:

event_type (str): Type of event. event_data (dict, optional): Event data.

static subscribe(subscriber: str) → sip_config_db._events.event_queue.EventQueue

Subscribe to state events.

Args:

subscriber (str): Subscriber name.

Returns:

events.EventQueue, Event queue object for querying events.

target_state

Get the target state.

target_timestamp

Get the target state timestamp.

update_current_state(value: str, force: bool = False) → datetime.datetime

Update the current state.

Args:

value (str): New value for sdp state force (bool): If true, ignore allowed transitions

Returns:

datetime, update timestamp

Raises:

ValueError: If the specified current state is not allowed.

update_target_state(value: str, force: bool = True) → datetime.datetime

Set the target state.

Args:

value (str): New value for target state force (bool): If true, ignore allowed transitions

Returns:

datetime, update timestamp

Raises:

RuntimeError, if it is not possible to currently set the target state. ValueError, if the specified target stat is not allowed.

Service States

High-level service state API.

class sip_config_db.states.service_state.ServiceState(subsystem: str, name: str, version: str)[source]

SDP state data object.

static get_service_state_object_id(subsystem: str, name: str, version: str) → str[source]

Return service state data object key.

Args:

subsystem (str): Subsystem the service belongs to name (str): Name of the Service version (str): Version of the Service

Returns:

str, Key used to store the service state data object

name

Get the name of the service.

subsystem

Get the subsystem the service belongs to.

version

Get the version of the service.

Services

High-level service state API.

class sip_config_db.states.service_state.ServiceState(subsystem: str, name: str, version: str)[source]

SDP state data object.

static get_service_state_object_id(subsystem: str, name: str, version: str) → str[source]

Return service state data object key.

Args:

subsystem (str): Subsystem the service belongs to name (str): Name of the Service version (str): Version of the Service

Returns:

str, Key used to store the service state data object

name

Get the name of the service.

subsystem

Get the subsystem the service belongs to.

version

Get the version of the service.

Utility modules

Misc functions for interacting with datetime.

sip_config_db.utils.datetime_utils.datetime_from_isoformat(value: str)[source]

Return a datetime object from an isoformat string.

Args:

value (str): Datetime string in isoformat.

Module to generate test data for sbi and pb client.

FIXME(BMo) Move this to a method on the SchedulingBlockInstance object?

sip_config_db.utils.generate_sbi_config.add_workflow_definitions(sbi_config: dict)[source]

Add any missing SBI workflow definitions as placeholders.

This is a utility function used in testing and adds mock / test workflow definitions to the database for workflows defined in the specified SBI config.

Args:

sbi_config (dict): SBI configuration dictionary.

sip_config_db.utils.generate_sbi_config.generate_pb_config(pb_id: str, pb_config: dict = None, workflow_config: dict = None) → dict[source]

Generate a PB configuration dictionary.

Args:

pb_id (str): Processing Block Id pb_config (dict, optional) PB configuration. workflow_config (dict, optional): Workflow configuration

Returns:

dict, PB configuration dictionary.

sip_config_db.utils.generate_sbi_config.generate_sb(date: datetime.datetime, project: str, programme_block: str) → dict[source]

Generate a Scheduling Block data object.

Args:

date (datetime.datetime): UTC date of the SBI project (str): Project Name programme_block (str): Programme

Returns:

str, Scheduling Block Instance (SBI) ID.

sip_config_db.utils.generate_sbi_config.generate_sbi_config(num_pbs: int = 3, project: str = 'sip', programme_block: str = 'sip_demos', pb_config: Union[dict, List[dict]] = None, workflow_config: Union[dict, List[dict]] = None, register_workflows=False) → dict[source]

Generate a SBI configuration dictionary.

Args:

num_pbs (int, optional): Number of Processing Blocks (default = 3) project (str, optional): Project to associate the SBI with. programme_block (str, optional): SBI programme block pb_config (dict, List[dict], optional): PB configuration workflow_config (dict, List[dict], optional): Workflow configuration register_workflows (bool, optional): If true also register workflows.

Returns:

dict, SBI configuration dictionary

sip_config_db.utils.generate_sbi_config.generate_sbi_json(num_pbs: int = 3, project: str = 'sip', programme_block: str = 'sip_demos', pb_config: Union[dict, List[dict]] = None, workflow_config: Union[dict, List[dict]] = None, register_workflows=True) → str[source]

Return a JSON string used to configure an SBI.

sip_config_db.utils.generate_sbi_config.generate_version(max_major: int = 1, max_minor: int = 7, max_patch: int = 15) → str[source]

Select a random version.

Args:

max_major (int, optional) maximum major version max_minor (int, optional) maximum minor version max_patch (int, optional) maximum patch version

Returns:

str, Version String