diff --git a/btrdbextras/ds_utils/__init__.py b/btrdbextras/ds_utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/btrdbextras/ds_utils/stream_info.py b/btrdbextras/ds_utils/stream_info.py new file mode 100644 index 0000000..ddac98f --- /dev/null +++ b/btrdbextras/ds_utils/stream_info.py @@ -0,0 +1,303 @@ +# btrdbextras.ds_utils.stream_info +# Module for general utility functions for PingThings DataSci Team. +# +# Author: PingThings +# Created: Fri Dec 1 17:48:23 2023 -0800 +# +# For license information, see LICENSE.txt +# ID: stream_info.py [] jeff.lin@pingthings.io + +""" +Module for general utility functions for checking stream information by PingThings DataSci Team. +""" +########################################################################## +# Imports +########################################################################## +import warnings +from typing import List, Union + +import numpy as np +import pandas as pd + +from btrdb.stream import Stream +from btrdb.utils.timez import ns_delta + +__all__ = [ + "describe_streams", + "find_samplerate", + "StreamType", + "ANGLE", + "CONDUCTANCE", + "CURRENT", + "DQ", + "FREQUENCY", + "POWER", + "RESISTANCE", + "VOLTAGE", + ] + +########################################################################## +# Constants +########################################################################## +KNOWN_HZ = np.array([1 / (15 * 60), 15, 30, 60, 120]) + +# Unit types +ANGLE = "angle" +CONDUCTANCE = "conductance" +CURRENT = "current" +DQ = "data quality" +FREQUENCY = "frequency" +POWER = "power" +RESISTANCE = "resistance" +VOLTAGE = "voltage" + +# StreamTypeMap categorize streams based on provided units +StreamTypeMap = { + "": (None, 1), + "A": (CURRENT, 1), + "Amp": (CURRENT, 1), + "Amps": (CURRENT, 1), + "amps": (CURRENT, 1), + "arcmin": (ANGLE, 1 / 60), + "CurrentAcInternalShunt": (CURRENT, 1), + "DC/V": (VOLTAGE, 1), + "Deg": (ANGLE, 1), + "deg": (ANGLE, 1), + "Degree": (ANGLE, 1), + "Degrees": (ANGLE, 1), + "degrees": (ANGLE, 1), + "FLAG": (DQ, 1), + "FREQ": (FREQUENCY, 1), + "Freq": (FREQUENCY, 1), + "Hz": (FREQUENCY, 1), + "IPHA": (ANGLE, 1), + "IPHM": (CURRENT, 1), + "KA": (CURRENT, 1e3), + "kA": (CURRENT, 1e3), + "kAmps": (CURRENT, 1e3), + "kilovolts": (VOLTAGE, 1e3), + "kOhms": (RESISTANCE, 1e3), + "KV": (VOLTAGE, 1e3), + "kV": (VOLTAGE, 1e3), + "kVolt": (VOLTAGE, 1e3), + "kVolts": (VOLTAGE, 1e3), + "millivolts": (VOLTAGE, 1e-3), + "MVAR": (POWER, 1e6), + "MVar": (POWER, 1e6), + "MVARS": (POWER, 1e6), + "MVr": (VOLTAGE, 1e6), + "MW": (POWER, 1e6), + "Ohms": (RESISTANCE, 1), + "radian": (ANGLE, 1), + "Radian": (ANGLE, 1), + "radians": (ANGLE, 1), + "Radians": (ANGLE, 1), + "rpm": (FREQUENCY, 1), + "Siemens": (CONDUCTANCE, 1), + "STAT": (DQ, 1), + "STAcAmp": (CURRENT, 1), + "STActivePower": (POWER, 1), + "STAcVolt": (VOLTAGE, 1), + "STApparentPower": (POWER, 1), + "STDcAmp": (CURRENT, 1), + "STDcVolt": (VOLTAGE, 1), + "STReactivePower": (POWER, 1), + "V": (VOLTAGE, 1), + "v": (VOLTAGE, 1), + "VAR": (POWER, 1), + "VA": (POWER, 1), + "Vars": (POWER, 1), + "Volt": (VOLTAGE, 1), + "VoltageAc": (VOLTAGE, 1), + "Volts": (VOLTAGE, 1), + "volts": (VOLTAGE, 1), + "VPHA": (ANGLE, 1), + "VPHM": (VOLTAGE, 1), + "Vr": (VOLTAGE, 1), + "W": (POWER, 1), + "Watts": (POWER, 1), + # "%": (TODO, 1), # TODO: fill in missing value from Dominion + # "ALOG": (TODO, 1), # TODO: fill in missing value from Dominion + # "Analog": (TODO, 1), # TODO: fill in missing value from Dominion + # "DIGI": (TODO, 1), # TODO: fill in missing value from Dominion + # "Digital": (TODO, 1), # TODO: fill in missing value from Dominion + # "digital": (TODO, 1), # TODO: fill in missing value from Dominion + # "Phasor": (TODO, 1), # TODO: fill in missing value from Dominion + # "PU": (TODO, 1), # TODO: fill in missing value from Dominion + # "QUAL": (TODO, 1), # TODO: fill in missing value from Dominion + } + + +########################################################################## +# General utilities +########################################################################## +def describe_streams( + streams: List[Stream], display_annotations=False, filter_annotations=None + ): + """ + This function displays streams info such as collection, UUID, tags and annotations in Dataframe format. + + Parameters + ---------- + streams : list of Stream objects or Streamset. + Streams to display streams info. + display_annotations: bool, optional. + Whether to display annotations. Default is False. + filter_annotations : [str], optional. + Only display the specified annotations along with the tags, collection and UUID. + + Returns + ------- + pandas.DataFrame + DataFrame displaying streams info. + + Examples + ---------- + >>> streams = conn.streams_in_collection('sunshine/PMU3') + >>> describe_streams(streams) + """ + + table = [] + for idx, stream in enumerate(streams): + stream_info_dict = {"collection": stream.collection, "UUID": str(stream.uuid)} + stream_info_dict.update(stream.tags()) + + if display_annotations: + annotations = stream.annotations()[0] + stream_info_dict.update(annotations) + + table.append(stream_info_dict) + + table_df = pd.DataFrame(table) + + if filter_annotations is not None: + if not isinstance(filter_annotations, list) or not all( + isinstance(annotation, str) for annotation in filter_annotations + ): + raise TypeError("filter_annotations has to be a list of str.") + + # find filter_annotations that are not in "annotations", this might break if python version is below 3.7 + # since dictionaries are not ordered for python version lower than 3.7 + not_found_annotations = [ + annotation + for annotation in filter_annotations + if annotation not in table_df.columns[6:] + ] + if len(not_found_annotations) > 0: + raise ValueError(f"{not_found_annotations} not found in annotations.") + + table_df = table_df[ + ["collection", "UUID", "name", "unit", "distiller", "ingress"] + + filter_annotations + ] + + return table_df + + +def find_samplerate(stream, pw=50, update=False, version=0): + """ + Find the sample rate of the stream. + + Parameters + ---------- + stream : Stream + BTrDB Stream object + pw : int + pointwidth (default is 50) approximately 13 days, creating ~28 time points in a + year to calculate sampling rate from. + update: bool + (default - False) whether to update the stream annotation with the sample rate + version: int + specify which version to use + + Returns + ------- + Fs : int + Calculated sampling rate of the stream. + """ + + def _find_sampling(stream, s, e, v): + d = [_[0].time for _ in stream.values(s, e, v)] + if len(d) in {0, 1}: + return + temp = pd.Series(d, name="time") + + # find the median time difference from timestamps convert nanoseconds to seconds + fs = round((temp.diff().median() / 1e6), 2) / 1.0e3 + if fs == 0: + return + else: + return 1 / fs + + start = stream.earliest()[0].time + end = stream.latest()[0].time + n_range = ns_delta(minutes=60) + # find sample rates from 60 mins of every 2nd time window of the aligned_windows (if + # there's more than 10 windows) + windows = stream.aligned_windows( + start, end, pw, version=version + ) # returns list of tuples + if len(windows) > 10: + windows = windows[1:-2:2] + warnings.warn( + f" {stream.uuid.hex}: calculating sample rate from every other window of" + f"pointwidht = {pw})" + ) + Fs = [ + _find_sampling(stream, _[0].time, _[0].time + n_range, version) for _ in windows + ] + Fs = np.mean([_ for _ in Fs if _]) + if np.isnan(Fs): + warnings.warn(f" {stream.uuid.hex} has no data to calculate sampling rate from") + else: + Fs = float(KNOWN_HZ[np.abs(KNOWN_HZ - Fs).argmin()]) + if update: + stream.update(annotations={"sample_rate": Fs}) + return Fs + + +# StreamType from Unit +class StreamType: + """ + The StreamType class is used to determine a stream's tagged unit to their corresponding general + unit types, such as Voltage, Current, etc and scale factors to base unit of measurements. + + Examples + -------- + + Create a stream type object for voltage with a scale factor of 1 and print the unit type and + scale factor + + >>> voltage_streamtype = StreamType.from_unit("VPHM") + >>> print(voltage_streamtype.unit_type, voltage_streamtype.scalefactor) + (Voltage, 1) + + Create a stream type object for current with a scale factor of 1000 + + >>> current_streamtype = StreamType.from_unit("kA") + >>> print(current_streamtype.unit_type, current_streamtype.scalefactor) + CURRENT, 1000 + + Can check each stream's unit type to be used in per-unit calculations + + >>> from btrdbextras.ds_utils.stream_info import VOLTAGE, CURRENT + >>> tagged_unit = stream.unit() # Stream's unit is 'VPHM' + >>> if StreamType.from_unit(tagged_unit) == VOLTAGE: + >>> # calculation for per-unit with basekV value + >>> pass + >>> else: + >>> # calculation for per-unit with basePower value + >>> pass + + """ + + def __init__(self, unit_type: str, scalefactor: Union[int, float]): + self.unit_type = unit_type + self.scalefactor = scalefactor + + def __repr__(self): + return f"<{self.unit_type.title()}: scale-factor={self.scalefactor:.0f}>" + + @classmethod + def from_unit(cls, stream_unit: str): + return cls(*StreamTypeMap[stream_unit]) diff --git a/tests/test_ds_utils/test_stream_info.py b/tests/test_ds_utils/test_stream_info.py new file mode 100644 index 0000000..617fc8d --- /dev/null +++ b/tests/test_ds_utils/test_stream_info.py @@ -0,0 +1,260 @@ +import uuid +from unittest.mock import Mock, PropertyMock + +import pandas as pd +import pytest +from ds_utils.ds_api.stream_info import ANGLE, describe_streams, StreamType, VOLTAGE +from pandas.testing import assert_frame_equal + +from btrdb import BTrDB +from btrdb.stream import Stream, StreamSet + + +@pytest.fixture(scope="session") +def mock_stream1(): + uu = uuid.UUID("0d22a53b-e2ef-4e0a-ab89-b2d48fb25921") + stream = Mock(Stream) + type(stream)._btrdb = PropertyMock(return_value=Mock(BTrDB)) + type(stream).uuid = PropertyMock(return_value=uu) + type(stream).collection = PropertyMock(return_value="animals/dog") + type(stream).name = PropertyMock(return_value="bob") + type(stream).unit = PropertyMock(return_value="puppy") + + stream.annotations = Mock( + return_value=({"color": "red", "animals": "dog", "breed": "corgi"}, 11) + ) + stream.tags = Mock( + return_value={"name": "bob", "unit": "puppy", "distiller": "", "ingress": ""} + ) + + # stream.windows = Mock(return_value=[()]) + + return stream + + +@pytest.fixture(scope="session") +def mock_stream2(): + uu = uuid.UUID("0d22a53b-e2ef-4e0a-ab89-b2d48fb25922") + stream = Mock(Stream) + type(stream)._btrdb = PropertyMock(return_value=Mock(BTrDB)) + type(stream).uuid = PropertyMock(return_value=uu) + type(stream).collection = PropertyMock(return_value="animals/dog") + type(stream).name = PropertyMock(return_value="peter") + type(stream).unit = PropertyMock(return_value="puppy") + + stream.annotations = Mock( + return_value=({"color": "green", "animals": "dog", "breed": "pug"}, 11) + ) + stream.tags = Mock( + return_value={"name": "peter", "unit": "puppy", "distiller": "", "ingress": ""} + ) + + stream.values = Mock(return_value=[()]) + + return stream + + +class TestStreamInfo: + def test_describe_streams_return_tags_only(self, mock_stream1, mock_stream2): + """ + Assert describe_streams() returns all the tags along with UUID and collection. + """ + streamset = StreamSet([mock_stream1, mock_stream2]) + stream1_dict = { + "collection": "animals/dog", + "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25921", + "name": "bob", + "unit": "puppy", + "distiller": "", + "ingress": "", + } + + stream2_dict = { + "collection": "animals/dog", + "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25922", + "name": "peter", + "unit": "puppy", + "distiller": "", + "ingress": "", + } + + assert_frame_equal( + describe_streams(streamset), pd.DataFrame([stream1_dict, stream2_dict]) + ) + + def test_describe_streams_return_all_info(self, mock_stream1, mock_stream2): + """ + Assert describe_streams() returns all the tags and annotations along with UUID and collection. + """ + streamset = StreamSet([mock_stream1, mock_stream2]) + stream1_dict = { + "collection": "animals/dog", + "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25921", + "name": "bob", + "unit": "puppy", + "distiller": "", + "ingress": "", + "color": "red", + "animals": "dog", + "breed": "corgi", + } + + stream2_dict = { + "collection": "animals/dog", + "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25922", + "name": "peter", + "unit": "puppy", + "distiller": "", + "ingress": "", + "color": "green", + "animals": "dog", + "breed": "pug", + } + + assert_frame_equal( + describe_streams(streamset, display_annotations=True), + pd.DataFrame([stream1_dict, stream2_dict]), + ) + + def test_describe_streams_return_filter_annotations( + self, mock_stream1, mock_stream2 + ): + """ + Assert describe_streams() returns all the tags and specified annotations along with UUID and collection. + """ + streamset = StreamSet([mock_stream1, mock_stream2]) + stream1_dict = { + "collection": "animals/dog", + "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25921", + "name": "bob", + "unit": "puppy", + "distiller": "", + "ingress": "", + "color": "red", + } + + stream2_dict = { + "collection": "animals/dog", + "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25922", + "name": "peter", + "unit": "puppy", + "distiller": "", + "ingress": "", + "color": "green", + } + assert_frame_equal( + describe_streams( + streamset, display_annotations=True, filter_annotations=["color"] + ), + pd.DataFrame([stream1_dict, stream2_dict]), + ) + + def test_describe_streams_return_filter_annotations_nonexist( + self, mock_stream1, mock_stream2 + ): + """ + Assert describe_streams() returns all the tags and specified annotations along with UUID and collection. + """ + streamset = StreamSet([mock_stream1, mock_stream2]) + # stream1_dict = { + # "collection": "animals/dog", + # "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25921", + # "name": "bob", + # "unit": "puppy", + # "distiller": "", + # "ingress": "", + # "color": "red", + # } + # + # stream2_dict = { + # "collection": "animals/dog", + # "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25922", + # "name": "peter", + # "unit": "puppy", + # "distiller": "", + # "ingress": "", + # "color": "green", + # } + + with pytest.raises( + ValueError, match=r"(?<=\[)(.*?)(?=] not found in annotations)" + ): + assert describe_streams( + streamset, display_annotations=True, filter_annotations=["address"] + ) + + def test_describe_streams_return_filter_annotations_wrong_type_str( + self, mock_stream1, mock_stream2 + ): + """ + Assert describe_streams() throws an exception when the input type is wrong. + """ + streamset = StreamSet([mock_stream1, mock_stream2]) + # stream1_dict = { + # "collection": "animals/dog", + # "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25921", + # "name": "bob", + # "unit": "puppy", + # "distiller": "", + # "ingress": "", + # "color": "red", + # } + # + # stream2_dict = { + # "collection": "animals/dog", + # "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25922", + # "name": "peter", + # "unit": "puppy", + # "distiller": "", + # "ingress": "", + # "color": "green", + # } + + with pytest.raises( + TypeError, match="filter_annotations has to be a list of str." + ): + assert describe_streams( + streamset, display_annotations=True, filter_annotations="location" + ) + + def test_describe_streams_return_filter_annotations_wrong_type_int_list( + self, mock_stream1, mock_stream2 + ): + """ + Assert describe_streams() throws an exception when the input type is wrong. + """ + streamset = StreamSet([mock_stream1, mock_stream2]) + # stream1_dict = { + # "collection": "animals/dog", + # "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25921", + # "name": "bob", + # "unit": "puppy", + # "distiller": "", + # "ingress": "", + # "color": "red", + # } + # + # stream2_dict = { + # "collection": "animals/dog", + # "UUID": "0d22a53b-e2ef-4e0a-ab89-b2d48fb25922", + # "name": "peter", + # "unit": "puppy", + # "distiller": "", + # "ingress": "", + # "color": "green", + # } + + with pytest.raises( + TypeError, match="filter_annotations has to be a list of str." + ): + assert describe_streams( + streamset, display_annotations=True, filter_annotations=[1, 2, 3] + ) + + def test_streamtype_from_unit(self): + truths = {"VPHM": StreamType(VOLTAGE, 1), "VPHA": StreamType(ANGLE, 1)} + for unit in truths.keys(): + assert truths[unit].unit_type == StreamType.from_unit(unit).unit_type + + def test_find_samplerate(self, mock_stream1): + assert False