StreamSet

pingthings.timeseries.client.StreamSet ¶

StreamSet(streams: list[Stream] = [])

A collection of Stream objects.

Create a collection of stream objects.

Methods:

Name	Description
`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 `StreamSet` 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.
`raw_values`	Return the raw time,value pairs of data from the streams.
`refresh_metadata`	Update all metadata for each stream in the `StreamSet`
`windowed_values`	Return statistical summaries of the data.

Functions¶

count ¶

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

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

Parameters:

Name	Type	Description	Default
`start` ¶	`Optional[int \| datetime \| Timestamp]`	Bound the lower end of this query by a start time, by default MINIMUM_TIME.	`MINIMUM_TIME`
`end` ¶	`Optional[int \| datetime \| Timestamp]`	Bound the upper end of this query by an end time, by default MAXIMUM_TIME	`MAXIMUM_TIME`
`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, Optional[int]]`	Mapping of individual `stream.uuid`'s to `count` values.

earliest ¶

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 ¶

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

Create a new StreamSet 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[StreamSet]`	A subset of the streams that match the provided filters, if any.

flush ¶

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 ¶

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 ¶

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 ¶

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 when duplicate timestamp points are inserted into the stream	`'replace'`

Returns:

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

latest ¶

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 ¶

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

Pin the versions of all streams to specific values.

raw_values ¶

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

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

StreamSet raw value queries de-duplicate timestamps

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

Parameters:

Name	Type	Description	Default
`start` ¶	`int \| datetime \| Timestamp`	Start time to get data (inclusive)	required
`end` ¶	`int \| datetime \| Timestamp`	End time for the data query (exclusive)	required
`snap_period` ¶	`Optional[int \| timedelta \| 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 ¶

refresh_metadata() -> None

Update all metadata for each stream in the StreamSet

windowed_values ¶

windowed_values(
    start: int | datetime | Timestamp,
    end: int | datetime | Timestamp,
    width: int | timedelta | Timedelta | _PW,
    precise: Optional[bool] = False,
    versions: Optional[list[int]] = None,
    schema: Optional[Schema] = STAT_F32_SCHEMA,
) -> Table

Return statistical summaries of the data.

Parameters:

Name	Type	Description	Default
`start` ¶	`int \| datetime \| Timestamp`	The approximate start time to get data (inclusive). See notes.	required
`end` ¶	`int \| datetime \| Timestamp`	The approximate end time for the data query (exclusive). See notes.	required
`width` ¶	`int \| timedelta \| Timedelta \| _PW`	The approximate size of statistical windows to return, will contain summaries of the data between `start` to `end` of size `width`.	required
`precise` ¶	`Optional[bool]`	Pass in `precise=True` to use the exact `start`, `end` and `width` values specified. See notes.	`False`
`versions` ¶	`Optional[list[int]]`	What versions of the stream to query against, using the default of `None` will use version 0.	`None`
`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
`Table`	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

pingthings.timeseries.client.StreamSet ¶

Functions¶

count ¶

`start` ¶

`end` ¶

`versions` ¶

`precise` ¶

earliest ¶

filter ¶

`collection` ¶

`tags` ¶

`annotations` ¶

`refresh_metadata` ¶

flush ¶

get_latest_version ¶

get_version_at_time ¶

`realtime` ¶

insert ¶

`data_map` ¶

`merge_policy` ¶

latest ¶

pin_versions ¶

raw_values ¶

`start` ¶

`end` ¶

`snap_period` ¶

`versions` ¶

`schema` ¶

refresh_metadata ¶

windowed_values ¶

`start` ¶

`end` ¶

`width` ¶

`precise` ¶

`versions` ¶

`schema` ¶

StreamSet

pingthings.timeseries.client.StreamSet ¶

Functions¶

count ¶

start ¶

end ¶

versions ¶

precise ¶

earliest ¶

filter ¶

collection ¶

tags ¶

annotations ¶

refresh_metadata ¶

flush ¶

get_latest_version ¶

get_version_at_time ¶

realtime ¶

insert ¶

data_map ¶

merge_policy ¶

latest ¶

pin_versions ¶

raw_values ¶

start ¶

end ¶

snap_period ¶

versions ¶

schema ¶

refresh_metadata ¶

windowed_values ¶

start ¶

end ¶

width ¶

precise ¶

versions ¶

schema ¶

`start` ¶

`end` ¶

`versions` ¶

`precise` ¶

`collection` ¶

`tags` ¶

`annotations` ¶

`refresh_metadata` ¶

`realtime` ¶

`data_map` ¶

`merge_policy` ¶

`start` ¶

`end` ¶

`snap_period` ¶

`versions` ¶

`schema` ¶

`start` ¶

`end` ¶

`width` ¶

`precise` ¶

`versions` ¶

`schema` ¶