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: CHIRPBroadcaster, HeartbeatChecker

Simple controller class to send commands to a Constellation.

await_state(target: SatelliteState, timeout: int = 60) → None#: Blocks until the desired global SatelliteState of the controller satellites is reached.

command(payload: Any = None, sat: str = '', satcls: str = '', cmd: str = '') → Any#: Wrapper for _command_satellite function. Handle sending commands to all hosts

property constellation: SatelliteArray#: Returns the currently active SatelliteArray of controlled Satellites.

reentry() → None#: Stop the controller.

property state: ControllerState#

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

Based on heartbeat information.

property states: dict[str, 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.

class core.controller.CommandWrapper(handler: Callable[[str, str, str, Any], Tuple[str, Any, dict[str, Any] | None]], sat: str, satcls: str, 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, str, str, Any], Tuple[str, Any, dict[str, Any] | None]])#

Bases: object

Provide object-oriented control of connected Satellites.

get_satellite(sat_class: str, sat_name: str) → SatelliteCommLink#: Return a link to a Satellite given by its class and name.

property satellites: dict[str, SatelliteCommLink]#: Return the dict of Satellites names and their SatelliteCommLink.

class core.controller.SatelliteClassCommLink(name: str)#

Bases: object

A link to a Satellite Class.

class core.controller.SatelliteCommLink(name: str, cls: str)#

Bases: SatelliteClassCommLink

A link to a Satellite.

class core.controller.SatelliteResponse#

Bases: object

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

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

Bases: BaseController

Simple controller class to which initializes logging and networking.

core.controller.main(args: Any = None) → None#

Start a Constellation CSCP controller.

This Controller provides a command-line interface to the selected Constellation group via IPython terminal.