CHIRPManager module

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

CHIRPManger module provides classes for managing CHIRP multicasts within Constellation Satellites.

class core.chirpmanager.CHIRPManager(name: str, group: str, interface: list[str] | None, **kwds: Any)

Bases: BaseSatelliteFrame

Manages service discovery and sends multicast messages via the CHIRP protocol.

Listening and reacting to CHIRP multicast messages is implemented in a dedicated thread that can be started after the class has been instantiated.

Discovered services are added to an internal cache. Callback methods can be registered either by calling register_request() or by using the @chirp_callback() decorator. The callback will be added to the satellite’s internal task queue once the corresponding service has been offered by other satellites via multicast.

Offered services can be registered via register_offer() and are announced on incoming request messages or via emit_offers().

emit_depart() → None

Emit DEPART message for all registered services.

emit_offers(serviceid: CHIRPServiceIdentifier | None = None) → None

Emit messages all registered services matching serviceid.

Specify None for all registered services.

emit_requests() → None

Emit messages for all requests registered via register_request().

get_discovered(serviceid: CHIRPServiceIdentifier) → list[DiscoveredService]

Return a list of already discovered services for a given identifier.

register_offer(serviceid: CHIRPServiceIdentifier, port: int) → None

Register new offered service or overwrite existing service.

register_request(serviceid: CHIRPServiceIdentifier, callback: Callable[[B, DiscoveredService], None]) → None

Register new callback for ServiceIdentifier.

request(serviceid: CHIRPServiceIdentifier) → None

Request specific service.

Should already have a registered callback via register_request(), or any incoming OFFERS will go unnoticed.

class core.chirpmanager.DiscoveredService(host_uuid: UUID, serviceid: CHIRPServiceIdentifier, address: str, port: int, alive: bool = True)

Bases: object

Class to hold discovered service data.

core.chirpmanager.chirp_callback(request_service: CHIRPServiceIdentifier) → Callable[[Callable[[P], T]], Callable[[P], T]]

Mark a function as a callback for CHIRP service requests.

core.chirpmanager.get_chirp_callbacks(cls: object) → dict[CHIRPServiceIdentifier, Callable[[B, DiscoveredService], None]]

Loop over all class methods and return those marked as CHIRP callback.