Timeseries Documentation

Timeseries Documentation#

For general users Stream and StreamSet should be the major classes of interest. Avoid using the timeseries client and instead use pingthings.Client as the main interaction point with the system.

Stream#

class pingthings.timeseries.Stream#

Bases: object

aligned_windows_iter(start, end, point_width, version=None, schema=None)#

Return statistical summary timeseries data aligned with the internal tree structure of the database.

Query BTrDB for aggregates (or roll ups or windows) of the time series with version between time start (inclusive) and end (exclusive) in nanoseconds [start, end). Each point returned is a statistical aggregate of all the raw data within a window of width 2**pointwidth nanoseconds. These statistical aggregates currently include the mean, minimum, maximum, count, and standard deviation of the data composing the window.

Note

Understanding aligned_windows queries

start is inclusive, but end is exclusive.
Results will be returned for all windows that start in the interval \([start, end)\).
If \(end < start + 2^{pointwidth}\), you will not get any results.
If start and end are not powers of two, the bottom pointwidth bits will be cleared, aligning them to the nearest multiple of \(2^{pointwidth}\). For example, if you query data between [31, 121) with pointwidth = 4, the actual query will be performed on [16, 112].
Each window will contain statistical summaries of the window.
Statistical points with count == 0 will be omitted.

Tip

Memory efficient queries

For long time ranges with small window widths (many statistical summaries), using the _iter methods is significantly more memory efficient. These methods return data in batches, which are guaranteed to be returned in sorted order, allowing you to query much larger time ranges.

Parameters:

start – Start time to get data (inclusive)
end – End time for the data query (exclusive)
point_width – What size statistical windows to return, aligned to a size of the internal tree. Value will be interpreted as \(2^{point\_width}\).
width – The size of statistical windows to return, will contain summaries of the data between start to end of size width, this is mutually exclusive with point_width.
version – What version of the stream to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,statistical summary timeseries be returned as, default will be the server default.

Yields:

Statistical summary timeseries of the stream between \([start, end)\)

annotations(copy=True, refresh=None)#

Return the metadata that is not present in the tag-metadata of the stream.

Parameters:

copy – Should the returned dict be copied or passed by reference, by default True
refresh – Do we need to get the latest metadata from the platform first, by default False

Returns:

A mapping of annotation metadata key,value pairs.

changes(from_version, to_version, resolution)#

Return a list of intervals of time that have changed between two versions.

Unlike the _iter version, this method buffers and returns the changed intervals in a single list.

Warning

Advanced user feature

This method can have unexpected return values if you are not familiar with the underlying timeseries database that is part of the PredictiveGrid platform.

Note

Time ranges can overlap with previously inserted data

The returned timeranges will not be exact start and end times for the data inserted. Instead, they will be aligned as closely as they can be, but also in aligment with the underlying database’s tree structure.

Parameters:

from_version – The stream version to compare from (exclusive).
to_version – The stream version to compare to (inclusive).
resolution – The “coarseness” of change detection in tree width. A resolution of 0 typically returns the finest granularity of changed intervals (which is still aligned with our internal tree structure), a resolution of 30, would be aligned exactly as a pointwidth of 2^30 nanoseconds. This resolution is the same as a pointwidth, please refer to the explanation page here on stat point widths of our tree.

Returns:

A list of dictionaries representing changed intervals. Typically each dictionary looks like: {"start": <nanosecond_timestamp>, "end": <nanosecond_timestamp>}.

Return type:

list[dict[str, int] | None]

Examples

>>> # Suppose we want to iterate over all changes between version 9 and 42
>>> for changed_range in stream.changes(from_version=9, to_version=42, resolution=0):
...     print("Changed interval:", changed_range)

>>> # Suppose we want to iterate over all changes between version 9 and 42 and get the lastest data that has not been inserted yet
>>> for changed_range in stream.changes(from_version=9, to_version=0, resolution=0):
...     print("Changed interval:", changed_range)

changes_iter(from_version, to_version, resolution)#

Return the intervals of time that have changed between two versions as a generator.

Tip

Memory efficient queries

This method yields each changed interval one at a time. If you need to process many intervals without storing them all in memory at once, this _iter method is more memory efficient than its non-iter counterpart.

Warning

Advanced user feature

This method can have unexpected return values if you are not familiar with the underlying timeseries database that is part of the PredictiveGrid platform.

Note

Time ranges can overlap with previously inserted data

The returned timeranges will not be exact start and end times for the data inserted. Instead, they will be aligned as closely as they can be, but also in aligment with the underlying database.

Parameters:

from_version – The stream version to compare from.
to_version – The stream version to compare to.
resolution – The “coarseness” of change detection in tree width. A resolution of 0 typically returns the finest granularity of changed intervals (which is still aligned with our internal tree structure), a resolution of 30, would be aligned exactly as a pointwidth of 2^30 nanoseconds. This resolution is the same as a pointwidth, please refer to the explanation page here on stat point widths of our tree.

Yields:

dict[str, int] – A dictionary representing a changed interval, with a structure like: {"start": <nanosecond_timestamp>, "end": <nanosecond_timestamp>}.

Examples

>>> # Suppose we want to iterate over all changes between version 9 and 42
>>> for changed_range in stream.changes_iter(from_version=9, to_version=42, resolution=0):
...     print("Changed interval:", changed_range)

>>> # Suppose we want to iterate over all changes between version 9 and 42 and get the lastest data that has not been inserted yet
>>> for changed_range in stream.changes_iter(from_version=9, to_version=0, resolution=0):
...     print("Changed interval:", changed_range)

collection#

count(start=None, end=None, version=None, precise=None)#

Get the total count of raw measurements that are present in the stream.

Parameters:

start – Bound the lower end of this query by a start time, by default MINIMUM_TIME.
end – Bound the upper end of this query by an end time, by default MAXIMUM_TIME
version – Version of the stream to query against, by default None, which means a version of 0 is used.
precise – Do we need an exact count or is an estimate reasonable, by default False

Returns:

Count of points in the stream.

delete(start, end)#

“Delete” all points between [start, end)

“Delete” all points between start (inclusive) and end (exclusive), both in nanoseconds.

Note

This is a soft delete

The PingThings timeseries data structure has persistent multiversioning. This means that the deleted points will still exist as part of an older version of the stream.

Parameters:

start – Time to begin deleting points in the stream (inclusive).
end – Time to stop deleting points in the stream (exclusive).

Returns:

The updated version number of the stream.

earliest(version=None)#

Find the earliest point (in time) that is present in the stream at version.

Parameters:: version – The version of the stream to query against, by default None, which means a version of 0 will be used.
Returns:: The earliest point in the stream at version.

flush()#

Force a flush of the buffered data to persistent storage.

If data was present, the version number will be positively incremented.

Returns:: The major version of the stream after the flush.

get_latest_version()#

Get the current version of the stream.

Returns:: The version number of the stream.

get_retention(copy=True, refresh=None)#

Get the retention policy of the stream.

Parameters:

copy – Should the returned dict be copied or passed by reference, by default True
refresh – Do we need to get the latest metadata from the platform first, by default True

Returns:

Retention policy.

Examples

>>> stream_a.get_retention()
{}
>>> stream_b.get_retention()
{'remove_older_than': datetime.timedelta(hours=2)}

get_version_at_time(pytime)#

Return the version of the stream at the time provided.

Parameters:: realtime – The time to check the version of the stream at, in nanoseconds.
Returns:: The version of the stream at time realtime.

insert(data, merge_policy=None, round_spec=None)#

Add new timeseries data to the stream.

Tip

Default merge policy has changed!

Starting with the new pingthings api, the default merge policy is now replace. Please refer to the merge policy docs

Tip

Data must follow a specific schema for insertion

Your data pyarrow table or record batch must have a schema that matches the TIME_VALUE_F64_SCHEMA, defined. Order of the columns matter.

Parameters:

data – A pyarrow table or record batch of time series data.
merge_policy – How should the database handle data with the same timestamp, by default “replace”
round_spec – Set how many bits of precision to keep inserted data. Default behavior is the full 64 bits.

Returns:

The version of the stream after data insertion, this number can increment multiple times depending on how many points are inserted.

latest(version=None)#

Find the latest point (in time) that is present in the stream at version.

Parameters:: version – The version of the stream to query against, by default None, which means a version of 0 will be used.
Returns:: The latest point in the stream, if present.

name#

nearest(time, backward=None, version=None)#

Get the nearest datapoint at time with version.

Parameters:

time – The time to query for the nearest point.
backward – Can the query find the closest time that is before time, by default False
version – Version of the stream to query against, by default None, which means version 0 will be used.

Returns:

The time,value Point that is closest to the time in question, if available.

obliterate()#: Completely remove the stream from the platform.

Danger

Will delete data!

Obliterating the stream will remove the stream, as well as all of its data! This stream will no longer be accessible, make sure you are completely sure you want to do this. If you have Admin privileges, you can obliterate streams!

pin_version(version)#

Set the version of the stream to a specific version number, useful when wanting reproducible data queries.

Default behavior is to pin the stream to the latest version number when this method is executed.

Tip

Useful version number

If you do not want to pin the stream to a version and want to instead always use the latest version of the stream, which includes data that is being streamed into the platform, use a version number of 0. This is a “magic” value which tells the platform to always use the latest data it can find.

Parameters:: version – Version number to pin the stream to, by default None, which will pin the stream to its latest version, as returned by Stream.get_latest_version.
Returns:: The version pinned.

pinned_version#

precise_windows_iter(start, end, width, depth=0, version=None, schema=None)#

Return custom-sized statistical summaries of the data.

Note

Understanding windows queries

windows returns arbitrary precision statistical summary windows from the platform.
It is slower than aligned_windows, but can be significantly faster than raw value queries (raw_values).
Each returned window will be width nanoseconds long.
start is inclusive, but end is exclusive (e.g., if end < start + width you will get no results).
Results will be returned for all windows that start at a time less than the end timestamp.
If (end - start) is not a multiple of width, then end will be decreased to the greatest value less than end such that (end - start) is a multiple of width (i.e., we set end = start + width * floordiv(end - start, width)).
Windows that have no data points count==0 will be omitted from the returned table

Tip

Memory efficient queries

Parameters:

start – Start time to get data (inclusive)
end – End time for the data query (exclusive)
width – The size of statistical windows to return, will contain summaries of the data between start to end of size width.
depth – What is the maximum tradeoff in computation of the statistical summaries by how far we need to walk down the tree, by default 0 is the most accurate, and a range of 0->63 can be used based on the time range of data.
version – What version of the stream to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,statistical summary timeseries be returned as, default will be the server default.

Yields:

The statistical summary information of the stream as a timeseries.

raw_values(start, end, version=None, schema=None)#

Return the raw time,value pairs of data from the stream between start and end.

Parameters:

start – Start time to get data (inclusive)
end – End time for the data query (exclusive)
version – What version of the stream to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,value pairs be returned as, default will be the server default.

Returns:

A table of timeseries data in the interval of [start, end)

raw_values_iter(start, end, version=None, schema=None)#

Return the raw timeseries data from the stream between start and end, but as an iterator instead of buffering all at once.

Tip

Memory efficient queries

If you are working with a lot of raw data and can afford to do processing in batches (which are guaranteed to return in sorted order), the _iter based methods are much more memory efficient and allow you to query much larger ranges of time.

Parameters:

start – Start time to get data (inclusive)
end – End time for the data query (exclusive)
version – What version of the stream to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,value pairs be returned as, default will be the server default.

Yields:

Tables of timeseries data from the stream in the interval of [start, end)

Examples

Query for a large range of data and process in batches.

>>> value_generator = stream.raw_values_iter(start, end)
>>> value_counter = 0
>>> for batch in value_generator:
>>>     value_counter += batch.num_rows
>>> print(f"Processed {value_counter} rows")

refresh_metadata()#

Retrieve and update all metadata for this stream.

This will update the tags and annotations for the stream as well as any other metadata that might not be set during manual instantiation of the Stream object

set_retention(remove_older_than=None)#

Set retention policy on a stream.

Parameters:: remove_older_than – Trim time period after which the data can be removed. Specifying None or not passing the parameter disables trimming.

Examples

Keep the data for only two hours.

>>> stream.set_retention(datetime.timedelta(hours=2))

tags(copy=True, refresh=None)#

Return the tag-based metadata of the stream.

Parameters:

copy – Should the tag dictionary be copied or passed by reference, by default True
refresh – Do we need to query the platform to get the metadata first, by default False

Returns:

A mapping of tag metadata key,value pairs.

update(collection=None, tags=None, annotations=None, replace_tags=None, replace_annotations=None)#

Update the stream metadata. tags must be from one of the below fields.

Parameters:

collection – Change the collection the stream is located under, by default None
tags – Update any tag metadata of the stream, by default None
annotations – Update any non-tag metadata of the stream, by default None
replace_tags – If you want to fully replace the current stream tags with the ones provided here, set to True, by default False
replace_annotations – If you want to fully replace the current stream annotations metadata with the ones provided here set to True by default False.

Note

Update Behavior

This function updates a field if it has been set, otherwise it inserts a new value. Not passing a value for an existing field will not clear the value from that field. For that you must choose replace=True. If replace=True name can not be None or an empty string.

Tip

Available tags in the stream table

column_name	Description
name	name of the stream
source	Where did the stream come from
device_id	What device is the stream attached to
samples_per_second	What is the nominal report rate of the data, in Hz?
time_precision	How accurate is each timestep (+- nanoseconds)
value_precision	Bit precision
dynamic	Can the timeseries stream send us data at varying report rates?
continuous	Should we expect a continuous stream of data?
description	Additional information about the timeseries stream
alt_name	Additional name that the stream may be referenced by
alt_id	Alternative ID (most likely SignalID from STTP)
norm_factor	A factor to scale the measurement by to normalize. Like “baseKV” for voltage, etc.
scaling_factor	The “m” in y=mx + b factor, line-line scaling factor, etc
bias_factor	The “b” in y=mx+b
hidden	Whether the stream should be presented to non-admin users
unit	Unit of measurement of the stream data
unit_id	unit_id for the corresponding unit in the unit table

uuid#

windowed_values(start, end, width, precise=None, version=None, schema=None, depth=0)#

Return statistical summaries of the data.

Parameters:

start – The approximate start time to get data (inclusive). See notes.
end – The approximate end time for the data query (exclusive). See notes.
width – The approximate size of statistical windows to return, will contain summaries of the data between start to end of size width.
precise – Pass in precise=True to use the exact start, end and width values specified. See notes.
version – What version of the stream to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,statistical summary timeseries be returned as, default will be the server default.

Returns:

The statistical summary information of the stream as a timeseries.

Notes

By default (precise=False), the values provided to start, end and width may not fully align with the values actually used by the query. Instead, the window width used will be the largest power of 2 ns that is smaller than the provided window. Doing so aligns the query with BTrDB’s internal tree structure and thus increases query performance by several orders of magnitude.

Consequently, the actual time-range of the data pulled will be the aligned values that fall within the range [start, end).

StreamSet#

class pingthings.timeseries.StreamSet(streams)#

Bases: object

annotations(copy=True, refresh=None)#

Return the metadata that is not present in the tag-metadata of the StreamSet.

Parameters:

copy – Should the returned dict be copied or passed by reference, by default True
refresh – Do we need to get the latest metadata from the platform first, by default False

Returns:

A mapping of annotation metadata key,value pairs.

changes(from_version_map, to_version_map, resolution)#

Return a mapping of intervals of time that have changed between two versions for each stream.

Warning

Advanced user feature

This method can have unexpected return values if you are not familiar with the underlying timeseries database that is part of the PredictiveGrid platform.

Note

Time ranges can overlap with previously inserted data

Parameters:

from_version_map – The stream versions to compare from (exclusive).
to_version_map – The stream versions to compare to (inclusive).
resolution – The “coarseness” of change detection in tree width. A resolution of 0 typically returns the finest granularity of changed intervals (which is still aligned with our internal tree structure), a resolution of 30, would be aligned exactly as a pointwidth of 2^30 nanoseconds. This resolution is the same as a pointwidth, please refer to the explanation page here on stat point widths of our tree.

Returns:

A dictionary keyed on stream UUID’s where the value is a list of dictionaries representing changed intervals. Typically the return looks like: {UUID: [{"start": <nanosecond_timestamp>, "end": <nanosecond_timestamp>}, ...].

Return type:

changes

Examples

>>> # Suppose we want to iterate over all changes between version 9 and 42
>>> # and we have a two stream streamset
>>> from_version_map = {s.uuid: 9 for s in streamset}
>>> to_version_map = {s.uuid: 42 for s in streamset}
>>> changed_range = streamset.changes(from_version_map=from_version_map,
...                                        to_version_map=to_version_map, resolution=0):
>>> for stream_uuid, changed_range in changed_range.items():
...     print(f"Stream UUID: {stream_uuid} Changed intervals: {changed_range}")

>>> # Suppose we want to iterate over all changes between version 42 and the latest data that has not been inserted yet
>>> from_version_map = {s.uuid: 42 for s in streamset}
>>> to_version_map = {s.uuid: 0 for s in streamset}
>>> changed_range = streamset.changes(from_version_map=from_version_map,
...                                        to_version_map=to_version_map, resolution=0):
>>> for stream_uuid, changed_range in changed_range.items():
...     print(f"Stream UUID: {stream_uuid} Changed intervals: {changed_range}")

count(start=None, end=None, versions=None, precise=None)#

Get the total count of raw measurements that are present in each stream.

Parameters:

start – Bound the lower end of this query by a start time, by default MINIMUM_TIME.
end – Bound the upper end of this query by an end time, by default MAXIMUM_TIME
versions – Versions of the streams to query against, by default None, which means a version of 0 is used.
precise – Do we need an exact count or is an estimate reasonable, by default False

Returns:

Mapping of individual stream.uuid’s to count values.

earliest()#

Find the earliest point (in time) that is present in the streams.

Returns:: The earliest points in the streams.

filter(collection=None, tags=None, annotations=None, refresh_metadata=None)#

Create a new StreamSet that is a subset of the streams based on metadata filtering.

Parameters:

collection – The collection string to filter by, by default None
tags – Tag metadata to filter on, by default None
annotations – Annotation metadata to filter on, by default None
refresh_metadata – Should we use the latest metadata of the streams to filter on, by default True

Returns:

A subset of the streams that match the provided filters, if any.

flush()#

Force a flush of the buffered data to persistent storage for all streams.

If data was present, the version number will be positively incremented.

Returns:: version dictionary mapping.
Return type:: The version number of each stream after the flush as a uuid

get_latest_version()#

Get the latest version of each stream.

Returns:: A stream.uuid, latest version mapping.

get_version_at_time(time)#

Get the version of each stream at time realtime.

Parameters:: realtime – The time to check the stream version against
Returns:: A stream.uuid, version mapping.

insert(data_map, merge_policy=None, round_spec=None)#

Insert new timeseries data into the streams.

Tip

Default merge policy has changed!

Starting with the new pingthings api, the default merge policy is now replace. Please refer to the merge policy docs

Tip

Data must follow a specific schema for insertion

Your data pyarrow table or record batch must have a schema that matches the TIME_VALUE_F64_SCHEMA, defined. Order of the columns matter.

Parameters:

data_map – A mapping of stream.uuid to pyarrow.Table timeseries data.
merge_policy – How to handle when duplicate timestamp points are inserted into the stream
round_spec – Set how many bits of precision to keep inserted data. Default behavior is the full 64 bits.

Returns:

A mapping of stream.uuid, version of the stream after insertion.

latest()#

Find the latest point (in time) that is present in the streams.

Returns:: The latest point in the streams, if present.

nearest(time, backward=None)#

raw_values(start, end, snap_period=None, versions=None, schema=None)#

Return the raw time,value pairs of data from the streams.

Tip

StreamSet raw value queries de-duplicate timestamps

Current behavior for the accelerated “multistream” streamset queries de-duplicates timestamps internally.

Parameters:

start – Start time to get data (inclusive)
end – End time for the data query (exclusive)
snap_period – What period of time (if any) should the data be aligned to?
versions – What versions of the streams to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,values be returned as, default will be the server default.

Returns:

A table of timeseries data in the interval of [start, end)

raw_values_iter(start, end, snap_period_ns=None, versions=None, schema=None)#

refresh_metadata()#: Update all metadata for each stream in the StreamSet

tags(copy=True, refresh=None)#

Return the tag-based metadata of the StreamSet.

Parameters:

copy – Should the tag dictionary be copied or passed by reference, by default True
refresh – Do we need to query the platform to get the metadata first, by default False

Returns:

A mapping of tag metadata key,value pairs.

update(collection=Ellipsis, tags=None, custom_tags=None, annotations=None, custom_annotations=None, names=None, replace_tags=None, replace_annotations=None)#

windowed_values(start, end, width, precise=None, versions=None, schema=None, depth=0)#

Return statistical summaries of the data.

Parameters:

start – The approximate start time to get data (inclusive). See notes.
end – The approximate end time for the data query (exclusive). See notes.
width – The approximate size of statistical windows to return, will contain summaries of the data between start to end of size width.
precise – Pass in precise=True to use the exact start, end and width values specified. See notes.
versions – What versions of the stream to query against, using the default of None will use version 0.
schema – What pyarrow.Schema should the time,statistical summary timeseries be returned as, default will be the server default.

Returns:

The statistical summary information of the stream as a timeseries.

Notes

Consequently, the actual time-range of the data pulled will be the aligned values that fall within the range [start, end).

Utility Functions#

pingthings.timeseries.utils.currently_as_ns()#: Returns the current UTC time as nanoseconds since epoch

pingthings.timeseries.utils.to_nanoseconds(val)#

Converts datetime, datetime64, float, str (RFC 2822) to nanoseconds. If a datetime-like object is received then nanoseconds since epoch is returned.

Note

The following string formats are supported for conversion.

Format String	Description
`%Y-%m-%d %H:%M:%S.%f%z`	most common RFC3339 nanoseconds
`%Y-%m-%d %H:%M:%S.%f`	expects UTC default timezone
`%Y-%m-%dT%H:%M:%S.%fZ`	JSON encoding, UTC timezone
`%Y-%m-%dT%H:%M:%SZ`	JSON encoding, UTC timezone
`%Y-%m-%dT%H:%M:%S.%f%z`	less common JSON-ish encoding
`%Y-%m-%dT%H:%M:%S.%f`	for completeness, UTC+0 default timezone
`%Y-%m-%d %H:%M:%S%z`	human readable date time with TZ
`%Y-%m-%d %H:%M:%S`	human readable date time UTC+0 default
`%Y-%m-%d`	human readable date time at midnight UTC default

Parameters:: val – An object to convert to nanoseconds
Returns:: Object converted to nanoseconds

pingthings.timeseries.utils.datetime_to_ns(dt)#

Converts a datetime object to nanoseconds since epoch. If a timezone-aware object is received then it will be converted to UTC. If a timezone-naive object is received then it will be assumed to be in UTC.

Parameters:: dt – The datetime object to convert.
Returns:: The number of nanoseconds since the Unix epoch.

pingthings.timeseries.utils.ns_delta(days=0.0, hours=0.0, minutes=0.0, seconds=0.0, milliseconds=0.0, microseconds=0.0, nanoseconds=0)#

Similar to timedelta, ns_delta represents a span of time but as the total number of nanoseconds.

Parameters:

days – Days (as 24 hours) to convert to nanoseconds
hours – Hours to convert to nanoseconds
minutes – Minutes to convert to nanoseconds
seconds – Seconds to convert to nanoseconds
milliseconds – Milliseconds to convert to nanoseconds
microseconds – Microseconds to convert to nanoseconds
nanoseconds – Nanoseconds to add to the time span

Returns:

Amount of time in nanoseconds

pingthings.timeseries.utils.ns_to_datetime(val)#

Convert nanoseconds relative to the Unix epoch to a UTC datetime.

Tip

Due to the limitation of Python’s datetime module, the datetime representation is only accurate to the nearest microsecond.

Parameters:: ns – Nanoseconds since epoch (can be negative)
Returns:: A timezone-aware datetime in UTC.

pingthings.timeseries.utils.nearest_pointwidth(days=0.0, hours=0.0, minutes=0.0, seconds=0.0, milliseconds=0.0, microseconds=0.0, nanoseconds=0, hertz=0.0)#

Convert time delta of days, hours, minutes, seconds, milliseconds, microseconds, nanoseconds or hertz to the nearest point-width.

Parameters:

days – Days (as 24 hours) to convert to nanoseconds
hours – Hours to convert to nanoseconds
minutes – Minutes to convert to nanoseconds
seconds – Seconds to convert to nanoseconds
milliseconds – Milliseconds to convert to nanoseconds
microseconds – Microseconds to convert to nanoseconds
nanoseconds – Nanoseconds to add to the time span
hertz – Alternatively, provide the frequency in Hz

Returns:

The nearest point-width-aligned value.

Useful Constants#

pingthings.timeseries.constants.PW#

Durations that naturally align with the platform timeseries data structure. Utilizing windowed value queries where the start, end and averaging windows align with these point-widths ensures optimal query performance.

These point-width durations are as follows:

point-width	number of nanoseconds	duration
PW[0]	1	1 ns
PW[1]	2	2 ns
PW[2]	4	4 ns
PW[3]	8	8 ns
PW[4]	16	16 ns
PW[5]	32	32 ns
PW[6]	64	64 ns
PW[7]	128	128 ns
PW[8]	256	256 ns
PW[9]	512	512 ns
PW[10]	1024	1 μs 24 ns
PW[11]	2048	2 μs 48 ns
PW[12]	4096	4 μs 96 ns
PW[13]	8192	8 μs 192 ns
PW[14]	16384	16 μs 384 ns
PW[15]	32768	32 μs 768 ns
PW[16]	65536	65 μs 536 ns
PW[17]	131072	131 μs 72 ns
PW[18]	262144	262 μs 144 ns
PW[19]	524288	524 μs 288 ns
PW[20]	1.04858e+06	1 ms 48 μs 576 ns
PW[21]	2.09715e+06	2 ms 97 μs 152 ns
PW[22]	4.1943e+06	4 ms 194 μs 304 ns
PW[23]	8.38861e+06	8 ms 388 μs 608 ns
PW[24]	1.67772e+07	16 ms 777 μs 216 ns
PW[25]	3.35544e+07	33 ms 554 μs 432 ns
PW[26]	6.71089e+07	67 ms 108 μs 864 ns
PW[27]	1.34218e+08	134 ms 217 μs 728 ns
PW[28]	2.68435e+08	268 ms 435 μs 456 ns
PW[29]	5.36871e+08	536 ms 870 μs 912 ns
PW[30]	1.07374e+09	1 sec 73 ms 741 μs
PW[31]	2.14748e+09	2 sec 147 ms 483 μs
PW[32]	4.29497e+09	4 sec 294 ms 967 μs
PW[33]	8.58993e+09	8 sec 589 ms 934 μs
PW[34]	1.71799e+10	17 sec 179 ms 869 μs
PW[35]	3.43597e+10	34 sec 359 ms 738 μs
PW[36]	6.87195e+10	1 min 8 sec 719 ms
PW[37]	1.37439e+11	2 mins 17 sec 438 ms
PW[38]	2.74878e+11	4 mins 34 sec 877 ms
PW[39]	5.49756e+11	9 mins 9 sec 755 ms
PW[40]	1.09951e+12	18 mins 19 sec 511 ms
PW[41]	2.19902e+12	36 mins 39 sec 23 ms
PW[42]	4.39805e+12	1 hour 13 mins 18 sec
PW[43]	8.79609e+12	2 hours 26 mins 36 sec
PW[44]	1.75922e+13	4 hours 53 mins 12 sec
PW[45]	3.51844e+13	9 hours 46 mins 24 sec
PW[46]	7.03687e+13	19 hours 32 mins 48 sec
PW[47]	1.40737e+14	1 day 15 hours 5 mins
PW[48]	2.81475e+14	3 days 6 hours 11 mins
PW[49]	5.6295e+14	6 days 12 hours 22 mins
PW[50]	1.1259e+15	13 days 44 mins 59 sec
PW[51]	2.2518e+15	26 days 1 hour 29 mins
PW[52]	4.5036e+15	52 days 2 hours 59 mins
PW[53]	9.0072e+15	104 days 5 hours 59 mins
PW[54]	1.80144e+16	208 days 11 hours 59 mins
PW[55]	3.60288e+16	1 year 51 days 23 hours
PW[56]	7.20576e+16	2 years 103 days 23 hours
PW[57]	1.44115e+17	4 years 207 days 23 hours
PW[58]	2.8823e+17	9 years 50 days 23 hours
PW[59]	5.76461e+17	18 years 101 days 23 hours
PW[60]	1.15292e+18	36 years 203 days 23 hours

alias of [<constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>, <constants._PW object>]

pingthings.timeseries.constants.MAXIMUM_TIME#: alias of 3458764513820540927

pingthings.timeseries.constants.MINIMUM_TIME#: alias of -1152921504606846976

pingthings.timeseries.constants.TIME_VALUE_F64_SCHEMA#: alias of time: timestamp[ns, tz=UTC] not null value: double not null

pingthings.timeseries.constants.TIME_VALUE_F32_SCHEMA#: alias of time: timestamp[ns, tz=UTC] not null value: float not null

pingthings.timeseries.constants.STAT_F64_SCHEMA#: alias of time: timestamp[ns, tz=UTC] not null min: double not null mean: double not null max: double not null count: uint64 not null stddev: double not null

pingthings.timeseries.constants.STAT_F32_SCHEMA#: alias of time: timestamp[ns, tz=UTC] not null min: float not null mean: float not null max: float not null count: uint64 not null stddev: float not null

Client#

class pingthings.timeseries.Client#

Bases: object

create(uuid, collection, tags=None, annotations=None)#

Create a new stream with the given uuid and collection name.

Parameters:

uuid – The UUID for the new stream.
collection – The collection to which the stream belongs.
tags – Tags associated with the stream.
annotations – Annotations for the stream.

Returns:

The newly created stream.

Note

Required and Allowable Tags/Annotations

The name field in tags must be set and currently all tags/annotations must be encoded as strings and two streams can not share both the same collection and name.

Tip

Available tags in the stream table

column_name	Description
name	name of the stream
source	Where did the stream come from
device_id	What device is the stream attached to
samples_per_second	What is the nominal report rate of the data, in Hz?
time_precision	How accurate is each timestep (+- nanoseconds)
value_precision	Bit precision
dynamic	Can the timeseries stream send us data at varying report rates?
continuous	Should we expect a continuous stream of data?
description	Additional information about the timeseries stream
alt_name	Additional name that the stream may be referenced by
alt_id	Alternative ID (most likely SignalID from STTP)
norm_factor	A factor to scale the measurement by to normalize. Like “baseKV” for voltage, etc.
scaling_factor	The “m” in y=mx + b factor, line-line scaling factor, etc
bias_factor	The “b” in y=mx+b
hidden	Whether the stream should be presented to non-admin users

create_device(device)#

Creates a device with fields matching the passed dictionary. name can not be None or an empty string, two devices can not share the same name and geo, and enabled=True by default.

Parameters:: device – dictionary with the possible keys and description of expected values below.
Returns:: A dictionary with the same fields passed with create plus the device["id"] set

Tip

Available columns in the device table

column_name	Description
name	Name of device
device_type	Type of device (DFR/Relay/etc)
description	User readable description
geo	Coordinates, a dict {“latitude”:float,”logitude”:float}
elevation	Elevation of the device, float
alt_id	Alternative id, int
alt_name	Alternative name
enabled	Whether the device is expected to be sending data, bool default: True
owner	Who owns the device
protocol	What protocol the device is using to send data
model_name	The model name of the device
vendor	The company that creates/sells the device

create_streams(uuids, collection, tags=None, custom_tags=None, annotations=None, custom_annotations=None, names=None)#

create_unit(name, symbol, base, canonical, source_name)#

Create a new representation of a unit definition in the platform.

Units are a main way to assign further meaning to the timeseries stored in the platform. There are usually a handful of default units provided with each cluster, however, based on user needs, additional units will need to be included to support the wide range of timeseries added to the platform.

Parameters:

name – Name of the new unit.
symbol – Symbol of the new unit.
base – Multiplicative factor to convert the unit to the canonical representation.
canonical – The “main” unit that this unit can be ultimately converted to.
source_name – List of other potential names that this unit can be considered in the platform. For example, if the unit name is degrees the source_name could include the following: ["deg", "VPHA", "IPHA", "VpHA", "Deg", "DEG"]

Returns:

Dictionary with fields corresponding to the values in the database.

Examples

>>> unit_def = {"name":"volts", "symbol":"V", "base":1.0, "canonical":"volts", source_name:["VPHM", "voltage", "VOLTS", "VLTGE"]}
>>> unit = client.create_unit(**unit_def)
>>> print(unit)
{'id': 6,
'name': 'volts',
'symbol': 'V',
'canonical': 'volts',
'base': 1.0,
'source_name': ['VPHM', 'voltage', 'VOLTS', 'VLTGE']}

delete_device(device_id)#

Deletes the device with the specified id

Parameters:: device_id – Corresponding id of the device to be deleted
Returns:: None

delete_unit(unit_id)#

Remove a unit from the platform.

Parameters:: unit_id – Integer corresponding to the id of the unit in the database.
Returns:: None

Examples

>>> client.delete_unit(6)

get_collection_properties(collection)#

Get properties of a collection.

Parameters:: collection – The name of the collection.

Examples

Get the retention policy of a collection.

>>> conn.get_collection_properties("bar")
{'retention': {'remove_older_than': datetime.timedelta(days=7)}}

get_device(device_id)#

Returns a device with a matching id in the device table.

Parameters:: device_id – Integer corresponding to the id of the device in the database.
Returns:: Dictionary with fields corresponding to the values in the database.

get_devices(device_ids)#

Returns devices with matching ids in the device table.

Parameters:: device_ids – List of integers corresponding to the ids of the devices in the database.
Returns:: Dictionary with fields corresponding to the values in the database.

get_unit(unit_id)#

Returns a unit with a matching id in the unit table.

Parameters:: unit_id – Integer corresponding to the id of the unit in the database.
Returns:: Dictionary with fields corresponding to the values in the unit table.

info()#

Retrieve information about the server and proxy server the client is connected to.

Returns:: A dictionary containing server and proxy server information.

list_collections(prefix=None)#

Returns a list of collection paths using the prefix argument for filtering.

Parameters:: prefix – Filter collections that start with the string provided, if none passed, will list all collections.
Returns:: All collections that match the provided prefix.

Examples

Assuming we have the following collections in the platform: foo, bar, foo/baz, bar/baz

>>> conn = pt.connect()
>>> conn.list_collections().sort()
["bar", "bar/baz", "foo", "foo/bar"]

>>> conn.list_collections(prefix="foo")
["foo", "foo/bar"]

list_devices()#

Returns a list of devices the user has permission to see.

Returns:: Dictionary with keys matching the columns in the device table.

list_units()#

Returns a list of units generated in the database.

Returns:: Dictionary with keys matching the columns in the units table.

set_collection_retention(collection, override_per_stream, remove_older_than=None)#

Set retention policy on a collection of streams.

Parameters:

collection – The name of the collection.
override_per_stream – Whether stream-specific retention policy should be overridden.
remove_older_than – Trim time period - after which the data will get removed. Not specifying this parameter disables the trimming.

Examples

Keep the data for only one week.

>>> conn.set_collection_retention("bar", false, datetime.timedelta(days=7))

set_retention(uuids, remove_older_than=None)#

Set retention policy on a list of streams.

Parameters:

uuids – The UUIDs of the streams to apply the retention policy to.
remove_older_than – Trim time period after which the data can be removed. Specifying None or not passing the parameter disables trimming.

stream_from_uuid(py_uuid)#

Retrieve a stream based on its UUID.

Parameters:: uuid – The UUID of the stream.
Returns:: The stream associated with the provided UUID.
Raises:: TypeError – If the provided uuid is not a valid UUID

streams_in_collection(collection=None, is_collection_prefix=None, tags=None, annotations=None)#

Search for streams matching given parameters

Parameters:

collection – collections to use when searching for streams, case sensitive.
is_collection_prefix – Whether the collection is a prefix of the whole collection name.
tags – The tags to identify the stream.
annotations – The annotations to identify the stream.

Returns:

The grouping of streams matching given parameters.

streamset_from_uuids(uuids, fetch_metadata=None)#

Return a StreamSet from an iterable of UUIDs.

Parameters:

uuids – List of stream identifiers
fetch_metadata – Whether to fetch metadata for the streams in the set. Default is True.

Warning

Advanced user feature

Be cautious about using fetch_metadata=False. Many stream metadata values like collection, name, unit, tags, annotations will not be available, meaning filtering and other operations that require metadata will not work.

Returns:: The StreamSet associated with the provided iterable of UUIDs.

update_device(device, replace=False)#

Updates a device with matching device["id"] using fields matching the passed dictionary. name can not be None or an empty string.

Parameters:

device – dictionary with the possible keys and description of expected values below.
replace – whether to change all unspecified values to None (device["enabled"]=True by default).

Returns:

Dictionary with the updated fields and original unmodified fields of the device

Note

Update Behavior

Tip

Available columns in the device table

column_name	Description
id	Internal id of device
name	Name of device
device_type	Type of device (DFR/Relay/etc)
description	User readable description
geo	Coordinates, a dict {“latitude”:float,”logitude”:float}
elevation	Elevation of the device, float
alt_id	Alternative id, int
alt_name	Alternative name
enabled	Whether the device is expected to be sending data, bool default: True
owner	Who owns the device
protocol	What protocol the device is using to send data
model_name	The model name of the device
vendor	The company that creates/sells the device

update_unit(unit)#

Updates a unit_dict with matching id using fields from the passed dictionary.

Modify the unit where the unit_dict['id'] is the id number of the unit you want to update.

Note

Update Behavior

You are not requried to include fields that you want to keep the same. Although including them will not cause issues.

Parameters:: unit_dict – Dictionary with fields corresponding to the values in the database you want to update
Returns:: Dictionary with the updated fields and original unmodified fields of the unit

Examples

>>> original_unit = client.get_unit(6)
>>> print(original_unit)
{'id': 6,
'name': 'volts',
'symbol': 'V',
'canonical': 'volts',
'base': 1.0,
'source_name': ['VPHM', 'voltage', 'VOLTS', 'VLTGE']}
>>> updated_unit = {"id":6, "name":"volt"}
>>> client.update_unit(updated_unit)
{'id': 6,
 'name': 'volt',
 'symbol': 'V',
 'canonical': 'volts',
 'base': 1.0,
 'source_name': ['VPHM', 'voltage', 'VOLTS', 'VLTGE']}

Timeseries Documentation

Contents

Timeseries Documentation#

Stream#

StreamSet#

Utility Functions#

Useful Constants#

Client#