clock

clock

Classes:

Name	Description
Clock	A class for managing time cycles and converting between timestamp and cycle indices.

Classes

Clock

Clock(min_delta: float = -1)

A class for managing time cycles and converting between timestamp and cycle indices.

This class interacts with an NTP server to synchronize time and supports operations to track cycles, manage timestamps, and calculate the time differences between cycles.

Initialize a Clock instance.

Parameters:

Name	Type	Description	Default
min_delta	float	Minimum time (in seconds) between consecutive cycles. If less than or equal to zero, the cycles will be real-time-based.	-1

Methods:

Name	Description
time2cycle	Convert a given timestamp to the corresponding cycle index.
cycle2time	Convert a cycle index to the corresponding timestamp.
get_time	Get the current time based on the NTP server synchronization.
get_time_as_string	Get the current time as a string (ISO format).
get_cycle	Get the current cycle index.
get_cycle_time	Get the timestamp corresponding to the current cycle.
next_cycle	Move to the next cycle if the minimum delta time has passed or if cycles are not constrained.

Source code in unaiverse/clock.py

def __init__(self, min_delta: float = -1):
    """Initialize a Clock instance.

    Args:
        min_delta (float): Minimum time (in seconds) between consecutive cycles.
                            If less than or equal to zero, the cycles will be real-time-based.
    """
    self.min_delta = min_delta  # Min-time passed between consecutive cycles (seconds) - if <=0, it is real-time
    self.cycle = -1  # Internal index, not shared outside (the value -1 is only used at creation/reset time)
    self.__servers = [
        'pool.ntp.org',
        'north-america.pool.ntp.org'
        'asia.pool.ntp.org',
        'europe.pool.ntp.org',
    ]
    self.__global_initial_t = self.__get_time_from_server()  # Real-time, wall-clock
    self.__local_initial_t = datetime.now(timezone.utc).timestamp()  # Corresponding local time
    self.__timestamps = []  # List to store timestamps for cycles
    self.__time2cycle_cache = 0  # Cached cycle value for optimization

Methods:

time2cycle

time2cycle(timestamp: float, delta: float | None = None) -> int

Convert a given timestamp to the corresponding cycle index.

Parameters:

Name	Type	Description	Default
timestamp	float	The timestamp to convert.	required
delta	float | None	The optional delta value for converting time to cycles.	None

Returns:

Name	Type	Description
int	int	The cycle index corresponding to the given timestamp.

Source code in unaiverse/clock.py

def time2cycle(self, timestamp: float, delta: float | None = None) -> int:
    """Convert a given timestamp to the corresponding cycle index.

    Args:
        timestamp (float): The timestamp to convert.
        delta (float | None): The optional delta value for converting time to cycles.

    Returns:
        int: The cycle index corresponding to the given timestamp.
    """
    if delta is not None and delta > 0:
        passed = self.get_time() - timestamp  # Precision: microseconds
        return self.cycle - int(passed * delta)
    else:
        self.__time2cycle_cache = Clock.__search(self.__timestamps, timestamp, self.__time2cycle_cache)
        return self.__time2cycle_cache

cycle2time

cycle2time(cycle: int, delta: float | None = None) -> float

Convert a cycle index to the corresponding timestamp.

Parameters:

Name	Type	Description	Default
cycle	int	The cycle index to convert.	required
delta	float | None	The optional delta value for converting cycles to time.	None

Returns:

Name	Type	Description
float	float	The timestamp corresponding to the given cycle index.

Source code in unaiverse/clock.py

def cycle2time(self, cycle: int, delta: float | None = None) -> float:
    """Convert a cycle index to the corresponding timestamp.

    Args:
        cycle (int): The cycle index to convert.
        delta (float | None): The optional delta value for converting cycles to time.

    Returns:
        float: The timestamp corresponding to the given cycle index.
    """
    if delta is not None and delta > 0:
        return cycle * delta
    else:
        return self.__timestamps[cycle] if cycle >= 0 else -1.

get_time

get_time(passed: bool = False) -> float

Get the current time based on the NTP server synchronization.

Returns:

Name	Type	Description
float	float	The current synchronized time (in seconds since the Unix epoch).

Source code in unaiverse/clock.py

def get_time(self, passed: bool = False) -> float:
    """Get the current time based on the NTP server synchronization.

    Returns:
        float: The current synchronized time (in seconds since the Unix epoch).
    """
    passed_since_beginning = datetime.now(timezone.utc).timestamp() - self.__local_initial_t
    return self.__global_initial_t + passed_since_beginning if not passed else passed_since_beginning

get_time_as_string

get_time_as_string() -> str

Get the current time as a string (ISO format).

Returns:

Name	Type	Description
str	str	A string representation of the current time (ISO format, UTC).

Source code in unaiverse/clock.py

def get_time_as_string(self) -> str:
    """Get the current time as a string (ISO format).

    Returns:
        str: A string representation of the current time (ISO format, UTC).
    """
    dt_object = datetime.fromtimestamp(self.get_time(), tz=timezone.utc)
    return dt_object.isoformat(timespec='milliseconds')

get_cycle

get_cycle()

Get the current cycle index.

Returns:

Name	Type	Description
int		The current cycle index.

Source code in unaiverse/clock.py

def get_cycle(self):
    """Get the current cycle index.

    Returns:
        int: The current cycle index.
    """
    return self.cycle

get_cycle_time

get_cycle_time()

Get the timestamp corresponding to the current cycle.

Returns:

Name	Type	Description
float		The timestamp corresponding to the current cycle index.

Source code in unaiverse/clock.py

def get_cycle_time(self):
    """Get the timestamp corresponding to the current cycle.

    Returns:
        float: The timestamp corresponding to the current cycle index.
    """
    return self.cycle2time(self.cycle)

next_cycle

next_cycle() -> bool

Move to the next cycle if the minimum delta time has passed or if cycles are not constrained.

Returns:

Name	Type	Description
bool	bool	True if the cycle was successfully moved to the next one, False otherwise.

Source code in unaiverse/clock.py

def next_cycle(self) -> bool:
    """Move to the next cycle if the minimum delta time has passed or if cycles are not constrained.

    Returns:
        bool: True if the cycle was successfully moved to the next one, False otherwise.
    """
    if self.cycle >= 0 and (self.min_delta > 0 and len(self.__timestamps) > 0 and
                            (self.get_time() - self.__timestamps[-1]) < self.min_delta):
        return False
    else:
        self.cycle += 1  # Increment the cycle index
        self.__add_timestamp(self.get_time())
        return True