acme.plugins.services.TimeManager.TimeManager

class documentation

class TimeManager(object):

View In Hierarchy

Managing time related CSE functions.

Method addLoSTimeSyncBeacon Add a Loss-of-sync timeSyncBeacon resource.
Method addPeriodicTimeSyncBeacon Add a worker for a periodic timeSyncBeacon resource.
Method addTimeSyncBeacon Add a timeSyncBeacon resource.
Method isLossOfSynchronization Check if a request is a loss-of-synchronization request.
Method removeLosTimeSyncBeacon Remove a Loss-of-sync timeSyncBeacon resource.
Method removePeriodicTimeSyncBeacon Remove a periodic timeSyncBeacon resource. A running worker is stopped.
Method removeTimeSyncBeacon Remove a timeSyncBeacon resource.
Method requestReveivedHandler Handle a received request.
Method responseReveivedHandler Handle a received response.
Method restart Restart the time manager services.
Method start Initialize the TimeManager.
Method stop Shutdown the TimeManager.
Method updateTimeSyncBeacon Update a timeSyncBeacon resource.
Class Variable __slots__ Define slots for instance variables.
Class Variable notification Injected NotificationManager instance.
Class Variable storage Injected Storage instance.
Class Variable timeManager Reference to the TimeManager plugin instance.
Instance Variable cseActiveSchedule List of active schedules when the CSE is active and will process requests.
Instance Variable losTimeSyncBeacons Table for Loss-of-sync timeSyncBeacons.
Instance Variable periodicTimeSyncBeacons Table for periodic timeSyncBeacons.
Method _getAllPeriodicTimeSyncBeacons Get all periodic timeSyncBeacons from the storage.
Method _stopPeriodicBeacons Stop all the running periodic timers.
def addLoSTimeSyncBeacon(self, tsb: TSB):

Add a Loss-of-sync timeSyncBeacon resource.

Parameters
tsb:TSBtimeSyncBeacon resource
def addPeriodicTimeSyncBeacon(self, tsb: TSB):

Add a worker for a periodic timeSyncBeacon resource.

Parameters
tsb:TSBtimeSyncBeacon resource
def addTimeSyncBeacon(self, tsb: TSB):

Add a timeSyncBeacon resource.

Parameters
tsb:TSBtimeSyncBeacon resource
def isLossOfSynchronization(self, req: CSERequest) -> str | None:

Check if a request is a loss-of-synchronization request.

Parameters
req:CSERequestThe request.
Returns
str | NoneThe duration of the loss-of-synchronization or None.
def removeLosTimeSyncBeacon(self, tsb: TSB):

Remove a Loss-of-sync timeSyncBeacon resource.

Parameters
tsb:TSBThe timeSyncBeacon resource.
def removePeriodicTimeSyncBeacon(self, tsb: TSB):

Remove a periodic timeSyncBeacon resource. A running worker is stopped.

Parameters
tsb:TSBThe timeSyncBeacon resource.
def removeTimeSyncBeacon(self, tsb: TSB):

Remove a timeSyncBeacon resource.

Parameters
tsb:TSBThe timeSyncBeacon resource
@onEvent(eventManager.requestReceived)
def requestReveivedHandler(self, eventData: EventData):

Handle a received request.

Parameters
eventData:EventDataThe event data containing the request.
@onEvent(eventManager.responseReceived)
def responseReveivedHandler(self, eventData: EventData):

Handle a received response.

Parameters
eventData:EventDataThe event data containing the response.
def restart(self):

Restart the time manager services.

def start(self):

Initialize the TimeManager.

def stop(self) -> bool:

Shutdown the TimeManager.

Returns
boolBoolean, always True.
def updateTimeSyncBeacon(self, tsb: TSB, originalBcnc: BeaconCriteria):

Update a timeSyncBeacon resource.

Parameters
tsb:TSBThe timeSyncBeacon resource
originalBcnc:BeaconCriteriaThe original beacon criteria
__slots__: tuple[str, ...] =

Define slots for instance variables.

notification: NotificationManager =

Injected NotificationManager instance.

storage: Storage =

Injected Storage instance.

timeManager: Any | None =

Reference to the TimeManager plugin instance.

cseActiveSchedule: list[str] =

List of active schedules when the CSE is active and will process requests.

losTimeSyncBeacons: dict[str, tuple[float, str]] =

Table for Loss-of-sync timeSyncBeacons.

periodicTimeSyncBeacons: dict[str, BackgroundWorker] =

Table for periodic timeSyncBeacons.

def _getAllPeriodicTimeSyncBeacons(self) -> list[TSB]:

Get all periodic timeSyncBeacons from the storage.

Returns
list[TSB]List of periodic timeSyncBeacons
def _stopPeriodicBeacons(self):

Stop all the running periodic timers.