Controller Module

SPDX-FileCopyrightText: 2024 DESY and the Constellation authors SPDX-License-Identifier: EUPL-1.2

class core.controller.BaseController(group: str, **kwargs: Any)

Bases: MonitoringSender, CHIRPManager, HeartbeatChecker

Simple controller class to send commands to a Constellation.

await_n_satellites(satellites: int, timeout: int = 60) → None

Blocks until a defined number of satellites is connected.

await_satellites(satellites: list[str], timeout: int = 60) → None

Blocks until all desired satellites are connected.

await_state(target: SatelliteState, timeout: int = 60) → None

Blocks until the desired global state of the connected satellites is reached with check for state changes.

command(cmd: str, payload: Any, sat_type: str | None, sat_name: str | None) → Any

Wrapper for _command_satellite function. Handle sending commands to all hosts

property constellation: SatelliteArray

Returns the currently active SatelliteArray of controlled Satellites.

property last_state_change: case_insensitive_dict[datetime]

Return a dictionary of connected Satellite’s last state change.

Based on heartbeat information.

reentry() → None

Stop the controller.

property state: ControllerState

Return the global state of all connected Satellite’s state.

Based on heartbeat information.

property states: case_insensitive_dict[SatelliteState]

Return an up-to-date dictionary of connected Satellite’s state.

Based on heartbeat information.

property status: str

Return the global status summary of the Constellation.

Based on heartbeat information.

terminate() → None

Terminate the satellite.

class core.controller.CommandWrapper(handler: Callable[[str, Any, str | None, str | None], tuple[str, Any, dict[str, Any] | None]], sat_name: str | None, sat_type: str | None, cmd: str)

Bases: object

Class to wrap command calls.

Allows to mimic the signature of the Satellite command being wrapped.

call(payload: Any = None) → tuple[str, Any, dict[str, Any] | None]

Perform call. This doc string will be overwritten.

class core.controller.ControllerState(*values)

Bases: Enum

Global Controller state.

ERROR = 240

INIT = 32

NEW = 16

ORBIT = 48

RUN = 64

TRANSITIONING = 85

class core.controller.SatelliteArray(group: str, handler: Callable[[str, Any, str | None, str | None], tuple[str, Any, dict[str, Any] | None]])

Bases: object

Provide object-oriented control of connected Satellites.

get_satellite(sat_type: str, sat_name: str) → SatelliteCommLink

Return a link to a Satellite given by its type and name.

property satellites: dict[str, SatelliteCommLink]

Return the dict of Satellites names and their SatelliteCommLink.

class core.controller.SatelliteCommLink(satellite_name: str, satellite_type: str)

Bases: SatelliteTypeCommLink

A link to a Satellite.

class core.controller.SatelliteResponse

Bases: object

Class to hold the response of a Satellite to a CSCP command.

errmsg: str

meta: dict[str, Any] | None

msg: str

payload: Any

success: bool

class core.controller.SatelliteTypeCommLink(satellite_type: str)

Bases: object

A link to a Satellite Type.

class core.controller.SatelliteUpdate(*values)

Bases: Enum

Event type for satellite connection callbacks.

ADDED = 1

REMOVED = 2

class core.controller.ScriptableController(group: str, log_level: str = 'INFO', name: str = 'ScriptableController', interface: list[str] | None = None)

Bases: BaseController

Simple controller class to which initializes logging and networking.