AsyncStreamSet

pingthings.timeseries.async_client.AsyncStreamSet ¶

AsyncStreamSet(streams: list[AsyncStream] = [])

A collection of AsyncStream objects.

Create a collection of AsyncStream objects.

Methods:

Name	Description
`aligned_windows`	Return statistical summary timeseries data aligned with the internal tree structure of the database.
`count`	Get the total count of raw measurements that are present in each stream.
`earliest`	Find the earliest point (in time) that is present in the streams.
`filter`	Create a new `AsyncStreamSet` that is a subset of the streams based on metadata filtering.
`flush`	Force a flush of the buffered data to persistent storage for all streams.
`get_latest_version`	Get the latest version of each stream.
`get_version_at_time`	Get the version of each stream at time `realtime`.
`insert`	Insert new timeseries data into the streams.
`latest`	Find the latest point (in time) that is present in the streams.
`pin_versions`	Pin the versions of all streams to specific values.
`precise_windows`	Return custom-sized statistical summaries of the data.
`raw_values`	Return the raw time,value pairs of data from the streams.
`refresh_metadata`	Update all metadata for each stream in the `AsyncStreamSet`

Functions¶

aligned_windows `async` ¶

aligned_windows(
    start: int | datetime,
    end: int | datetime,
    point_width: int | _PW,
    versions: Optional[list[int]] = None,
    single_stream_schema: Optional[
        Schema
    ] = STAT_F32_SCHEMA,
) -> list[Table | 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.

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.

Parameters:

Name	Type	Description	Default
`start` ¶	`int \| datetime`	Start time to get data (inclusive)	required
`end` ¶	`int \| datetime`	End time for the data query (exclusive)	required
`point_width` ¶	`int \| _PW`	What size statistical windows to return, aligned to a size of the internal tree. Value will be interpreted as \(2^{point\_width}\).	required
`versions` ¶	`Optional[list[int]]`	What version of the stream to query against, using the default of `None` will use version 0.	`None`
`single_stream_schema` ¶	`Optional[Schema]`	What `pyarrow.Schema` should the time,statistical summary timeseries be returned as, default will be the server default.	`STAT_F32_SCHEMA`

Returns:

Type	Description
`list[Table \| None]`	Statistical summary timeseries of the streams between \([start, end)\)

count `async` ¶

count(
    start: Optional[int] = None,
    end: Optional[int] = None,
    versions: Optional[int | dict[UUID, int]] = None,
    precise: bool = False,
) -> dict[UUID, int]

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

Parameters:

Name	Type	Description	Default
`start` ¶	`Optional[int]`	Bound the lower end of this query by a start time, by default MINIMUM_TIME.	`None`
`end` ¶	`Optional[int]`	Bound the upper end of this query by an end time, by default MAXIMUM_TIME	`None`
`versions` ¶	`Optional[int \| dict[UUID, int]]`	Versions of the streams to query against, by default None, which means a version of 0 is used.	`None`
`precise` ¶	`bool`	Do we need an exact count or is an estimate reasonable, by default False	`False`

Returns:

Type	Description
`dict[UUID, int]`	Mapping of individual `stream.uuid`'s to `count` values.

earliest `async` ¶

earliest() -> dict[UUID, Optional[Point]]

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

Returns:

Type	Description
`dict[UUID, Optional[Point]]`	The earliest points in the streams.

filter `async` ¶

filter(
    collection: Optional[str] = None,
    tags: Optional[dict[str, str]] = None,
    annotations: Optional[dict[str, str]] = None,
    refresh_metadata: bool = True,
) -> Optional[AsyncStreamSet]

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

Parameters:

Name	Type	Description	Default
`collection` ¶	`Optional[str]`	The collection string to filter by, by default None	`None`
`tags` ¶	`Optional[dict[str, str]]`	Tag metadata to filter on, by default None	`None`
`annotations` ¶	`Optional[dict[str, str]]`	Annotation metadata to filter on, by default None	`None`
`refresh_metadata` ¶	`bool`	Should we use the latest metadata of the streams to filter on, by default True	`True`

Returns:

Type	Description
`Optional[AsyncStreamSet]`	A subset of the streams that match the provided filters, if any.

flush `async` ¶

flush() -> dict[UUID, int]

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:

Type	Description
`dict[UUID, int]`	The version number of each stream after the flush as a uuid:version dictionary mapping.

get_latest_version `async` ¶

get_latest_version() -> dict[UUID, int]

Get the latest version of each stream.

Returns:

Type	Description
`dict[UUID, int]`	A `stream.uuid`, latest `version` mapping.

get_version_at_time `async` ¶

get_version_at_time(
    realtime: int | datetime,
) -> dict[UUID, int]

Get the version of each stream at time realtime.

Parameters:

Name	Type	Description	Default
`realtime` ¶	`int \| datetime`	The time to check the stream version against	required

Returns:

Type	Description
`dict[UUID, int]`	A `stream.uuid`, `version` mapping.

insert `async` ¶

insert(
    data_map: dict[UUID, Table],
    merge_policy: MergePolicy = "replace",
) -> dict[UUID, int]

Insert new timeseries data into the streams.

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

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:

Name	Type	Description	Default
`data_map` ¶	`dict[UUID, Table]`	A mapping of `stream.uuid` to `pyarrow.Table` timeseries data.	required
`merge_policy` ¶	`MergePolicy`	How to handle duplicate timestamps in the data	`'replace'`

Returns:

Type	Description
`dict[UUID, int]`	A mapping of `stream.uuid`, `version` of the stream after insertion.

latest `async` ¶

latest() -> dict[UUID, Optional[Point]]

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

Returns:

Type	Description
`dict[UUID, Optional[Point]]`	The latest point in the streams, if present.

pin_versions `async` ¶

pin_versions(versions: Optional[list[int]] = None) -> None

Pin the versions of all streams to specific values.

precise_windows `async` ¶

precise_windows(
    start: int | datetime,
    end: int | datetime,
    width: int,
    depth: Optional[int] = 0,
    versions: Optional[list[int]] = None,
    single_stream_schema: Optional[
        Schema
    ] = STAT_F32_SCHEMA,
) -> list[Table | None]

Return custom-sized statistical summaries of the data.

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

Parameters:

Name	Type	Description	Default
`start` ¶	`int \| datetime`	Start time to get data (inclusive)	required
`end` ¶	`int \| datetime`	End time for the data query (exclusive)	required
`width` ¶	`int`	The size of statistical windows to return, will contain summaries of the data between `start` to `end` of size `width`.	required
`depth` ¶	`Optional[int]`	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.	`0`
`versions` ¶	`Optional[list[int]]`	What version of the streams to query against, using the default of `None` will use version 0.	`None`
`single_stream_schema` ¶	`Optional[Schema]`	What `pyarrow.Schema` should the time,statistical summary timeseries be returned as, default will be the server default.	`STAT_F32_SCHEMA`

Returns:

Type	Description
`list[Table \| None]`	The statistical summary information of the streams as a timeseries.

raw_values `async` ¶

raw_values(
    start: int | datetime,
    end: int | datetime,
    snap_period: int | timedelta = 0,
    versions: Optional[list[int]] = None,
    schema: Optional[Schema] = None,
) -> Table

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

Parameters:

Name	Type	Description	Default
`start` ¶	`int \| datetime`	Start time to get data (inclusive)	required
`end` ¶	`int \| datetime`	End time for the data query (exclusive)	required
`snap_period` ¶	`int \| timedelta`	What period of time (if any) should the data be aligned to?	`0`
`versions` ¶	`Optional[list[int]]`	What versions of the streams to query against, using the default of `None` will use version 0.	`None`
`schema` ¶	`Optional[Schema]`	What `pyarrow.Schema` should the time,values be returned as, default will be the server default.	`None`

Returns:

Type	Description
`Table`	A table of timeseries data in the interval of \([start, end)\)

refresh_metadata `async` ¶

refresh_metadata() -> None

Update all metadata for each stream in the AsyncStreamSet

AsyncStreamSet

pingthings.timeseries.async_client.AsyncStreamSet ¶

Functions¶

aligned_windows async ¶

start ¶

end ¶

point_width ¶

versions ¶

single_stream_schema ¶

count async ¶

start ¶

end ¶

versions ¶

precise ¶

earliest async ¶

filter async ¶

collection ¶

tags ¶

annotations ¶

refresh_metadata ¶

flush async ¶

get_latest_version async ¶

get_version_at_time async ¶

realtime ¶

insert async ¶

data_map ¶

merge_policy ¶

latest async ¶

pin_versions async ¶

precise_windows async ¶

start ¶

end ¶

width ¶

depth ¶

versions ¶

single_stream_schema ¶

raw_values async ¶

start ¶

end ¶

snap_period ¶

versions ¶

schema ¶

refresh_metadata async ¶

aligned_windows `async` ¶

`start` ¶

`end` ¶

`point_width` ¶

`versions` ¶

`single_stream_schema` ¶

count `async` ¶

`start` ¶

`end` ¶

`versions` ¶

`precise` ¶

earliest `async` ¶

filter `async` ¶

`collection` ¶

`tags` ¶

`annotations` ¶

`refresh_metadata` ¶

flush `async` ¶

get_latest_version `async` ¶

get_version_at_time `async` ¶

`realtime` ¶

insert `async` ¶

`data_map` ¶

`merge_policy` ¶

latest `async` ¶

pin_versions `async` ¶

precise_windows `async` ¶

`start` ¶

`end` ¶

`width` ¶

`depth` ¶

`versions` ¶

`single_stream_schema` ¶

raw_values `async` ¶

`start` ¶

`end` ¶

`snap_period` ¶

`versions` ¶

`schema` ¶

refresh_metadata `async` ¶