Miscellaneous

Timestamps

class metricq.Timestamp(nanoseconds)

A MetricQ Timestamp

Parameters:

nanoseconds : int: number of nanoseconds elapsed since the UNIX epoch

__add__(delta)

Return the timestamp delta in the future (or in the past, if delta is negative), relative to a Timestamp instance.

>>> epoch = Timestamp(0)
>>> a_week_later = epoch + Timedelta.from_string("7 days")
>>> str(a_week_later)
'[604800000000000] 1970-01-08 01:00:00+01:00'

Parameters:

delta : Timedelta: a time duration, possibly negative

Return type:

Timestamp

Returns:

Timestamp

__eq__(other)

Check whether two Timestamps refer to the same instance of time:

>>> now = Timestamp.now()
>>> later = now + Timedelta.from_s(10)
>>> now == later
False

Parameters:

other : object: another timestamp

Return type:

bool

__lt__(other)

Compare whether this timestamp describes a time before another timestamp.

>>> now = Timestamp.now()
>>> later = now + Timedelta.from_s(10)
>>> now < later
True

Together with __eq__(), all relational operations (<=, >, !=, etc.) are supported. Timestamps are totally ordered (as supplied by functools.total_ordering()).

Parameters:

other : Timestamp: another timestamp

Return type:

bool

__str__()

Yield a human-readable date-time string in the local timezone:

>>> # in UTC+01:00, it was already 1 in the night when the UNIX epoch happened
:rtype: :py:class:`str`

>>> str(Timestamp(0))
'[0] 1970-01-01 01:00:00+01:00'

classmethod ago(delta)

Return a timestamp delta in the past.

This is equivalent to:

Timestamp.now() - delta

Parameters:

delta : Timedelta: the time difference to now

Return type:

Timestamp

property datetime: datetime

Create an aware UTC datetime object

All MetricQ timestamps are POSIX timestamps, hence UTC.

classmethod from_datetime(dt)

Create a Timestamp from an aware datetime object. If you have a naive datetime object, consider using from_local_datetime() instead.

Return type:: Timestamp

classmethod from_iso8601(iso_string)

Create a Timestamp from a ISO 8601 date-time string.

>>> Timestamp.from_iso8601("1970-01-01T00:00:00.0Z") == Timestamp(0)
True

This is a convenience method that parses the date-time string into a datetime.datetime using dateutil.parser.isoparse(), and then calls from_datetime() to create a Timestamp from that.

The provided iso_string must include timezone information. To parse local time strings, you must convert them yourself and use from_local_datetime(). Or better yet, somehow create an aware datetime.datetime and use from_datetime().

Note

The parser only supports up to 6 sub-second digits, further digits are simply dropped. If you need to parse timestamps with higher precision, you need to convert the string to nanoseconds yourself.

Parameters:

iso_string : str: a date-time string in ISO 8601 format including a timezone specifier.

Return type:

Timestamp

classmethod from_local_datetime(dt)

Create a timestamp from a naive datetime object that uses the local timezone. If you have a proper aware object, use from_datetime() instead.

This function uses dateutil.tz.gettz() which uses /etc/localtime. Using naive datetime object can be error-prone. Naive local times are ambiguous during daylight savings time adjustments. If you have to use this function, your workflow is probably broken.

Return type:: Timestamp

classmethod from_now(delta)

Return a timestamp delta in the future.

This is equivalent to:

Timestamp.now() + delta

Parameters:

delta : Timedelta: the time difference to now

Return type:

Timestamp

classmethod from_posix_seconds(seconds)

Create a Timestamp from a POSIX timestamp

Parameters:

seconds : float: number of seconds since the UNIX epoch

Return type:

Timestamp

Returns:

Timestamp

classmethod now()

Return a Timestamp corresponding to “now”

Return type:: Timestamp
Returns:: Timestamp

property posix: float: Number of seconds since the UNIX epoch

property posix_ms: float: Number of milliseconds since the UNIX epoch

property posix_ns: int: Number of nanoseconds since the UNIX epoch

property posix_us: float: Number of microseconds since the UNIX epoch

Timedeltas (Durations)

class metricq.Timedelta(nanoseconds)

A (possibly negative) duration of time

Parameters:

nanoseconds : int: duration in nanoseconds

Supported operations:

Operation	Types	Result
`d1 + d2`	`d1`, `d2`: `Timedelta`	the sum of the duration of `d1` and `d2` as a `Timedelta` object
`d + dt`	`d`: `Timedelta` `dt`: `datetime.timedelta`	the combined duration of `d` and `Timedelta.from_timedelta(dt)` as a `Timedelta` object
`d + t`	`d`: `Timedelta` `t`: `Timestamp`	a `Timestamp` offset by duration `d` (the same as `t + d`, see `Timestamp.__add__`)
`d1 == d2`	`d1`, `d2`: `Timedelta`	`True` if `d1` and `d2` describe the same duration of time
`d == dt`	`d`: `Timedelta` `dt`: `datetime.timedelta`	`True` if `d.datetime == dt` (see `datetime.timedelta`)
`d1 < d2`	`d1`, `d2`: `Timedelta`	`True` if `d1` is shorter than `d2`
`d < dt`	`d`: `Timedelta` `dt`: `datetime.timedelta`	`True` if `d.datetime < dt` (see `datetime.timedelta`)
`d * c`	`d`: `Timedelta` `c`: `float` or `int`	The duration scaled by the factor `c`, truncated to nanosecond precision See `__mul__()`.
`d / c`	`d`: `Timedelta` `c`: `float`	The duration divided by `c`, truncated to nanosecond precision. See `__truediv__()`.
`d // n`	`d`: `Timedelta` `n`: `int`	The duration divided by an integer factor `n`, truncated to nanosecond precision. See `__floordiv__()`.

In addition to < and =, all other relational operations are supported and behave as you would expect. The type is hashable.

__floordiv__(other)

Divide a duration by an integer or another Timedelta.

>>> td = Timedelta.from_s(10) // 3
>>> td.s
3.333333333
:rtype: :py:data:`~typing.Union`\[:py:class:`~metricq.timeseries.timedelta.Timedelta`, :py:class:`int`]

>>> td.precise_string
'3333333333ns'

This divides the number of nanoseconds in this duration by factor using Floor Division, i.e. for some n: int and td: Timedelta we have

>>> assert (td // n).ns == td.ns // n

Division by another Timedelta yields an integral result:

>>> assert Timedelta.from_s(10) // Timedelta.from_s(3) == 3

Note

Floor Division produces surprising results for float arguments and is therefore not supported here (i.e. 3 // 0.2 == 14.0). Use __truediv__() if you want to scale a duration by a non-integer factor.

__mul__(factor)

Scale a duration by a float factor.

Return type:: Timedelta

__str__()

A string containing the number of seconds of this duration

Return type:: str

>>> "One day has {} seconds".format(Timedelta.from_string("1 day"))
'One day has 86400.0s seconds'

__truediv__(other)

Divide a duration by a floating point or Timedelta.

Return type:: Union[Timedelta, float]

>>> (Timedelta.from_s(10) / 2.5).precise_string
'4s'

This divides the number of nanoseconds in this duration by factor using True Division, but truncates the result to an int:

>>> assert (td / x).ns == int(td.ns / x)

where td: Timedelta and x: float.

Division by another Timedelta instead yields a float result.

Note

Use __floordiv__() if you know that the divisor is an integer, this will prevent rounding errors.

staticmethod from_ms(milliseconds)

Create a duration from a number of milliseconds

Parameters:

milliseconds : float: number of milliseconds

Return type:

Timedelta

Returns:

A Timedelta object

staticmethod from_s(seconds)

Create a duration from a number of seconds

Parameters:

seconds : float: number of seconds

Return type:

Timedelta

Returns:

A Timedelta object

staticmethod from_string(duration_str)

Parse a human-readable string representation of a duration.

>>> "One day has {} seconds".format(Timedelta.from_string("1 day"))
'One day has 86400.0s seconds'

Parameters:

duration_str : str

A duration string in the form of “<number> <unit>”. Accepted units are:

Name

short

long

Seconds

"s"

"second", "seconds"

Milliseconds

"ms"

"millisecond", "milliseconds"

Microseconds

"us"

"microsecond", "microseconds"

Nanoseconds

"ns"

"nanosecond", "nanoseconds"

Minutes

"min"

"minute", "minutes"

Hours

"h"

"hour", "hours"

Days

"d"

"day", "days"

Return type:

Timedelta

Returns:

The parsed Timedelta object

Raises:

ValueError – if the string is not of the correct format or an invalid unit was specified

staticmethod from_timedelta(delta)

Convert from a standard timedelta object.

Parameters:

delta : timedelta: a standard datetime.timedelta object

Return type:

Timedelta

Returns:

A Timedelta object

staticmethod from_us(microseconds)

Create a duration from a number of microseconds

Parameters:

microseconds : float: number of microseconds

Return type:

Timedelta

Returns:

A Timedelta object

property ms: float: Number of milliseconds in this duration

property ns: int: Number of nanoseconds in this duration

property precise_string: str: An exact string representation of the duration preserving all nanosecond digits. If possible, a human-readable string is returned, e.g. 100ms. The result of this can be parsed by from_string() without losing precision.

property s: float: Number of seconds in this duration

property timedelta: timedelta: This duration as a standard datetime.timedelta object

property us: float: Number of microseconds in this duration

Time-value pairs and aggregates

class metricq.TimeValue(timestamp, value)

A timestamp-value pair.

Unpack it like so:

>>> tv = TimeValue(Timestamp.now(), 42.0)
>>> (timestamp, value) = tv

dict(): returns a dict representing the TimeValue instance. Keys are timestamp and value :rtype: Dict[str, Any]

Deprecated since version 5.0.0: Use the individual properties instead and select an appropriate timestamp type.

timestamp: Timestamp

value: float

class metricq.TimeAggregate(timestamp, minimum, maximum, sum, count, integral_ns, active_time)

Summary of a metric’s values within a certain period of time.

timestamp: Timestamp: starting time of this aggregate

minimum: float: minimum value

maximum: float: maximum value

sum: float: sum of all values

count: int: total number of values

integral_ns: float: Integral of values in this aggregate over its active time, nanoseconds-based

active_time: Timedelta: time spanned by this aggregate

staticmethod from_proto(timestamp, proto)

Return type:: TimeAggregate

staticmethod from_value(timestamp, value)

Return type:: TimeAggregate

staticmethod from_value_pair(timestamp_before, timestamp, value)

Create a TimeAggregate from a pair of timestamps (class:Timestamp) and one value

Raises:: NonMonotonicTimestamps – if the two timestamps are not strictly monotonic
Return type:: TimeAggregate

property integral_s: float: Integral of values in this aggregate over its active time, seconds-based

property mean: float

Mean value of this aggregate. This is the general way to access the mean value if nothing specific is known about the underlying raw data.

It may involve a heuristic to decide between mean_integral and mean_sum. Currently, it defaults to mean_integral.

property mean_integral: float

Mean value of this aggregate, calculated from the integral. Use this if you want to explicitly force this property.

In the HTA context, this should only be NaN if the aggregate interval is outside the of the interval from the earliest to the latest measurement point.

property mean_sum: float

Mean value of this aggregate, calculated from the sum. This can be useful when the underlying metric should be at regular intervals, but there are gaps in the data. Otherwise, mean_integral or just mean are more appropriate.

This value will be NaN if there are no raw data points in the aggregate interval.

dict(): returns a dict representing the TimeAggregate instance. Keys are timestamp, minimum, mean, maximum, and count. :rtype: Dict[str, Any]

Deprecated since version 5.0.0: Use the individual properties instead and chose between mean_integral and mean_sum based on your use-case

Helper types

metricq.Metric: alias of str

metricq.JsonDict: alias of dict[str, Any]

RPC handling

@metricq.rpc_handler(*function_tags)

A Decorator to mark an async method as an RPC handler

Parameters:

function_tags : str: The names of the RPCs that this method should handle

Return type:

Callable[[Callable[..., Optional[Any]]], Callable[..., Optional[Any]]]

Example

from metricq import Source, rpc_handler

class MySource(Source):

    ...

    @rpc_handler("config")
    async def on_config(self, **config):
        print(f"Received configuration: {config}")

    ...

Note

This only has an effect on methods of classes implementing MetricQ clients (see Client), i.e. Sink, Source, IntervalSource, HistoryClient, etc.