Services

Business logic for the API

server.app.services.api_services.check_ripe_measurement_scheduled(measurement_id)

Check if a RIPE Atlas measurement has been fully scheduled.

This function delegates to check_all_measurements_scheduled() to verify that all requested probes have been scheduled for the given RIPE measurement ID.

Parameters:

measurement_id (str) – The ID of the RIPE measurement to check.

Returns:

True if all requested probes are scheduled, False otherwise.

Return type:

bool

Raises:

ValueError – If the RIPE API returns an error or unexpected data.

server.app.services.api_services.fetch_historic_data_with_timestamps(server, start, end, session)

Fetches and reconstructs NTP measurements from the database within a specific time range.

Converts the provided human-readable datetime range into NTP-compatible timestamps, queries the database based on whether the server is an IP address or domain name, and reconstructs each result as an NtpMeasurement object.

Parameters:

• server (str) – An IPv4/IPv6 address or domain name string for which measurements should be fetched.

• start (datetime) – The start of the time range (in local or UTC timezone).

• end (datetime) – The end of the time range (in local or UTC timezone).

• session (Session) – The currently active database session.

Returns:

A list of NtpMeasurement objects representing the historical data for the given server within the time window.

Return type:

list[NtpMeasurement]

Notes

• The input datetimes are converted to UTC before processing.

• IP addresses are validated using the is_ip_address() utility function.

• Data is fetched using get_measurements_timestamps_ip or get_measurements_timestamps_dn depending on the server type.

• The PreciseTime wrapper is used to reconstruct accurate timestamps from database fields.

server.app.services.api_services.fetch_ripe_data(measurement_id)

Fetches and formats NTP measurement data from RIPE Atlas.

This function retrieves raw measurement data from the RIPE Atlas API using the given measurement ID, parses it into internal data structures, and formats it into a standardized dictionary format.

Parameters:

measurement_id (str) – The unique ID of the RIPE Atlas measurement to fetch.

Returns:

A list of dictionaries, each representing a formatted NTP measurement.

Return type:

list[dict]

server.app.services.api_services.get_format(measurement, jitter=None, nr_jitter_measurements=8)

Format an NTP measurement object into a dictionary suitable for JSON serialization.

Parameters:

• measurement (NtpMeasurement) – An object representing the NTP measurement result

• jitter (Optional[float]) – Optional jitter value if multiple measurements are performed

• nr_jitter_measurements (int) – The number of measurements used in the jitter calculation

Returns:

A dictionary containing key measurement details like this:

• Server info (ntp version, IP, name, reference IP, reference)

• Timestamps (client sent time, server receive time, server sent time, client receive time)

• Measurement metrics (offset, delay, stratum, precision, reachability)

• Extra details (root delay, last sync time, leap indicator)

Return type:

dict

server.app.services.api_services.get_ripe_format(measurement)

Converts a RipeMeasurement object into a standardized dictionary format.

This function extracts relevant information from the provided RipeMeasurement instance—including NTP server info, probe data, timing details, and measurement results—and formats it as a plain dictionary.

Parameters:

measurement (RipeMeasurement) – The parsed measurement object containing NTP and probe data

Returns:

A dictionary containing structured measurement data. Keys include: - NTP Server info (ntp version, ripe measurement id, IP, name, ref id) - Probe data (probe address, probe id in RIPE Atlas, probe location, time to result) - Measurement metrics (stratum, poll, precision, root delay, root dispersion, reachability) - NTP measurement data (rtt, offset, timestamps)

Return type:

dict[str, Any]

server.app.services.api_services.measure(server, wanted_ip_type, session, client_ip=None, measurement_no=8)

Performs an NTP measurement for a given server (IP or domain name) and stores the result in the database.

This function determines whether the input is an IP address or a domain name, then performs an NTP measurement using the appropriate method. The result is inserted into the database and returned.

Parameters:

• server (str) – A string representing either an IPv4/IPv6 address or a domain name.

• wanted_ip_type (int) – The IP type that we want to measure. Used for domain names.

• session (Session) – The currently active database session.

• client_ip (Optional[str]) – The client IP or None if it was not provided.

• measurement_no (int) – How many extra measurements to perform if the jitter_flag is True.

Returns:

• A list of pairs with a populated NtpMeasurement object if the measurement is successful, and the jitter.

• None if an exception occurs during the measurement process.

Return type:

list[tuple[NtpMeasurement, float, int]] | None

Raises:

DNSError – If the domain name is invalid, or it could not be resolved.

Notes

• If the server string is empty or improperly formatted, this may raise exceptions internally, which are caught and logged to stdout.

• This function modifies persistent state by inserting a measurement into the database.

server.app.services.api_services.override_desired_ip_type_if_input_is_ip(target_server, wanted_ip_type)

If the target server input is IP, then we want to perform its IP type measurements. Only for domain names, “wanted_ip_type” is considered. Otherwise, it is ignored.

Parameters:

• target_server (str) – The server we want to measure (domain name or IP address)

• wanted_ip_type (int) – The IP type the user said they wanted to measure.

Returns:

The IP type of the server in case the server input is IP, otherwise the wanted_ip_type unmodified.

Return type:

int

server.app.services.api_services.perform_ripe_measurement(ntp_server, client_ip, wanted_ip_type)

Initiate a RIPE Atlas measurement for a given server (IP address or domain name).

This function determines whether the provided server is an IP address or a domain name, and triggers the appropriate RIPE measurement.

Parameters:

• ntp_server (str) – The IP address or domain name of the target NTP server.

• client_ip (Optional[str]) – The IP address of the client requesting the measurement.

• wanted_ip_type (int) – The IP type that we want to measure. (4 or 6)

Returns:

The RIPE measurement ID. (as a string)

Return type:

str

Raises:

Exception – If the server string is invalid or the measurement failed.

Methods used for calculating data from the timestamps

class server.app.services.NtpCalculator.NtpCalculator

Bases: object

static calculate_float_time(time)

Converts a PreciseTime object to a float in seconds.

Parameters:

time (PreciseTime) – A single PreciseTime object, representing a single timestamp

Returns:

Time in seconds.

Return type:

float

static calculate_jitter(offsets)

Calculates the jitter of multiple NTP measurements based on their offsets.

Parameters:

offsets (list[float]) – A list of floats representing the offsets to calculate jitter.

Returns:

Jitter in seconds.

Return type:

float

static calculate_offset(timestamps)

Calculates the clock offset between client and server using NTP timestamps. Uses the formula ((t2 - t1) + (t3 - t4)) / 2

Parameters:

timestamps (NtpTimestamps) – A single NTP timestamps object, containing data about the 4 key timestamps

Returns:

Clock offset in seconds.

Return type:

float

static calculate_offset_from_dict(response_dict)

Calculates the clock offset between client and server using NTP timestamps from a dictionary. Uses the formula ((t2 - t1) + (t3 - t4)) / 2

Parameters:

response_dict (dict[str, Any]) – A dictionary with the timestamp values. (names are taken from RIPE results)

Returns:

Clock offset in seconds.

Return type:

float

static calculate_rtt(timestamps)

Calculates round-trip delay between client and server using NTP timestamps. It uses the formula (t4 - t1) - (t3 - t2)

Parameters:

timestamps (NtpTimestamps) – A single NTP timestamps object, containing data about the 4 key timestamps

Returns:

Delay in seconds

Return type:

float

Methods used for validating an NTP measurement

class server.app.services.NtpValidation.NtpValidation

Bases: object

static is_valid(details)

Checks the validity of the details object. According to ntp, the ‘leap’ attribute has only 2 bits and if its value is 3 (11 in binary) then it is invalid.

Parameters:

details (NtpExtraDetails) – The details objects to validate.

Returns:

True if the provided details have a ‘leap’ value different from 3,

False otherwise.

Return type:

bool