Stream

pingthings.timeseries.client.Stream ¶

Stream(
    conn: Client,
    stream: Optional[AsyncStream] = None,
    uuid: Optional[UUID] = None,
    known_to_exist: bool = False,
    collection: Optional[str] = None,
    tags: Optional[dict[str, str]] = None,
    annotations: Optional[dict[str, str]] = None,
    property_version: int = 0,
)

The base representation of a timeseries signal in the platform.

Parameters:

Name	Type	Description	Default
`conn` ¶	`Client`	An instance of the client connection to the platform database.	required
`stream` ¶	`Optional[AsyncStream]`	Asynchronous representation of this timeseries signal, by default None	`None`
`uuid` ¶	`Optional[UUID]`	Unique identifier of the timeseries signal, by default None	`None`
`known_to_exist` ¶	`bool`	If this stream is known to already exist, can ignore some roundtrips to the database, by default False	`False`
`collection` ¶	`Optional[str]`	The collection string that the signal belongs to, by default None	`None`
`tags` ¶	`Optional[dict[str, str]]`	Tag-based metadata of the signal, by default None	`None`
`annotations` ¶	`Optional[dict[str, str]]`	Non-tag metadata of the signal, by default None	`None`
`property_version` ¶	`int`	Internal version of the signal, by default 0	`0`

Returns:

Type	Description
`None`	The timeseries representation.

Methods:

Name	Description
`aligned_windows_iter`	Return statistical summary timeseries data aligned with the internal tree structure of the database.
`annotations`	Return the metadata that is not present in the tag-metadata of the stream.
`count`	Get the total count of raw measurements that are present in the stream.
`delete`	"Delete" all points between [`start`, `end`)
`earliest`	Find the earliest point (in time) that is present in the stream at `version`.
`flush`	Force a flush of the buffered data to persistent storage.
`get_latest_version`	Get the current version of the stream.
`get_version_at_time`	Return the version of the stream at the time provided.
`insert`	Add new timeseries data to the stream.
`latest`	Find the latest point (in time) that is present in the stream at `version`.
`nearest`	Get the nearest datapoint at `time` with `version`.
`obliterate`	Completely remove the stream from the platform.
`pin_version`	Set the version of the stream to a specific version number, useful when wanting reproducible data queries.
`pinned_version`	Return the pinned version of the stream.
`precise_windows_iter`	Return custom-sized statistical summaries of the data.
`raw_values`	Return the raw time,value pairs of data from the stream between `start` and `end`.
`raw_values_iter`	Return the raw timeseries data from the stream between `start` and `end`, but as an iterator instead of buffering all at once.
`refresh_metadata`	Retrieve and update all metadata for this stream.
`tags`	Return the tag-based metadata of the stream.
`update`	Update the stream metadata.
`windowed_values`	Return statistical summaries of the data.

Attributes:

Name	Type	Description
`collection`	`str \| None`	Return the collection the stream belongs to.
`name`	`str`	Return the name of the stream.
`uuid`	`UUID`	Return the unique identifier of the stream.

Attributes¶

collection `property` ¶

collection: str | None

Return the collection the stream belongs to.

Returns:

Type	Description
`str \| None`	The collection of the stream.

name `property` ¶

name: str

Return the name of the stream.

Returns:

Type	Description
`str`	The name of the stream.

uuid `property` ¶

uuid: UUID

Return the unique identifier of the stream.

Returns:

Type	Description
`UUID`	The unique identifier of the stream.

Functions¶

aligned_windows_iter ¶

aligned_windows_iter(
    start: int | datetime | Timestamp,
    end: int | datetime | Timestamp,
    point_width: Optional[int] = None,
    width: Optional[int | timedelta | Timedelta] = None,
    version: Optional[int] = None,
    schema: Optional[Schema] = STAT_F32_SCHEMA,
) -> Generator[Table, None, 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.

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:

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
`point_width` ¶	`Optional[int]`	What size statistical windows to return, aligned to a size of the internal tree. Value will be interpreted as \(2^{point\_width}\).	`None`
`width` ¶	`Optional[int \| timedelta \| Timedelta]`	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`.	`None`
`version` ¶	`Optional[int]`	What version 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`

Yields:

Type	Description
`Table`	Statistical summary timeseries of the stream between \([start, end)\)

annotations ¶

annotations(
    copy: bool = True, refresh: Optional[bool] = True
) -> Optional[dict[str, str]]

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

Parameters:

Name	Type	Description	Default
`copy` ¶	`bool`	Should the returned dict be copied or passed by reference, by default True	`True`
`refresh` ¶	`Optional[bool]`	Do we need to get the latest metadata from the platform first, by default True	`True`

Returns:

Type	Description
`Optional[dict[str, str]]`	A mapping of annotation metadata key,value pairs.

count ¶

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

Get the total count of raw measurements that are present in the 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`
`version` ¶	`Optional[int]`	Version of the stream to query against, by default None, which means a version of 0 is used.	`None`
`precise` ¶	`Optional[bool]`	Do we need an exact count or is an estimate reasonable, by default False	`False`

Returns:

Type	Description
`int`	Count of points in the stream.

delete ¶

delete(start: int, end: int) -> int

"Delete" all points between [start, end)

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

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:

Name	Type	Description	Default
`start` ¶	`int`	Time to begin deleting points in the stream (inclusive).	required
`end` ¶	`int`	Time to stop deleting points in the stream (exclusive).	required

Returns:

Type	Description
`int`	The updated version number of the stream.

earliest ¶

earliest(version: Optional[int] = None) -> Optional[Point]

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

Parameters:

Name	Type	Description	Default
`version` ¶	`Optional[int]`	The version of the stream to query against, by default None, which means a version of 0 will be used.	`None`

Returns:

Type	Description
`Optional[Point]`	The earliest point in the stream at `version`.

flush ¶

flush() -> int

Force a flush of the buffered data to persistent storage.

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

Returns:

Type	Description
`int`	The major version of the stream after the flush.

get_latest_version ¶

get_latest_version() -> int

Get the current version of the stream.

Returns:

Type	Description
`int`	The version number of the stream.

get_version_at_time ¶

get_version_at_time(realtime: int) -> int

Return the version of the stream at the time provided.

Parameters:

Name	Type	Description	Default
`realtime` ¶	`int`	The time to check the version of the stream at, in nanoseconds.	required

Returns:

Type	Description
`int`	The version of the stream at time `realtime`.

insert ¶

insert(
    data: Table | RecordBatch,
    merge_policy: MergePolicy = "never",
) -> int

Add new timeseries data to the stream.

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` ¶	`Table \| RecordBatch`	A pyarrow table or record batch of time series data.	required
`merge_policy` ¶	`MergePolicy`	How should the database handle data with the same timestamp, by default "never"	`'never'`

Returns:

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

latest ¶

latest(version: Optional[int] = None) -> Optional[Point]

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

Parameters:

Name	Type	Description	Default
`version` ¶	`Optional[int]`	The version of the stream to query against, by default None, which means a `version` of 0 will be used.	`None`

Returns:

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

nearest ¶

nearest(
    time: int | datetime | Timestamp,
    backward: bool = False,
    version: Optional[int] = None,
) -> Optional[Point]

Get the nearest datapoint at time with version.

Parameters:

Name	Type	Description	Default
`time` ¶	`int \| datetime \| Timestamp`	The time to query for the nearest point.	required
`backward` ¶	`bool`	Can the query find the closest time that is before `time`, by default False	`False`
`version` ¶	`Optional[int]`	Version of the stream to query against, by default None, which means version 0 will be used.	`None`

Returns:

Type	Description
`Optional[Point]`	The time,value Point that is closest to the time in question, if available.

obliterate ¶

obliterate() -> None

Completely remove the stream from the platform.

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 ¶

pin_version(version: Optional[int] = None) -> int

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.

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:

Name	Type	Description	Default
`version` ¶	`Optional[int]`	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`.	`None`

Returns:

Type	Description
`int`	The version pinned.

pinned_version ¶

pinned_version() -> int

Return the pinned version of the stream.

Returns:

Type	Description
`int`	The version of the stream it is pinned to.

precise_windows_iter ¶

precise_windows_iter(
    start: int | datetime | Timestamp,
    end: int | datetime | Timestamp,
    width: int | timedelta | Timedelta,
    depth: int = 0,
    version: Optional[int] = None,
    schema: Optional[Schema] = STAT_F32_SCHEMA,
) -> Generator[Table, None, 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

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:

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
`width` ¶	`int \| timedelta \| Timedelta`	The size of statistical windows to return, will contain summaries of the data between `start` to `end` of size `width`.	required
`depth` ¶	`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`
`version` ¶	`Optional[int]`	What version 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`

Yields:

Type	Description
`Table`	The statistical summary information of the stream as a timeseries.

raw_values ¶

raw_values(
    start: int | datetime | Timestamp,
    end: int | datetime | Timestamp,
    version: Optional[int] = None,
    schema: Optional[Schema] = TIME_VALUE_F32_SCHEMA,
) -> Table

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

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
`version` ¶	`Optional[int]`	What version 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,value pairs be returned as, default will be the server default.	`TIME_VALUE_F32_SCHEMA`

Returns:

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

raw_values_iter ¶

raw_values_iter(
    start: int | datetime | Timestamp,
    end: int | datetime | Timestamp,
    version: Optional[int] = None,
    schema: Optional[Schema] = TIME_VALUE_F32_SCHEMA,
) -> Generator[Table, None, None]

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

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:

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
`version` ¶	`Optional[int]`	What version 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,value pairs be returned as, default will be the server default.	`TIME_VALUE_F32_SCHEMA`

Yields:

Type	Description
`Table`	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 ¶

refresh_metadata() -> None

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

tags ¶

tags(
    copy: bool = True, refresh: Optional[bool] = True
) -> dict[str, str]

Return the tag-based metadata of the stream.

Parameters:

Name	Type	Description	Default
`copy` ¶	`bool`	Should the tag dictionary be copied or passed by reference, by default True	`True`
`refresh` ¶	`Optional[bool]`	Do we need to query the platform to get the metadata first, by default True	`True`

Returns:

Type	Description
`dict[str, str]`	A mapping of tag metadata key,value pairs.

update ¶

update(
    collection: Optional[str] = None,
    tags: Optional[dict[str, str]] = None,
    annotations: Optional[dict[str, str]] = None,
    replace_tags: Optional[bool] = False,
    replace_annotations: Optional[bool] = False,
)

Update the stream metadata.

Parameters:

Name	Type	Description	Default
`collection` ¶	`Optional[str]`	Change the collection the stream is located under, by default None	`None`
`tags` ¶	`Optional[dict[str, str]]`	Update any tag metadata of the stream, by default None	`None`
`annotations` ¶	`Optional[dict[str, str]]`	Update any non-tag metadata of the stream, by default None	`None`
`replace_tags` ¶	`Optional[bool]`	If you want to fully replace the current stream `tags` with the ones provided here, set to `True`, by default False	`False`
`replace_annotations` ¶	`Optional[bool]`	If you want to fully replace the current stream `annotations` metadata with the ones provided here set to `True` by default False.	`False`

windowed_values ¶

windowed_values(
    start: int | datetime | Timestamp,
    end: int | datetime | Timestamp,
    width: int | timedelta | Timedelta | _PW,
    precise: Optional[bool] = False,
    version: Optional[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`
`version` ¶	`Optional[int]`	What version 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)\).

Stream

pingthings.timeseries.client.Stream ¶

conn ¶

stream ¶

uuid ¶

known_to_exist ¶

collection ¶

tags ¶

annotations ¶

property_version ¶

Attributes¶

collection property ¶

name property ¶

uuid property ¶

Functions¶

aligned_windows_iter ¶

start ¶

end ¶

point_width ¶

width ¶

version ¶

schema ¶

annotations ¶

copy ¶

refresh ¶

count ¶

start ¶

end ¶

version ¶

precise ¶

delete ¶

start ¶

end ¶

earliest ¶

version ¶

flush ¶

get_latest_version ¶

get_version_at_time ¶

realtime ¶

insert ¶

data ¶

merge_policy ¶

latest ¶

version ¶

nearest ¶

time ¶

backward ¶

version ¶

obliterate ¶

pin_version ¶

version ¶

pinned_version ¶

precise_windows_iter ¶

start ¶

end ¶

width ¶

depth ¶

version ¶

schema ¶

raw_values ¶

start ¶

end ¶

version ¶

schema ¶

raw_values_iter ¶

start ¶

end ¶

version ¶

schema ¶

refresh_metadata ¶

tags ¶

copy ¶

refresh ¶

update ¶

collection ¶

tags ¶

annotations ¶

replace_tags ¶

replace_annotations ¶

windowed_values ¶

`conn` ¶

`stream` ¶

`uuid` ¶

`known_to_exist` ¶

`collection` ¶

`tags` ¶

`annotations` ¶

`property_version` ¶

collection `property` ¶

name `property` ¶

uuid `property` ¶

`start` ¶

`end` ¶

`point_width` ¶

`width` ¶

`version` ¶

`schema` ¶

`copy` ¶

`refresh` ¶

`start` ¶

`end` ¶

`version` ¶

`precise` ¶

`start` ¶

`end` ¶

`version` ¶

`realtime` ¶

`data` ¶

`merge_policy` ¶

`version` ¶

`time` ¶

`backward` ¶

`version` ¶

`version` ¶

`start` ¶

`end` ¶

`width` ¶

`depth` ¶

`version` ¶

`schema` ¶

`start` ¶

`end` ¶

`version` ¶

`schema` ¶

`start` ¶

`end` ¶

`version` ¶

`schema` ¶

`copy` ¶

`refresh` ¶

`collection` ¶

`tags` ¶

`annotations` ¶

`replace_tags` ¶

`replace_annotations` ¶

`start` ¶

`end` ¶

`width` ¶

`precise` ¶

`version` ¶

`schema` ¶