API Reference

Complete API documentation for pychrony, auto-generated from source code docstrings.

Connection

The main entry point for interacting with chronyd.

ChronyConnection

Context manager for chrony connections.

Provides connection reuse for multiple queries to chronyd within a single context, properly managing socket and session lifecycle.

Parameters:

Name	Type	Description	Default
address	str | None	Connection address. Supports: Unix socket path: "/run/chrony/chronyd.sock" IPv4: "192.168.1.1" or "192.168.1.1:323" IPv6: "2001:db8::1" or "[2001:db8::1]:323" None: Auto-detect. The default Unix socket paths, then the localhost command port, are each tried by attempting the connection; the first that connects wins. address and transport report which one did.	None

Methods:

Name	Description
get_tracking	Get current NTP tracking status (returns TrackingStatus).
get_sources	Get configured time sources (returns list[Source]).
get_source_stats	Get source statistics (returns list[SourceStats]).
get_rtc_data	Get RTC tracking data (returns RTCData or None).

Choosing a transport

The Unix socket is a control channel; the command port serves only monitoring commands, which covers every report this class reads. A read-only consumer should prefer it and can assert on transport.

Thread Safety

NOT thread-safe. Each thread needs its own connection.

See Also

Transport: The transport a connection resolved to, and why it matters. TrackingStatus: Tracking data model. Source: Time source data model. SourceStats: Source statistics data model. RTCData: RTC tracking data model.

chronyc man page (access methods and the monitoring command set): https://chrony-project.org/doc/4.9/chronyc.html

Examples:

>>> with ChronyConnection() as conn:
...     tracking = conn.get_tracking()
...     sources = conn.get_sources()
...     stats = conn.get_source_stats()
...     rtc = conn.get_rtc_data()

Source code in src/pychrony/_core/_bindings.py

class ChronyConnection:
    """Context manager for chrony connections.

    Provides connection reuse for multiple queries to chronyd within a single
    context, properly managing socket and session lifecycle.

    Args:
        address: Connection address. Supports:

            - Unix socket path: ``"/run/chrony/chronyd.sock"``
            - IPv4: ``"192.168.1.1"`` or ``"192.168.1.1:323"``
            - IPv6: ``"2001:db8::1"`` or ``"[2001:db8::1]:323"``
            - ``None``: Auto-detect. The default Unix socket paths, then the
              localhost command port, are each tried by attempting the
              connection; the first that connects wins. `address` and
              `transport` report which one did.

    Methods:
        get_tracking: Get current NTP tracking status (returns `TrackingStatus`).
        get_sources: Get configured time sources (returns ``list[Source]``).
        get_source_stats: Get source statistics (returns ``list[SourceStats]``).
        get_rtc_data: Get RTC tracking data (returns `RTCData` or ``None``).

    Choosing a transport:
        The Unix socket is a control channel; the command port serves only
        monitoring commands, which covers every report this class reads. A
        read-only consumer should prefer it and can assert on `transport`.

    Thread Safety:
        NOT thread-safe. Each thread needs its own connection.

    See Also:
        `Transport`: The transport a connection resolved to, and why it matters.
        `TrackingStatus`: Tracking data model.
        `Source`: Time source data model.
        `SourceStats`: Source statistics data model.
        `RTCData`: RTC tracking data model.

        chronyc man page (access methods and the monitoring command set):
        https://chrony-project.org/doc/4.9/chronyc.html

    Examples:
        >>> with ChronyConnection() as conn:
        ...     tracking = conn.get_tracking()
        ...     sources = conn.get_sources()
        ...     stats = conn.get_source_stats()
        ...     rtc = conn.get_rtc_data()
    """

    def __init__(self, address: str | None = None) -> None:
        """Initialize ChronyConnection with optional address.

        Args:
            address: Connection address (see class docstring for formats)
        """
        self._address = address
        self._resolved_address: str | None = None
        self._denied_socket: str | None = None
        self._fd: int | None = None
        self._session: Any = None
        self._session_ptr: Any = None
        self._in_context = False

    def __enter__(self) -> "ChronyConnection":
        """Enter context manager, opening connection to chronyd."""
        _check_library_available()
        self._open()
        self._in_context = True
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Exit context manager, closing connection to chronyd."""
        self._in_context = False
        self._close()

    @property
    def address(self) -> str | None:
        """Address this connection is actually using, or None if not connected.

        For an explicit address this is that address; for auto-detect it is
        whichever candidate connected. Cleared when the connection closes.
        """
        return self._resolved_address

    @property
    def transport(self) -> Transport | None:
        """Transport in use, or None if not connected.

        `Transport` covers why the two differ in privilege. A caller that must
        not hold a control channel can assert on this.
        """
        if self._resolved_address is None:
            return None
        return (
            Transport.UNIX_SOCKET
            if _is_unix_socket_address(self._resolved_address)
            else Transport.COMMAND_PORT
        )

    def _candidate_addresses(self) -> list[str]:
        """Return the addresses to try, in order.

        An explicit address is used as given, with no fallback: asking for a
        specific transport should not silently land on a different one. Only
        auto-detect walks the candidate chain.
        """
        if self._address is not None:
            return [self._address]
        return [*DEFAULT_SOCKET_PATHS, *DEFAULT_COMMAND_PORTS]

    def _open(self) -> None:
        """Open a socket to the first reachable candidate and start a session.

        Candidates are decided by attempting the connection: a Unix socket can
        be present yet refuse to connect, since connecting needs write
        permission, and can be unstattable while chronyd is reachable there.

        Raises:
            ChronyConnectionError: If no candidate could be connected to
            ChronyPermissionError: If a Unix socket was rejected on permissions
        """
        candidates = self._candidate_addresses()
        last_error_code = -1
        denied_address: str | None = None

        for address in candidates:
            fd = _lib.chrony_open_socket(address.encode())
            if fd >= 0:
                self._fd = fd
                self._resolved_address = address
                break
            last_error_code = fd
            if denied_address is None and _is_permission_denied(address):
                denied_address = address
        else:
            self._raise_open_error(candidates, denied_address, last_error_code)

        # Kept for diagnostics: it explains a later unanswered request.
        self._denied_socket = denied_address

        # Initialize session
        self._session_ptr = _ffi.new("chrony_session **")
        err = _lib.chrony_init_session(self._session_ptr, self._fd)
        if err != 0:
            # Clean up socket on failure
            _lib.chrony_close_socket(self._fd)
            self._fd = None
            self._resolved_address = None
            raise ChronyConnectionError(
                "Failed to initialize chrony session",
                error_code=err,
            )

        self._session = self._session_ptr[0]

    def _raise_open_error(
        self,
        candidates: list[str],
        denied_address: str | None,
        error_code: int,
    ) -> None:
        """Raise the most specific error for a failed set of connect attempts.

        Args:
            candidates: Addresses that were tried, in order
            denied_address: First candidate rejected on permissions, if any
            error_code: Error code from the last failed attempt

        Raises:
            ChronyPermissionError: If a Unix socket was rejected on permissions
            ChronyConnectionError: Otherwise
        """
        if denied_address is not None:
            raise ChronyPermissionError(
                f"Permission denied connecting to {denied_address}. This is "
                "chronyd's control socket, accessible only to the root or "
                "chrony user; joining chrony's group does not help, because "
                "the socket is not group-writable. For read-only monitoring "
                'use the command port instead - ChronyConnection("127.0.0.1") '
                "- which serves every report this library reads. chronyd binds "
                'it to localhost by default; "cmdport 0" disables it. See '
                "https://chrony-project.org/doc/4.9/chronyc.html",
                error_code=error_code,
            )

        if self._address is not None:
            raise ChronyConnectionError(
                f"Failed to connect to {self._address}. Is chronyd running?",
                error_code=error_code,
            )

        raise ChronyConnectionError(
            "Failed to connect to chronyd (auto-detect). Tried: "
            f"{', '.join(candidates)}. Is chronyd running? If it is, its Unix "
            "socket may be unreachable for this user and its command port "
            'disabled ("cmdport 0").',
            error_code=error_code,
        )

    def _close(self) -> None:
        """Close session and socket connection."""
        if self._session is not None and self._session != _ffi.NULL:
            _lib.chrony_deinit_session(self._session)
            self._session = None

        if self._fd is not None and self._fd >= 0:
            _lib.chrony_close_socket(self._fd)
            self._fd = None

        self._session_ptr = None
        self._resolved_address = None
        self._denied_socket = None

    def _ensure_context(self) -> None:
        """Ensure we're within a context manager.

        Raises:
            RuntimeError: If called outside context manager
        """
        if not self._in_context:
            raise RuntimeError(
                "ChronyConnection methods must be called within a 'with' block"
            )

    def _raise_request_error(self, description: str, err: int) -> NoReturn:
        """Raise the error matching a failed request or response.

        A failed send()/recv() means chronyd could not be reached, so it is a
        connection problem: callers handling "chronyd unreachable" should not
        have to catch a data error for it as well. Every other chrony_err
        concerns the report itself.

        Args:
            description: What was attempted, e.g. "Failed to process tracking
                response"
            err: chrony_err value from the failed call

        Raises:
            ChronyConnectionError: If the send or receive itself failed
            ChronyDataError: For any other failure
        """
        if err not in TRANSPORT_ERRORS:
            raise ChronyDataError(description, error_code=err)

        message = f"{description}: chronyd did not respond on {self._resolved_address}."
        if self.transport is Transport.COMMAND_PORT:
            message += (
                " The command port is UDP, so the connection opens even when "
                'chronyd is not listening. It may be disabled ("cmdport 0"), '
                "bound to another address (bindcmdaddress), or blocked."
            )
            if self._denied_socket is not None:
                message += (
                    f" Auto-detect used it because {self._denied_socket} "
                    "refused the connection."
                )
        raise ChronyConnectionError(message, error_code=err)

    def _request_report(self, report_name: bytes) -> int:
        """Request number of records for a report type.

        Args:
            report_name: Report name (e.g., b"tracking", b"sources")

        Returns:
            Number of records available

        Raises:
            ChronyConnectionError: If chronyd could not be reached
            ChronyDataError: If the report itself could not be retrieved
        """
        err = _lib.chrony_request_report_number_records(self._session, report_name)
        if err != 0:
            self._raise_request_error(
                f"Failed to request {report_name.decode()} report", err
            )

        while _lib.chrony_needs_response(self._session):
            err = _lib.chrony_process_response(self._session)
            if err != 0:
                self._raise_request_error(
                    f"Failed to process {report_name.decode()} response", err
                )

        return _lib.chrony_get_report_number_records(self._session)

    def _request_record(self, report_name: bytes, index: int) -> None:
        """Request a specific record from a report.

        Args:
            report_name: Report name (e.g., b"tracking", b"sources")
            index: Record index

        Raises:
            ChronyConnectionError: If chronyd could not be reached
            ChronyDataError: If the record itself could not be retrieved
        """
        err = _lib.chrony_request_record(self._session, report_name, index)
        if err != 0:
            self._raise_request_error(
                f"Failed to request {report_name.decode()} record {index}", err
            )

        while _lib.chrony_needs_response(self._session):
            err = _lib.chrony_process_response(self._session)
            if err != 0:
                self._raise_request_error(
                    f"Failed to process {report_name.decode()} record {index}", err
                )

    def get_tracking(self) -> TrackingStatus:
        """Get current tracking status from chronyd.

        Returns:
            TrackingStatus: Current tracking information from chronyd.

        Raises:
            RuntimeError: If called outside context manager
            ChronyConnectionError: If chronyd did not respond.
            ChronyDataError: If tracking data is invalid or incomplete.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     status = conn.get_tracking()
            ...     print(f"Offset: {status.offset:.6f} seconds")
        """
        self._ensure_context()

        num_records = self._request_report(b"tracking")
        if num_records < 1:
            raise ChronyDataError("No tracking records available")

        self._request_record(b"tracking", 0)

        # Extract fields
        ref_id = _get_uinteger_field(self._session, "reference ID")
        leap_status_int = _get_uinteger_field(self._session, "leap status")

        try:
            leap_status = LeapStatus(leap_status_int)
        except ValueError:
            raise ChronyDataError(
                f"Unknown leap_status value {leap_status_int}. "
                "This may indicate a newer chrony version - please update pychrony."
            )

        data = {
            "reference_id": ref_id,
            "reference_id_name": _ref_id_to_name(ref_id),
            "reference_ip": _get_string_field(self._session, "address"),
            "stratum": _get_uinteger_field(self._session, "stratum"),
            "leap_status": leap_status,
            "ref_time": _get_timespec_field(self._session, "reference time"),
            "offset": _get_float_field(self._session, "current correction"),
            "last_offset": _get_float_field(self._session, "last offset"),
            "rms_offset": _get_float_field(self._session, "RMS offset"),
            "frequency": _get_float_field(self._session, "frequency offset"),
            "residual_freq": _get_float_field(self._session, "residual frequency"),
            "skew": _get_float_field(self._session, "skew"),
            "root_delay": _get_float_field(self._session, "root delay"),
            "root_dispersion": _get_float_field(self._session, "root dispersion"),
            "update_interval": _get_float_field(self._session, "last update interval"),
        }

        # Validate
        self._validate_tracking(data)

        return TrackingStatus(**data)

    def _validate_tracking(self, data: dict) -> None:
        """Validate tracking data before creating TrackingStatus."""
        if not 0 <= data["stratum"] <= 15:
            raise ChronyDataError(f"Invalid stratum: {data['stratum']}")

        float_fields = [
            "ref_time",
            "offset",
            "last_offset",
            "rms_offset",
            "frequency",
            "residual_freq",
            "skew",
            "root_delay",
            "root_dispersion",
            "update_interval",
        ]
        for field in float_fields:
            if math.isnan(data[field]) or math.isinf(data[field]):
                raise ChronyDataError(f"Invalid {field}: {data[field]}")

        non_negative = [
            "ref_time",
            "rms_offset",
            "skew",
            "root_delay",
            "root_dispersion",
            "update_interval",
        ]
        for field in non_negative:
            if data[field] < 0:
                raise ChronyDataError(f"{field} must be non-negative: {data[field]}")

    def get_sources(self) -> list[Source]:
        """Get all configured time sources from chronyd.

        Returns:
            list[Source]: List of Source objects for each configured source.
                Empty list if no sources are configured.

        Raises:
            RuntimeError: If called outside context manager
            ChronyConnectionError: If chronyd did not respond.
            ChronyDataError: If source data is invalid or incomplete.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     sources = conn.get_sources()
            ...     for src in sources:
            ...         print(f"{src.address}: stratum {src.stratum}")
        """
        self._ensure_context()

        num_records = self._request_report(b"sources")
        if num_records < 1:
            return []

        sources = []
        for i in range(num_records):
            self._request_record(b"sources", i)
            data = self._extract_source()
            self._validate_source(data)
            sources.append(Source(**data))

        return sources

    def _extract_source(self) -> dict:
        """Extract source fields from the current session record."""
        state_int = _get_uinteger_field(self._session, "state")
        mode_int = _get_uinteger_field(self._session, "mode")

        try:
            state = SourceState(state_int)
        except ValueError:
            raise ChronyDataError(
                f"Unknown state value {state_int}. "
                "This may indicate a newer chrony version - please update pychrony."
            )

        try:
            mode = SourceMode(mode_int)
        except ValueError:
            raise ChronyDataError(
                f"Unknown mode value {mode_int}. "
                "This may indicate a newer chrony version - please update pychrony."
            )

        # In libchrony 0.2, sources report uses TYPE_ADDRESS_OR_UINT32_IN_ADDRESS
        # which exposes either "address" (NTP sources) or "reference ID" (refclocks).
        # We check mode to determine which field to fetch.
        if mode == SourceMode.REFCLOCK:
            ref_id = _get_uinteger_field(self._session, "reference ID")
            address = _ref_id_to_name(ref_id)
        else:
            address = _get_string_field(self._session, "address")

        return {
            "address": address,
            "poll": _get_integer_field(self._session, "poll"),
            "stratum": _get_uinteger_field(self._session, "stratum"),
            "state": state,
            "mode": mode,
            "flags": _get_uinteger_field(self._session, "flags"),
            "reachability": _get_uinteger_field(self._session, "reachability"),
            "last_sample_ago": _get_uinteger_field(self._session, "last sample ago"),
            "orig_latest_meas": _get_float_field(
                self._session, "original last sample offset"
            ),
            "latest_meas": _get_float_field(
                self._session, "adjusted last sample offset"
            ),
            "latest_meas_err": _get_float_field(self._session, "last sample error"),
        }

    def _validate_source(self, data: dict) -> None:
        """Validate source data before creating Source."""
        _validate_bounded_int(data["stratum"], "stratum", 0, 15)
        _validate_bounded_int(data["reachability"], "reachability", 0, 255)
        _validate_non_negative_int(data["last_sample_ago"], "last_sample_ago")

        for field in ["orig_latest_meas", "latest_meas", "latest_meas_err"]:
            _validate_finite_float(data[field], field)

        if data["latest_meas_err"] < 0:
            raise ChronyDataError(
                f"latest_meas_err must be non-negative: {data['latest_meas_err']}"
            )

    def get_source_stats(self) -> list[SourceStats]:
        """Get statistical data for all time sources from chronyd.

        Returns:
            list[SourceStats]: List of SourceStats objects for each source.
                Empty list if no sources are configured.

        Raises:
            RuntimeError: If called outside context manager
            ChronyConnectionError: If chronyd did not respond.
            ChronyDataError: If statistics data is invalid or incomplete.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     stats = conn.get_source_stats()
            ...     for s in stats:
            ...         print(f"{s.address}: {s.samples} samples")
        """
        self._ensure_context()

        num_records = self._request_report(b"sourcestats")
        if num_records < 1:
            return []

        stats = []
        for i in range(num_records):
            self._request_record(b"sourcestats", i)
            data = self._extract_sourcestats()
            self._validate_sourcestats(data)
            stats.append(SourceStats(**data))

        return stats

    def _extract_sourcestats(self) -> dict:
        """Extract sourcestats fields from the current session record."""
        return {
            "reference_id": _get_uinteger_field(self._session, "reference ID"),
            "address": _get_string_field(self._session, "address"),
            "samples": _get_uinteger_field(self._session, "samples"),
            "runs": _get_uinteger_field(self._session, "runs"),
            "span": _get_uinteger_field(self._session, "span"),
            "std_dev": _get_float_field(self._session, "standard deviation"),
            "resid_freq": _get_float_field(self._session, "residual frequency"),
            "skew": _get_float_field(self._session, "skew"),
            "offset": _get_float_field(self._session, "offset"),
            "offset_err": _get_float_field(self._session, "offset error"),
        }

    def _validate_sourcestats(self, data: dict) -> None:
        """Validate sourcestats data before creating SourceStats."""
        _validate_non_negative_int(data["samples"], "samples")
        _validate_non_negative_int(data["runs"], "runs")
        _validate_non_negative_int(data["span"], "span")

        for field in ["std_dev", "resid_freq", "skew", "offset", "offset_err"]:
            _validate_finite_float(data[field], field)

        if data["std_dev"] < 0:
            raise ChronyDataError(f"std_dev must be non-negative: {data['std_dev']}")
        if data["skew"] < 0:
            raise ChronyDataError(f"skew must be non-negative: {data['skew']}")
        if data["offset_err"] < 0:
            raise ChronyDataError(
                f"offset_err must be non-negative: {data['offset_err']}"
            )

    def get_rtc_data(self) -> RTCData | None:
        """Get Real-Time Clock tracking data from chronyd.

        Returns:
            RTCData if RTC tracking is enabled, None otherwise.

        Raises:
            RuntimeError: If called outside context manager
            ChronyConnectionError: If chronyd did not respond.
            ChronyDataError: If RTC data is invalid or malformed.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     rtc = conn.get_rtc_data()
            ...     if rtc:
            ...         print(f"RTC offset: {rtc.offset:.6f}s")
        """
        self._ensure_context()

        num_records = self._request_report(b"rtcdata")
        if num_records < 1:
            return None

        # Try to fetch rtcdata record - may fail if RTC not actually configured
        try:
            err = _lib.chrony_request_record(self._session, b"rtcdata", 0)
            if err != 0:
                return None

            while _lib.chrony_needs_response(self._session):
                err = _lib.chrony_process_response(self._session)
                if err != 0:
                    return None
        # Deliberately broad: this is a C-library boundary, and an absent RTC is a
        # supported outcome reported as None rather than an error. Narrowing the
        # catch risks surfacing a CFFI failure as a crash for that normal case.
        except Exception:  # noqa: BLE001
            return None

        data = self._extract_rtc()
        self._validate_rtc(data)

        return RTCData(**data)

    def _extract_rtc(self) -> dict:
        """Extract RTC fields from the current session record."""
        return {
            "ref_time": _get_timespec_field(self._session, "reference time"),
            "samples": _get_uinteger_field(self._session, "samples"),
            "runs": _get_uinteger_field(self._session, "runs"),
            "span": _get_uinteger_field(self._session, "span"),
            "offset": _get_float_field(self._session, "offset"),
            "freq_offset": _get_float_field(self._session, "frequency offset"),
        }

    def _validate_rtc(self, data: dict) -> None:
        """Validate RTC data before creating RTCData."""
        _validate_non_negative_int(data["samples"], "samples")
        _validate_non_negative_int(data["runs"], "runs")
        _validate_non_negative_int(data["span"], "span")

        for field in ["ref_time", "offset", "freq_offset"]:
            _validate_finite_float(data[field], field)

        if data["ref_time"] < 0:
            raise ChronyDataError(f"ref_time must be non-negative: {data['ref_time']}")

address property

Address this connection is actually using, or None if not connected.

For an explicit address this is that address; for auto-detect it is whichever candidate connected. Cleared when the connection closes.

transport property

Transport in use, or None if not connected.

Transport covers why the two differ in privilege. A caller that must not hold a control channel can assert on this.

__init__(address=None)

Initialize ChronyConnection with optional address.

Parameters:

Name	Type	Description	Default
address	str | None	Connection address (see class docstring for formats)	None

Source code in src/pychrony/_core/_bindings.py

def __init__(self, address: str | None = None) -> None:
    """Initialize ChronyConnection with optional address.

    Args:
        address: Connection address (see class docstring for formats)
    """
    self._address = address
    self._resolved_address: str | None = None
    self._denied_socket: str | None = None
    self._fd: int | None = None
    self._session: Any = None
    self._session_ptr: Any = None
    self._in_context = False

__enter__()

Enter context manager, opening connection to chronyd.

Source code in src/pychrony/_core/_bindings.py

def __enter__(self) -> "ChronyConnection":
    """Enter context manager, opening connection to chronyd."""
    _check_library_available()
    self._open()
    self._in_context = True
    return self

__exit__(exc_type, exc_val, exc_tb)

Exit context manager, closing connection to chronyd.

Source code in src/pychrony/_core/_bindings.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit context manager, closing connection to chronyd."""
    self._in_context = False
    self._close()

get_tracking()

Get current tracking status from chronyd.

Returns:

Name	Type	Description
TrackingStatus	TrackingStatus	Current tracking information from chronyd.

Raises:

Type	Description
RuntimeError	If called outside context manager
ChronyConnectionError	If chronyd did not respond.
ChronyDataError	If tracking data is invalid or incomplete.

Examples:

>>> with ChronyConnection() as conn:
...     status = conn.get_tracking()
...     print(f"Offset: {status.offset:.6f} seconds")

Source code in src/pychrony/_core/_bindings.py

def get_tracking(self) -> TrackingStatus:
    """Get current tracking status from chronyd.

    Returns:
        TrackingStatus: Current tracking information from chronyd.

    Raises:
        RuntimeError: If called outside context manager
        ChronyConnectionError: If chronyd did not respond.
        ChronyDataError: If tracking data is invalid or incomplete.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     status = conn.get_tracking()
        ...     print(f"Offset: {status.offset:.6f} seconds")
    """
    self._ensure_context()

    num_records = self._request_report(b"tracking")
    if num_records < 1:
        raise ChronyDataError("No tracking records available")

    self._request_record(b"tracking", 0)

    # Extract fields
    ref_id = _get_uinteger_field(self._session, "reference ID")
    leap_status_int = _get_uinteger_field(self._session, "leap status")

    try:
        leap_status = LeapStatus(leap_status_int)
    except ValueError:
        raise ChronyDataError(
            f"Unknown leap_status value {leap_status_int}. "
            "This may indicate a newer chrony version - please update pychrony."
        )

    data = {
        "reference_id": ref_id,
        "reference_id_name": _ref_id_to_name(ref_id),
        "reference_ip": _get_string_field(self._session, "address"),
        "stratum": _get_uinteger_field(self._session, "stratum"),
        "leap_status": leap_status,
        "ref_time": _get_timespec_field(self._session, "reference time"),
        "offset": _get_float_field(self._session, "current correction"),
        "last_offset": _get_float_field(self._session, "last offset"),
        "rms_offset": _get_float_field(self._session, "RMS offset"),
        "frequency": _get_float_field(self._session, "frequency offset"),
        "residual_freq": _get_float_field(self._session, "residual frequency"),
        "skew": _get_float_field(self._session, "skew"),
        "root_delay": _get_float_field(self._session, "root delay"),
        "root_dispersion": _get_float_field(self._session, "root dispersion"),
        "update_interval": _get_float_field(self._session, "last update interval"),
    }

    # Validate
    self._validate_tracking(data)

    return TrackingStatus(**data)

get_sources()

Get all configured time sources from chronyd.

Returns:

Type	Description
list[Source]	list[Source]: List of Source objects for each configured source. Empty list if no sources are configured.

Raises:

Type	Description
RuntimeError	If called outside context manager
ChronyConnectionError	If chronyd did not respond.
ChronyDataError	If source data is invalid or incomplete.

Examples:

>>> with ChronyConnection() as conn:
...     sources = conn.get_sources()
...     for src in sources:
...         print(f"{src.address}: stratum {src.stratum}")

Source code in src/pychrony/_core/_bindings.py

def get_sources(self) -> list[Source]:
    """Get all configured time sources from chronyd.

    Returns:
        list[Source]: List of Source objects for each configured source.
            Empty list if no sources are configured.

    Raises:
        RuntimeError: If called outside context manager
        ChronyConnectionError: If chronyd did not respond.
        ChronyDataError: If source data is invalid or incomplete.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     sources = conn.get_sources()
        ...     for src in sources:
        ...         print(f"{src.address}: stratum {src.stratum}")
    """
    self._ensure_context()

    num_records = self._request_report(b"sources")
    if num_records < 1:
        return []

    sources = []
    for i in range(num_records):
        self._request_record(b"sources", i)
        data = self._extract_source()
        self._validate_source(data)
        sources.append(Source(**data))

    return sources

get_source_stats()

Get statistical data for all time sources from chronyd.

Returns:

Type	Description
list[SourceStats]	list[SourceStats]: List of SourceStats objects for each source. Empty list if no sources are configured.

Raises:

Type	Description
RuntimeError	If called outside context manager
ChronyConnectionError	If chronyd did not respond.
ChronyDataError	If statistics data is invalid or incomplete.

Examples:

>>> with ChronyConnection() as conn:
...     stats = conn.get_source_stats()
...     for s in stats:
...         print(f"{s.address}: {s.samples} samples")

Source code in src/pychrony/_core/_bindings.py

def get_source_stats(self) -> list[SourceStats]:
    """Get statistical data for all time sources from chronyd.

    Returns:
        list[SourceStats]: List of SourceStats objects for each source.
            Empty list if no sources are configured.

    Raises:
        RuntimeError: If called outside context manager
        ChronyConnectionError: If chronyd did not respond.
        ChronyDataError: If statistics data is invalid or incomplete.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     stats = conn.get_source_stats()
        ...     for s in stats:
        ...         print(f"{s.address}: {s.samples} samples")
    """
    self._ensure_context()

    num_records = self._request_report(b"sourcestats")
    if num_records < 1:
        return []

    stats = []
    for i in range(num_records):
        self._request_record(b"sourcestats", i)
        data = self._extract_sourcestats()
        self._validate_sourcestats(data)
        stats.append(SourceStats(**data))

    return stats

get_rtc_data()

Get Real-Time Clock tracking data from chronyd.

Returns:

Type	Description
RTCData | None	RTCData if RTC tracking is enabled, None otherwise.

Raises:

Type	Description
RuntimeError	If called outside context manager
ChronyConnectionError	If chronyd did not respond.
ChronyDataError	If RTC data is invalid or malformed.

Examples:

>>> with ChronyConnection() as conn:
...     rtc = conn.get_rtc_data()
...     if rtc:
...         print(f"RTC offset: {rtc.offset:.6f}s")

Source code in src/pychrony/_core/_bindings.py

def get_rtc_data(self) -> RTCData | None:
    """Get Real-Time Clock tracking data from chronyd.

    Returns:
        RTCData if RTC tracking is enabled, None otherwise.

    Raises:
        RuntimeError: If called outside context manager
        ChronyConnectionError: If chronyd did not respond.
        ChronyDataError: If RTC data is invalid or malformed.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     rtc = conn.get_rtc_data()
        ...     if rtc:
        ...         print(f"RTC offset: {rtc.offset:.6f}s")
    """
    self._ensure_context()

    num_records = self._request_report(b"rtcdata")
    if num_records < 1:
        return None

    # Try to fetch rtcdata record - may fail if RTC not actually configured
    try:
        err = _lib.chrony_request_record(self._session, b"rtcdata", 0)
        if err != 0:
            return None

        while _lib.chrony_needs_response(self._session):
            err = _lib.chrony_process_response(self._session)
            if err != 0:
                return None
    # Deliberately broad: this is a C-library boundary, and an absent RTC is a
    # supported outcome reported as None rather than an error. Narrowing the
    # catch risks surfacing a CFFI failure as a crash for that normal case.
    except Exception:  # noqa: BLE001
        return None

    data = self._extract_rtc()
    self._validate_rtc(data)

    return RTCData(**data)

Data Models

Frozen dataclasses representing chronyd report data.

TrackingStatus dataclass

Chrony tracking status information.

Represents the current time synchronization state from chronyd, including offset, frequency, and accuracy metrics.

Attributes:

Name	Type	Description
reference_id	int	NTP reference identifier (uint32 as hex IP or name).
reference_id_name	str	Human-readable reference source name.
reference_ip	str	IP address of reference source (IPv4, IPv6, or ID#).
stratum	int	NTP stratum level (0=reference clock, 1-15=downstream).
leap_status	LeapStatus	Leap second status (see LeapStatus).
ref_time	float	Timestamp of last measurement (seconds since epoch).
offset	float	Current offset from reference (seconds, can be negative).
last_offset	float	Offset at last measurement (seconds).
rms_offset	float	Root mean square of recent offsets (seconds).
frequency	float	Clock frequency error (parts per million).
residual_freq	float	Residual frequency for current source (ppm).
skew	float	Estimated error bound on frequency (ppm).
root_delay	float	Total roundtrip delay to stratum-1 source (seconds).
root_dispersion	float	Total dispersion to reference (seconds).
update_interval	float	Seconds since last successful update.

See Also

LeapStatus: Enum for leap second status values. ChronyConnection.get_tracking: Method to retrieve this data.

Source code in src/pychrony/models.py

@dataclass(frozen=True)
class TrackingStatus:
    """Chrony tracking status information.

    Represents the current time synchronization state from chronyd,
    including offset, frequency, and accuracy metrics.

    Attributes:
        reference_id: NTP reference identifier (uint32 as hex IP or name).
        reference_id_name: Human-readable reference source name.
        reference_ip: IP address of reference source (IPv4, IPv6, or ID#).
        stratum: NTP stratum level (0=reference clock, 1-15=downstream).
        leap_status: Leap second status (see `LeapStatus`).
        ref_time: Timestamp of last measurement (seconds since epoch).
        offset: Current offset from reference (seconds, can be negative).
        last_offset: Offset at last measurement (seconds).
        rms_offset: Root mean square of recent offsets (seconds).
        frequency: Clock frequency error (parts per million).
        residual_freq: Residual frequency for current source (ppm).
        skew: Estimated error bound on frequency (ppm).
        root_delay: Total roundtrip delay to stratum-1 source (seconds).
        root_dispersion: Total dispersion to reference (seconds).
        update_interval: Seconds since last successful update.

    See Also:
        `LeapStatus`: Enum for leap second status values.
        `ChronyConnection.get_tracking`: Method to retrieve this data.
    """

    reference_id: int
    reference_id_name: str
    reference_ip: str
    stratum: int
    leap_status: LeapStatus
    ref_time: float
    offset: float
    last_offset: float
    rms_offset: float
    frequency: float
    residual_freq: float
    skew: float
    root_delay: float
    root_dispersion: float
    update_interval: float

    def is_synchronized(self) -> bool:
        """Check if chronyd is synchronized to a source.

        Returns:
            True if synchronized (reference_id != 0 and stratum < 16),
            False otherwise.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     status = conn.get_tracking()
            ...     if status.is_synchronized():
            ...         print(f"Synced to {status.reference_ip}")
        """
        return self.reference_id != 0 and self.stratum < 16

    def is_leap_pending(self) -> bool:
        """Check if a leap second adjustment is pending.

        Returns:
            True if leap_status is INSERT or DELETE,
            False otherwise.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     status = conn.get_tracking()
            ...     if status.is_leap_pending():
            ...         print(f"Leap second pending: {status.leap_status.name}")
        """
        return self.leap_status in (LeapStatus.INSERT, LeapStatus.DELETE)

is_synchronized()

Check if chronyd is synchronized to a source.

Returns:

Type	Description
bool	True if synchronized (reference_id != 0 and stratum < 16),
bool	False otherwise.

Examples:

>>> with ChronyConnection() as conn:
...     status = conn.get_tracking()
...     if status.is_synchronized():
...         print(f"Synced to {status.reference_ip}")

Source code in src/pychrony/models.py

def is_synchronized(self) -> bool:
    """Check if chronyd is synchronized to a source.

    Returns:
        True if synchronized (reference_id != 0 and stratum < 16),
        False otherwise.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     status = conn.get_tracking()
        ...     if status.is_synchronized():
        ...         print(f"Synced to {status.reference_ip}")
    """
    return self.reference_id != 0 and self.stratum < 16

is_leap_pending()

Check if a leap second adjustment is pending.

Returns:

Type	Description
bool	True if leap_status is INSERT or DELETE,
bool	False otherwise.

Examples:

>>> with ChronyConnection() as conn:
...     status = conn.get_tracking()
...     if status.is_leap_pending():
...         print(f"Leap second pending: {status.leap_status.name}")

Source code in src/pychrony/models.py

def is_leap_pending(self) -> bool:
    """Check if a leap second adjustment is pending.

    Returns:
        True if leap_status is INSERT or DELETE,
        False otherwise.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     status = conn.get_tracking()
        ...     if status.is_leap_pending():
        ...         print(f"Leap second pending: {status.leap_status.name}")
    """
    return self.leap_status in (LeapStatus.INSERT, LeapStatus.DELETE)

Source dataclass

Chrony source information.

Represents an NTP server, peer, or reference clock being used as a time source by chronyd.

Attributes:

Name	Type	Description
address	str	IP address or reference ID of the source (IPv4, IPv6, or refclock ID).
poll	int	Polling interval as log2 seconds (e.g., 6 means 64 seconds).
stratum	int	NTP stratum level of the source (0-15).
state	SourceState	Selection state (see SourceState).
mode	SourceMode	Source mode (see SourceMode).
flags	int	Source flags (bitfield).
reachability	int	Reachability register (8-bit, 377 octal = all recent polls succeeded).
last_sample_ago	int	Seconds since last valid sample was received.
orig_latest_meas	float	Original last sample offset (seconds).
latest_meas	float	Adjusted last sample offset (seconds).
latest_meas_err	float	Last sample error bound (seconds).

See Also

SourceState: Enum for source selection states. SourceMode: Enum for source operational modes. ChronyConnection.get_sources: Method to retrieve source list.

Source code in src/pychrony/models.py

@dataclass(frozen=True)
class Source:
    """Chrony source information.

    Represents an NTP server, peer, or reference clock being used
    as a time source by chronyd.

    Attributes:
        address: IP address or reference ID of the source (IPv4, IPv6, or refclock ID).
        poll: Polling interval as log2 seconds (e.g., 6 means 64 seconds).
        stratum: NTP stratum level of the source (0-15).
        state: Selection state (see `SourceState`).
        mode: Source mode (see `SourceMode`).
        flags: Source flags (bitfield).
        reachability: Reachability register (8-bit, 377 octal = all recent polls succeeded).
        last_sample_ago: Seconds since last valid sample was received.
        orig_latest_meas: Original last sample offset (seconds).
        latest_meas: Adjusted last sample offset (seconds).
        latest_meas_err: Last sample error bound (seconds).

    See Also:
        `SourceState`: Enum for source selection states.
        `SourceMode`: Enum for source operational modes.
        `ChronyConnection.get_sources`: Method to retrieve source list.
    """

    address: str
    poll: int
    stratum: int
    state: SourceState
    mode: SourceMode
    flags: int
    reachability: int
    last_sample_ago: int
    orig_latest_meas: float
    latest_meas: float
    latest_meas_err: float

    def is_reachable(self) -> bool:
        """Check if the source has been reachable recently.

        Returns:
            True if reachability register is non-zero (at least one successful poll).

        Examples:
            >>> with ChronyConnection() as conn:
            ...     for src in conn.get_sources():
            ...         if not src.is_reachable():
            ...             print(f"Source {src.address} is unreachable")
        """
        return self.reachability > 0

    def is_selected(self) -> bool:
        """Check if this source is currently selected for synchronization.

        Returns:
            True if state is SELECTED.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     for src in conn.get_sources():
            ...         if src.is_selected():
            ...             print(f"Currently using {src.address}")
        """
        return self.state == SourceState.SELECTED

is_reachable()

Check if the source has been reachable recently.

Returns:

Type	Description
bool	True if reachability register is non-zero (at least one successful poll).

Examples:

>>> with ChronyConnection() as conn:
...     for src in conn.get_sources():
...         if not src.is_reachable():
...             print(f"Source {src.address} is unreachable")

Source code in src/pychrony/models.py

def is_reachable(self) -> bool:
    """Check if the source has been reachable recently.

    Returns:
        True if reachability register is non-zero (at least one successful poll).

    Examples:
        >>> with ChronyConnection() as conn:
        ...     for src in conn.get_sources():
        ...         if not src.is_reachable():
        ...             print(f"Source {src.address} is unreachable")
    """
    return self.reachability > 0

is_selected()

Check if this source is currently selected for synchronization.

Returns:

Type	Description
bool	True if state is SELECTED.

Examples:

>>> with ChronyConnection() as conn:
...     for src in conn.get_sources():
...         if src.is_selected():
...             print(f"Currently using {src.address}")

Source code in src/pychrony/models.py

def is_selected(self) -> bool:
    """Check if this source is currently selected for synchronization.

    Returns:
        True if state is SELECTED.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     for src in conn.get_sources():
        ...         if src.is_selected():
        ...             print(f"Currently using {src.address}")
    """
    return self.state == SourceState.SELECTED

SourceStats dataclass

Chrony source statistics.

Represents statistical data about measurements from an NTP source, used for drift and offset estimation.

Attributes:

Name	Type	Description
reference_id	int	32-bit NTP reference identifier.
address	str	IP address of the source (empty for reference clocks).
samples	int	Number of sample points currently retained.
runs	int	Number of runs of residuals with same sign.
span	int	Time interval between oldest and newest samples (seconds).
std_dev	float	Estimated sample standard deviation (seconds).
resid_freq	float	Residual frequency (parts per million).
skew	float	Frequency skew (error bound) in ppm.
offset	float	Estimated offset of the source (seconds).
offset_err	float	Offset error bound (seconds).

See Also

ChronyConnection.get_source_stats: Method to retrieve statistics.

Source code in src/pychrony/models.py

@dataclass(frozen=True)
class SourceStats:
    """Chrony source statistics.

    Represents statistical data about measurements from an NTP source,
    used for drift and offset estimation.

    Attributes:
        reference_id: 32-bit NTP reference identifier.
        address: IP address of the source (empty for reference clocks).
        samples: Number of sample points currently retained.
        runs: Number of runs of residuals with same sign.
        span: Time interval between oldest and newest samples (seconds).
        std_dev: Estimated sample standard deviation (seconds).
        resid_freq: Residual frequency (parts per million).
        skew: Frequency skew (error bound) in ppm.
        offset: Estimated offset of the source (seconds).
        offset_err: Offset error bound (seconds).

    See Also:
        `ChronyConnection.get_source_stats`: Method to retrieve statistics.
    """

    reference_id: int
    address: str
    samples: int
    runs: int
    span: int
    std_dev: float
    resid_freq: float
    skew: float
    offset: float
    offset_err: float

    def has_sufficient_samples(self, minimum: int = 4) -> bool:
        """Check if enough samples exist for reliable statistics.

        Args:
            minimum: Minimum number of samples required (default 4).

        Returns:
            True if samples >= minimum.

        Examples:
            >>> with ChronyConnection() as conn:
            ...     for stats in conn.get_source_stats():
            ...         if stats.has_sufficient_samples(8):
            ...             print(f"{stats.address}: offset={stats.offset:.6f}s")
        """
        return self.samples >= minimum

has_sufficient_samples(minimum=4)

Check if enough samples exist for reliable statistics.

Parameters:

Name	Type	Description	Default
minimum	int	Minimum number of samples required (default 4).	4

Returns:

Type	Description
bool	True if samples >= minimum.

Examples:

>>> with ChronyConnection() as conn:
...     for stats in conn.get_source_stats():
...         if stats.has_sufficient_samples(8):
...             print(f"{stats.address}: offset={stats.offset:.6f}s")

Source code in src/pychrony/models.py

def has_sufficient_samples(self, minimum: int = 4) -> bool:
    """Check if enough samples exist for reliable statistics.

    Args:
        minimum: Minimum number of samples required (default 4).

    Returns:
        True if samples >= minimum.

    Examples:
        >>> with ChronyConnection() as conn:
        ...     for stats in conn.get_source_stats():
        ...         if stats.has_sufficient_samples(8):
        ...             print(f"{stats.address}: offset={stats.offset:.6f}s")
    """
    return self.samples >= minimum

RTCData dataclass

Chrony RTC (Real-Time Clock) data.

Represents information about the hardware RTC and its relationship to system time, as tracked by chronyd.

Note: RTC tracking must be enabled in chronyd configuration. If not enabled, get_rtc_data() returns None.

Attributes:

Name	Type	Description
ref_time	float	RTC reading at last error measurement (seconds since epoch).
samples	int	Number of previous measurements used for calibration.
runs	int	Number of runs of residuals (indicates linear model fit quality).
span	int	Time period covered by measurements (seconds).
offset	float	Estimated RTC offset (fast by) in seconds.
freq_offset	float	RTC frequency offset (drift rate) in parts per million.

See Also

ChronyConnection.get_rtc_data: Method to retrieve RTC data.

Source code in src/pychrony/models.py

@dataclass(frozen=True)
class RTCData:
    """Chrony RTC (Real-Time Clock) data.

    Represents information about the hardware RTC and its relationship
    to system time, as tracked by chronyd.

    Note: RTC tracking must be enabled in chronyd configuration.
    If not enabled, `get_rtc_data()` returns ``None``.

    Attributes:
        ref_time: RTC reading at last error measurement (seconds since epoch).
        samples: Number of previous measurements used for calibration.
        runs: Number of runs of residuals (indicates linear model fit quality).
        span: Time period covered by measurements (seconds).
        offset: Estimated RTC offset (fast by) in seconds.
        freq_offset: RTC frequency offset (drift rate) in parts per million.

    See Also:
        `ChronyConnection.get_rtc_data`: Method to retrieve RTC data.
    """

    ref_time: float
    samples: int
    runs: int
    span: int
    offset: float
    freq_offset: float

    def is_calibrated(self) -> bool:
        """Check if RTC has enough calibration data.

        Returns:
            True if samples > 0 (some calibration exists).

        Examples:
            >>> with ChronyConnection() as conn:
            ...     rtc = conn.get_rtc_data()
            ...     if rtc and rtc.is_calibrated():
            ...         print(f"RTC drift: {rtc.freq_offset:.2f} ppm")
        """
        return self.samples > 0

is_calibrated()

Check if RTC has enough calibration data.

Returns:

Type	Description
bool	True if samples > 0 (some calibration exists).

Examples:

>>> with ChronyConnection() as conn:
...     rtc = conn.get_rtc_data()
...     if rtc and rtc.is_calibrated():
...         print(f"RTC drift: {rtc.freq_offset:.2f} ppm")

Source code in src/pychrony/models.py

def is_calibrated(self) -> bool:
    """Check if RTC has enough calibration data.

    Returns:
        True if samples > 0 (some calibration exists).

    Examples:
        >>> with ChronyConnection() as conn:
        ...     rtc = conn.get_rtc_data()
        ...     if rtc and rtc.is_calibrated():
        ...         print(f"RTC drift: {rtc.freq_offset:.2f} ppm")
    """
    return self.samples > 0

Enums

Categorical values for status fields.

LeapStatus

Bases: Enum

Leap second status for NTP synchronization.

Indicates whether time is normal or if a leap second adjustment is scheduled at the next midnight UTC.

Attributes:

Name	Type	Description
NORMAL		No leap second pending.
INSERT		Leap second will be inserted at midnight (23:59:60).
DELETE		Leap second will be deleted at midnight (skip 23:59:59).
UNSYNC		Clock is unsynchronized.

Examples:

>>> from pychrony import ChronyConnection, LeapStatus
>>> with ChronyConnection() as conn:
...     status = conn.get_tracking()
...     if status.leap_status == LeapStatus.INSERT:
...         print("Leap second insertion scheduled")
...     elif status.leap_status == LeapStatus.UNSYNC:
...         print("Clock not synchronized")

Source code in src/pychrony/models.py

class LeapStatus(Enum):
    """Leap second status for NTP synchronization.

    Indicates whether time is normal or if a leap second adjustment
    is scheduled at the next midnight UTC.

    Attributes:
        NORMAL: No leap second pending.
        INSERT: Leap second will be inserted at midnight (23:59:60).
        DELETE: Leap second will be deleted at midnight (skip 23:59:59).
        UNSYNC: Clock is unsynchronized.

    Examples:
        >>> from pychrony import ChronyConnection, LeapStatus
        >>> with ChronyConnection() as conn:
        ...     status = conn.get_tracking()
        ...     if status.leap_status == LeapStatus.INSERT:
        ...         print("Leap second insertion scheduled")
        ...     elif status.leap_status == LeapStatus.UNSYNC:
        ...         print("Clock not synchronized")
    """

    NORMAL = 0
    INSERT = 1
    DELETE = 2
    UNSYNC = 3

SourceState

Bases: Enum

Selection state of a chrony time source.

Indicates whether chrony has selected, rejected, or is considering this source for time synchronization.

Attributes:

Name	Type	Description
SELECTED		Currently selected for synchronization.
NONSELECTABLE		Cannot be selected (bad measurements).
FALSETICKER		Detected as providing incorrect time.
JITTERY		Measurements have excessive jitter.
UNSELECTED		Valid but not currently selected.
SELECTABLE		Candidate for selection.

Examples:

>>> from pychrony import ChronyConnection, SourceState
>>> with ChronyConnection() as conn:
...     for src in conn.get_sources():
...         if src.state == SourceState.FALSETICKER:
...             print(f"Warning: {src.address} detected as falseticker")
...         elif src.state == SourceState.SELECTED:
...             print(f"Active source: {src.address}")

Source code in src/pychrony/models.py

class SourceState(Enum):
    """Selection state of a chrony time source.

    Indicates whether chrony has selected, rejected, or is
    considering this source for time synchronization.

    Attributes:
        SELECTED: Currently selected for synchronization.
        NONSELECTABLE: Cannot be selected (bad measurements).
        FALSETICKER: Detected as providing incorrect time.
        JITTERY: Measurements have excessive jitter.
        UNSELECTED: Valid but not currently selected.
        SELECTABLE: Candidate for selection.

    Examples:
        >>> from pychrony import ChronyConnection, SourceState
        >>> with ChronyConnection() as conn:
        ...     for src in conn.get_sources():
        ...         if src.state == SourceState.FALSETICKER:
        ...             print(f"Warning: {src.address} detected as falseticker")
        ...         elif src.state == SourceState.SELECTED:
        ...             print(f"Active source: {src.address}")
    """

    SELECTED = 0
    NONSELECTABLE = 1
    FALSETICKER = 2
    JITTERY = 3
    UNSELECTED = 4
    SELECTABLE = 5

SourceMode

Bases: Enum

Operational mode of a chrony time source.

Distinguishes between NTP client connections, peer relationships, and local reference clocks.

Attributes:

Name	Type	Description
CLIENT		NTP client polling a server.
PEER		NTP peer relationship (bidirectional).
REFCLOCK		Local reference clock (GPS, PPS, etc.).

Examples:

>>> from pychrony import ChronyConnection, SourceMode
>>> with ChronyConnection() as conn:
...     for src in conn.get_sources():
...         if src.mode == SourceMode.REFCLOCK:
...             print(f"Reference clock: {src.address}")
...         elif src.mode == SourceMode.CLIENT:
...             print(f"NTP server: {src.address}")

Source code in src/pychrony/models.py

class SourceMode(Enum):
    """Operational mode of a chrony time source.

    Distinguishes between NTP client connections, peer
    relationships, and local reference clocks.

    Attributes:
        CLIENT: NTP client polling a server.
        PEER: NTP peer relationship (bidirectional).
        REFCLOCK: Local reference clock (GPS, PPS, etc.).

    Examples:
        >>> from pychrony import ChronyConnection, SourceMode
        >>> with ChronyConnection() as conn:
        ...     for src in conn.get_sources():
        ...         if src.mode == SourceMode.REFCLOCK:
        ...             print(f"Reference clock: {src.address}")
        ...         elif src.mode == SourceMode.CLIENT:
        ...             print(f"NTP server: {src.address}")
    """

    CLIENT = 0
    PEER = 1
    REFCLOCK = 2

Transport

Bases: Enum

Transport a ChronyConnection is using to reach chronyd.

The two do not carry the same privileges: chrony describes full access through the Unix socket as "more or less equivalent to being able to modify the chronyd's configuration file and restart it". Every report pychrony reads is in the monitoring set the command port serves, so a read-only consumer loses nothing by requiring COMMAND_PORT.

Attributes:

Name	Type	Description
UNIX_SOCKET		chronyd's Unix domain socket, its read-write control channel. Accessible locally by the root or chrony user.
COMMAND_PORT		chronyd's UDP command port (default 323, bound to 127.0.0.1 and ::1). Serves monitoring commands only.

See Also

https://chrony-project.org/doc/4.9/chronyc.html

Examples:

>>> from pychrony import ChronyConnection, Transport
>>> with ChronyConnection() as conn:
...     if conn.transport is not Transport.COMMAND_PORT:
...         raise RuntimeError("refusing to hold a control channel")
...     tracking = conn.get_tracking()

Source code in src/pychrony/models.py

class Transport(Enum):
    """Transport a `ChronyConnection` is using to reach chronyd.

    The two do not carry the same privileges: chrony describes full access
    through the Unix socket as "more or less equivalent to being able to modify
    the chronyd's configuration file and restart it". Every report pychrony
    reads is in the monitoring set the command port serves, so a read-only
    consumer loses nothing by requiring `COMMAND_PORT`.

    Attributes:
        UNIX_SOCKET: chronyd's Unix domain socket, its read-write control
            channel. Accessible locally by the root or chrony user.
        COMMAND_PORT: chronyd's UDP command port (default 323, bound to
            127.0.0.1 and ::1). Serves monitoring commands only.

    See Also:
        https://chrony-project.org/doc/4.9/chronyc.html

    Examples:
        >>> from pychrony import ChronyConnection, Transport
        >>> with ChronyConnection() as conn:
        ...     if conn.transport is not Transport.COMMAND_PORT:
        ...         raise RuntimeError("refusing to hold a control channel")
        ...     tracking = conn.get_tracking()
    """

    UNIX_SOCKET = "unix_socket"
    COMMAND_PORT = "command_port"

Exceptions

Exception hierarchy for error handling.

ChronyError

Bases: Exception

Base exception for all chrony-related errors.

Attributes:

Name	Type	Description
message		Human-readable error description
error_code		Optional numeric error code from libchrony

Source code in src/pychrony/exceptions.py

class ChronyError(Exception):
    """Base exception for all chrony-related errors.

    Attributes:
        message: Human-readable error description
        error_code: Optional numeric error code from libchrony
    """

    def __init__(self, message: str, error_code: int | None = None):
        super().__init__(message)
        self.message = message
        self.error_code = error_code

    def __str__(self) -> str:
        if self.error_code is not None:
            return f"{self.message} (error code: {self.error_code})"
        return self.message

ChronyConnectionError

Bases: ChronyError

Raised when chronyd cannot be reached.

Covers a connection that could not be opened and one that opened but was never answered. The second is specific to the command port: it is UDP, so the socket opens whether or not chronyd is listening.

Common causes:

• chronyd is not running
• No candidate address could be connected to during auto-detect
• The command port is disabled (cmdport 0), bound elsewhere (bindcmdaddress), or blocked by a firewall
• The client is not permitted by cmdallow (chronyd accepts monitoring commands only from localhost by default)
• chrony_init_session() returns error

Examples:

>>> from pychrony import ChronyConnection, ChronyConnectionError
>>> try:
...     with ChronyConnection() as conn:
...         status = conn.get_tracking()
... except ChronyConnectionError as e:
...     print(f"Connection failed: {e}")

Source code in src/pychrony/exceptions.py

class ChronyConnectionError(ChronyError):
    """Raised when chronyd cannot be reached.

    Covers a connection that could not be opened and one that opened but was
    never answered. The second is specific to the command port: it is UDP, so
    the socket opens whether or not chronyd is listening.

    Common causes:

    - chronyd is not running
    - No candidate address could be connected to during auto-detect
    - The command port is disabled (``cmdport 0``), bound elsewhere
      (``bindcmdaddress``), or blocked by a firewall
    - The client is not permitted by ``cmdallow`` (chronyd accepts monitoring
      commands only from localhost by default)
    - ``chrony_init_session()`` returns error

    Examples:
        >>> from pychrony import ChronyConnection, ChronyConnectionError
        >>> try:
        ...     with ChronyConnection() as conn:
        ...         status = conn.get_tracking()
        ... except ChronyConnectionError as e:
        ...     print(f"Connection failed: {e}")
    """

ChronyPermissionError

Bases: ChronyError

Raised when chronyd's Unix socket exists but refuses the connection.

Connecting to a Unix socket needs write permission, and chronyd creates the socket owned by the chrony user without group write, so joining chrony's group does not grant access - only root does. chrony also requires the socket's directory to be accessible only by the root or chrony user, so widening these permissions works against its design.

Prefer the command port for read-only monitoring: chronyd enables it on localhost by default and serves every report pychrony reads, while refusing the control commands the Unix socket would expose.

Common causes:

• Running as an unprivileged user against chronyd's Unix socket
• SELinux/AppArmor restrictions

See Also

https://chrony-project.org/doc/4.9/chronyc.html

Examples:

>>> from pychrony import ChronyConnection, ChronyPermissionError
>>> try:
...     with ChronyConnection() as conn:
...         status = conn.get_tracking()
... except ChronyPermissionError as e:
...     print(f"Permission denied: {e}")
...     # Read-only fallback; needs chronyd without "cmdport 0"
...     with ChronyConnection("127.0.0.1") as conn:
...         status = conn.get_tracking()

Source code in src/pychrony/exceptions.py

class ChronyPermissionError(ChronyError):
    """Raised when chronyd's Unix socket exists but refuses the connection.

    Connecting to a Unix socket needs write permission, and chronyd creates the
    socket owned by the chrony user without group write, so **joining chrony's
    group does not grant access** - only root does. chrony also requires the
    socket's directory to be accessible only by the root or chrony user, so
    widening these permissions works against its design.

    Prefer the command port for read-only monitoring: chronyd enables it on
    localhost by default and serves every report pychrony reads, while refusing
    the control commands the Unix socket would expose.

    Common causes:

    - Running as an unprivileged user against chronyd's Unix socket
    - SELinux/AppArmor restrictions

    See Also:
        https://chrony-project.org/doc/4.9/chronyc.html

    Examples:
        >>> from pychrony import ChronyConnection, ChronyPermissionError
        >>> try:
        ...     with ChronyConnection() as conn:
        ...         status = conn.get_tracking()
        ... except ChronyPermissionError as e:
        ...     print(f"Permission denied: {e}")
        ...     # Read-only fallback; needs chronyd without "cmdport 0"
        ...     with ChronyConnection("127.0.0.1") as conn:
        ...         status = conn.get_tracking()
    """

ChronyDataError

Bases: ChronyError

Raised when a report is retrieved but its data is invalid or incomplete.

Reserved for problems with the report itself. A failure to reach chronyd, including a request sent but never answered, raises ChronyConnectionError instead, so "chronyd is unreachable" is one exception to catch, not two.

Common causes:

• chrony_get_field_index() returns < 0 (field not found)
• chronyd rejected the report request (unknown report, unauthorized)
• Field validation fails (NaN, out of range)
• Protocol version mismatch

Examples:

>>> from pychrony import ChronyConnection, ChronyDataError
>>> with ChronyConnection() as conn:
...     try:
...         status = conn.get_tracking()
...     except ChronyDataError as e:
...         print(f"Invalid data: {e}")

Source code in src/pychrony/exceptions.py

class ChronyDataError(ChronyError):
    """Raised when a report is retrieved but its data is invalid or incomplete.

    Reserved for problems with the report itself. A failure to reach chronyd,
    including a request sent but never answered, raises `ChronyConnectionError`
    instead, so "chronyd is unreachable" is one exception to catch, not two.

    Common causes:

    - ``chrony_get_field_index()`` returns < 0 (field not found)
    - chronyd rejected the report request (unknown report, unauthorized)
    - Field validation fails (NaN, out of range)
    - Protocol version mismatch

    Examples:
        >>> from pychrony import ChronyConnection, ChronyDataError
        >>> with ChronyConnection() as conn:
        ...     try:
        ...         status = conn.get_tracking()
        ...     except ChronyDataError as e:
        ...         print(f"Invalid data: {e}")
    """

ChronyLibraryError

Bases: ChronyError

Raised when libchrony is not available.

Common causes:

• libchrony not installed at runtime
• CFFI bindings not compiled (missing libchrony-devel at build time)
• Library version incompatible

Examples:

>>> from pychrony import ChronyConnection, ChronyLibraryError
>>> try:
...     with ChronyConnection() as conn:
...         status = conn.get_tracking()
... except ChronyLibraryError as e:
...     print(f"Library not available: {e}")
...     print("Install libchrony-devel and rebuild")

Source code in src/pychrony/exceptions.py

class ChronyLibraryError(ChronyError):
    """Raised when libchrony is not available.

    Common causes:

    - libchrony not installed at runtime
    - CFFI bindings not compiled (missing libchrony-devel at build time)
    - Library version incompatible

    Examples:
        >>> from pychrony import ChronyConnection, ChronyLibraryError
        >>> try:
        ...     with ChronyConnection() as conn:
        ...         status = conn.get_tracking()
        ... except ChronyLibraryError as e:
        ...     print(f"Library not available: {e}")
        ...     print("Install libchrony-devel and rebuild")
    """

    def __init__(self, message: str):
        super().__init__(message, error_code=None)

Testing Utilities

Factory functions and pytest fixtures for testing code that uses pychrony.

testing

Testing utilities for pychrony.

This module provides factory functions and pytest fixtures for creating test instances of pychrony dataclasses with sensible defaults.

Factory Functions (for any test framework): from pychrony.testing import make_tracking, make_source status = make_tracking(stratum=3, offset=-0.001)

Pytest Fixtures (auto-discovered via plugin): def test_something(tracking_status, source): assert tracking_status.is_synchronized()

make_tracking(**overrides)

Create a TrackingStatus instance with sensible defaults.

Default state is synchronized (reference_id != 0, stratum=2).

Parameters:

Name	Type	Description	Default
**overrides	Any	Field values to override defaults	{}

Returns:

Type	Description
TrackingStatus	TrackingStatus instance

Examples:

>>> make_tracking()  # Synchronized status
>>> make_tracking(stratum=16, reference_id=0)  # Unsynchronized
>>> make_tracking(leap_status=LeapStatus.INSERT)  # Leap pending

Source code in src/pychrony/testing.py

def make_tracking(**overrides: Any) -> TrackingStatus:
    """Create a TrackingStatus instance with sensible defaults.

    Default state is synchronized (reference_id != 0, stratum=2).

    Args:
        **overrides: Field values to override defaults

    Returns:
        TrackingStatus instance

    Examples:
        >>> make_tracking()  # Synchronized status
        >>> make_tracking(stratum=16, reference_id=0)  # Unsynchronized
        >>> make_tracking(leap_status=LeapStatus.INSERT)  # Leap pending
    """
    return TrackingStatus(**{**TRACKING_DEFAULTS, **overrides})

make_source(**overrides)

Create a Source instance with sensible defaults.

Default state is selected and reachable.

Parameters:

Name	Type	Description	Default
**overrides	Any	Field values to override defaults.	{}

Returns:

Type	Description
Source	Source instance.

Examples:

>>> make_source()  # Selected, reachable source
>>> make_source(state=SourceState.FALSETICKER)  # Falseticker
>>> make_source(reachability=0)  # Unreachable source

Source code in src/pychrony/testing.py

def make_source(**overrides: Any) -> Source:
    """Create a Source instance with sensible defaults.

    Default state is selected and reachable.

    Args:
        **overrides: Field values to override defaults.

    Returns:
        Source instance.

    Examples:
        >>> make_source()  # Selected, reachable source
        >>> make_source(state=SourceState.FALSETICKER)  # Falseticker
        >>> make_source(reachability=0)  # Unreachable source
    """
    return Source(**{**SOURCE_DEFAULTS, **overrides})

make_source_stats(**overrides)

Create a SourceStats instance with sensible defaults.

Default has 8 samples (sufficient for statistics).

Parameters:

Name	Type	Description	Default
**overrides	Any	Field values to override defaults.	{}

Returns:

Type	Description
SourceStats	SourceStats instance.

Examples:

>>> make_source_stats()  # Stats with 8 samples
>>> make_source_stats(samples=2)  # Insufficient samples
>>> make_source_stats(offset=0.001)  # Custom offset

Source code in src/pychrony/testing.py

def make_source_stats(**overrides: Any) -> SourceStats:
    """Create a SourceStats instance with sensible defaults.

    Default has 8 samples (sufficient for statistics).

    Args:
        **overrides: Field values to override defaults.

    Returns:
        SourceStats instance.

    Examples:
        >>> make_source_stats()  # Stats with 8 samples
        >>> make_source_stats(samples=2)  # Insufficient samples
        >>> make_source_stats(offset=0.001)  # Custom offset
    """
    return SourceStats(**{**SOURCESTATS_DEFAULTS, **overrides})

make_rtc_data(**overrides)

Create an RTCData instance with sensible defaults.

Default is calibrated (samples > 0).

Parameters:

Name	Type	Description	Default
**overrides	Any	Field values to override defaults.	{}

Returns:

Type	Description
RTCData	RTCData instance.

Examples:

>>> make_rtc_data()  # Calibrated RTC
>>> make_rtc_data(samples=0)  # Uncalibrated RTC
>>> make_rtc_data(freq_offset=-5.0)  # Custom drift rate

Source code in src/pychrony/testing.py

def make_rtc_data(**overrides: Any) -> RTCData:
    """Create an RTCData instance with sensible defaults.

    Default is calibrated (samples > 0).

    Args:
        **overrides: Field values to override defaults.

    Returns:
        RTCData instance.

    Examples:
        >>> make_rtc_data()  # Calibrated RTC
        >>> make_rtc_data(samples=0)  # Uncalibrated RTC
        >>> make_rtc_data(freq_offset=-5.0)  # Custom drift rate
    """
    return RTCData(**{**RTCDATA_DEFAULTS, **overrides})

tracking_status()

Fixture providing a synchronized TrackingStatus with defaults.

Source code in src/pychrony/testing.py

@pytest.fixture
def tracking_status() -> TrackingStatus:
    """Fixture providing a synchronized TrackingStatus with defaults."""
    return make_tracking()

source()

Fixture providing a selected, reachable Source with defaults.

Source code in src/pychrony/testing.py

@pytest.fixture
def source() -> Source:
    """Fixture providing a selected, reachable Source with defaults."""
    return make_source()

source_stats()

Fixture providing a SourceStats with sufficient samples.

Source code in src/pychrony/testing.py

@pytest.fixture
def source_stats() -> SourceStats:
    """Fixture providing a SourceStats with sufficient samples."""
    return make_source_stats()

rtc_data()

Fixture providing a calibrated RTCData with defaults.

Source code in src/pychrony/testing.py

@pytest.fixture
def rtc_data() -> RTCData:
    """Fixture providing a calibrated RTCData with defaults."""
    return make_rtc_data()