Skip to content

ntp_time

frog.hardware.plugins.time.ntp_time ¤

This module provides an interface for querying time from an NTP server.

Attributes¤

Classes¤

NTPTime(ntp_host=TIME_NTP_HOST, ntp_version=TIME_NTP_VERSION, ntp_port=TIME_NTP_PORT, ntp_timeout=TIME_NTP_TIMEOUT, ntp_poll_interval=TIME_NTP_POLL_INTERVAL) ¤

Bases: TimeBase

A time source that queries an NTP server.

Create a new NTPTime.

Parameters:

Name Type Description Default
ntp_host str

The IP address or hostname of the NTP server

 TIME_NTP_HOST
ntp_version int

The NTP version to use

 TIME_NTP_VERSION
ntp_port int

The port to connect to

 TIME_NTP_PORT
ntp_timeout float

The maximum time to wait for a response

 TIME_NTP_TIMEOUT
ntp_poll_interval float

How often to query the NTP server (seconds)

 TIME_NTP_POLL_INTERVAL
Source code in frog/hardware/plugins/time/ntp_time.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    ntp_host: str = TIME_NTP_HOST,
    ntp_version: int = TIME_NTP_VERSION,
    ntp_port: int = TIME_NTP_PORT,
    ntp_timeout: float = TIME_NTP_TIMEOUT,
    ntp_poll_interval: float = TIME_NTP_POLL_INTERVAL,
) -> None:
    """Create a new NTPTime.

    Args:
        ntp_host: The IP address or hostname of the NTP server
        ntp_version: The NTP version to use
        ntp_port: The port to connect to
        ntp_timeout: The maximum time to wait for a response
        ntp_poll_interval: How often to query the NTP server (seconds)
    """
    super().__init__()
    self._client = NTPClient()
    self._ntp_host = ntp_host
    self._ntp_version = ntp_version
    self._ntp_port = ntp_port
    self._ntp_timeout = ntp_timeout

    # Set up time offset polling.
    self.poll_time_offset()
    self._poll_timer = QTimer()
    self._poll_timer.timeout.connect(self.poll_time_offset)
    self._poll_timer.start(int(ntp_poll_interval * 1000))
Functions¤
close() ¤

Close the device.

Source code in frog/hardware/plugins/time/ntp_time.py 
87
88
89
90
def close(self) -> None:
    """Close the device."""
    self._poll_timer.stop()
    super().close()
get_time_offset() ¤

Get the current time offset in seconds.

Returns:

Type Description
float

A float representing the current time offset.
Source code in frog/hardware/plugins/time/ntp_time.py 
79
80
81
82
83
84
85
def get_time_offset(self) -> float:
    """Get the current time offset in seconds.

    Returns:
        A float representing the current time offset.
    """
    return self._response.offset
poll_time_offset() ¤

Query the NTP server for the current time offset.

Source code in frog/hardware/plugins/time/ntp_time.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
def poll_time_offset(self) -> None:
    """Query the NTP server for the current time offset."""
    logging.info("Polling NTP server for time offset.")
    try:
        self._response = self._client.request(
            self._ntp_host,
            version=self._ntp_version,
            port=self._ntp_port,
            timeout=self._ntp_timeout,
        )
        logging.info(f"NTP time offset: {self._response.offset}")
    except Exception as e:
        raise NTPTimeError(f"Error querying NTP server: {e}")

NTPTimeError ¤

Bases: Exception

Indicates that an error occurred while querying the NTP time server.