Fix missing logging import, rerun pre-commit

.pre-commit-config.yaml

@@ -24,7 +24,7 @@ repos:
   hooks:
   - id: flake8
     args: [--config=setup.cfg]
-    exclude: ^(btrdb/grpcinterface|tests|setup.py|btrdb4|docs)
+    exclude: ^(btrdb/grpcinterface|tests|setup.py|btrdb4|docs|benchmarks)
 - repo: local
   hooks:
     - id: pytest-check

benchmarks/benchmark_stream_reads.py

@@ -258,7 +258,9 @@ def main():
     """Run a single run of the benchmarks"""
     conn = btrdb.connect(profile="andy")
     stream1 = conn.stream_from_uuid(
-        list(conn.streams_in_collection("andy/7064-6684-5e6e-9e14-ff9ca7bae46e"))[0].uuid
+        list(conn.streams_in_collection("andy/7064-6684-5e6e-9e14-ff9ca7bae46e"))[
+            0
+        ].uuid
     )
     start = stream1.earliest()[0].time
     end = stream1.latest()[0].time
@@ -267,10 +269,17 @@ def main():
     print(f"pointwidth of: {pointwidth}")
     for f in [time_single_stream_arrow_raw_values, time_single_stream_raw_values]:
         print(f(stream1, start, end, 0))
-    for f in [time_single_stream_arrow_windows_values, time_single_stream_windows_values]:
+    for f in [
+        time_single_stream_arrow_windows_values,
+        time_single_stream_windows_values,
+    ]:
         print(f(stream1, start, end, width_ns=width_ns, version=0))
-    for f in [time_single_stream_arrow_aligned_windows_values, time_single_stream_aligned_windows_values]:
+    for f in [
+        time_single_stream_arrow_aligned_windows_values,
+        time_single_stream_aligned_windows_values,
+    ]:
         print(f(stream1, start, end, pointwidth=pointwidth, version=0))
 
-if __name__=="__main__":
-    main()
+
+if __name__ == "__main__":
+    main()

benchmarks/benchmark_streamset_reads.py

@@ -102,7 +102,11 @@ def time_streamset_arrow_raw_values(
 
 
 def time_streamset_windows_values(
-    streamset: btrdb.stream.StreamSet, start: int, end: int, width_ns: int, version: int = 0
+    streamset: btrdb.stream.StreamSet,
+    start: int,
+    end: int,
+    width_ns: int,
+    version: int = 0,
 ) -> Dict[str, Union[str, int, float]]:
     """Return the elapsed time for the streamset windows values query
 
@@ -139,7 +143,11 @@ def time_streamset_windows_values(
 
 
 def time_streamset_arrow_windows_values(
-    streamset: btrdb.stream.StreamSet, start: int, end: int, width_ns: int, version: int = 0
+    streamset: btrdb.stream.StreamSet,
+    start: int,
+    end: int,
+    width_ns: int,
+    version: int = 0,
 ) -> Dict[str, Union[str, int, float]]:
     """Return the elapsed time for the streamset arrow window values query
 
@@ -180,7 +188,11 @@ def time_streamset_arrow_windows_values(
 
 
 def time_streamset_aligned_windows_values(
-    streamset: btrdb.stream.StreamSet, start: int, end: int, pointwidth: int, version: int = 0
+    streamset: btrdb.stream.StreamSet,
+    start: int,
+    end: int,
+    pointwidth: int,
+    version: int = 0,
 ) -> Dict[str, Union[str, int, float]]:
     """Return the elapsed time for the streamset window values query
 
@@ -217,7 +229,11 @@ def time_streamset_aligned_windows_values(
 
 
 def time_streamset_arrow_aligned_windows_values(
-    streamset: btrdb.stream.StreamSet, start: int, end: int, pointwidth: int, version: int = 0
+    streamset: btrdb.stream.StreamSet,
+    start: int,
+    end: int,
+    pointwidth: int,
+    version: int = 0,
 ) -> Dict[str, Union[str, int, float]]:
     """Return the elapsed time for the streamset arrow aligned window values query
 
@@ -264,7 +280,6 @@ def _create_streamset_result_dict(
     return {
         "n_streams": len(streamset),
         "total_points": point_count,
-
         "total_time_seconds": total_time,
         "streamset_version": version,
     }
@@ -274,9 +289,15 @@ def main():
     """Run a single run of the benchmarks"""
     conn = btrdb.connect(profile="andy")
     stream1 = conn.stream_from_uuid(
-        list(conn.streams_in_collection("andy/7064-6684-5e6e-9e14-ff9ca7bae46e"))[0].uuid
+        list(conn.streams_in_collection("andy/7064-6684-5e6e-9e14-ff9ca7bae46e"))[
+            0
+        ].uuid
+    )
+    stream2 = conn.stream_from_uuid(
+        list(conn.streams_in_collection("andy/30e6-d72f-5cc7-9966-bc1579dc4a72"))[
+            0
+        ].uuid
     )
-    stream2 = conn.stream_from_uuid(list(conn.streams_in_collection("andy/30e6-d72f-5cc7-9966-bc1579dc4a72"))[0].uuid)
     streamset = btrdb.stream.StreamSet([stream1, stream2])
     start = max(stream1.earliest()[0].time, stream2.earliest()[0].time)
     end = min(stream1.latest()[0].time, stream2.latest()[0].time)
@@ -286,7 +307,7 @@ def main():
     res_list = []
     for f in [time_streamset_raw_values, time_streamset_arrow_raw_values]:
         res = f(streamset, start, end, 0)
-        res["func"] =  f.__name__
+        res["func"] = f.__name__
     # for f in [time_streamset_windows_values, time_streamset_arrow_windows_values]:
     #     res = f(streamset, start, end, width_ns=width_ns, version=0)
     #     res["func"] = f.__name__
@@ -295,6 +316,8 @@ def main():
     #     res["func"] = res
 
     return res
-if __name__=="__main__":
+
+
+if __name__ == "__main__":
     results = main()
-    print(pandas.DataFrame(results))
+    print(pandas.DataFrame(results))

btrdb/conn.py

@@ -16,6 +16,7 @@
 ##########################################################################
 
 import json
+import logging
 import os
 import re
 import uuid as uuidlib
@@ -113,6 +114,7 @@ def __init__(self, addrportstr, apikey=None):
                 )
             self.channel = grpc.insecure_channel(addrportstr, chan_ops)
 
+
 def _is_arrow_enabled(info):
     info = {
         "majorVersion": info.majorVersion,
@@ -127,6 +129,7 @@ def _is_arrow_enabled(info):
     else:
         return False
 
+
 class BTrDB(object):
     """
     The primary server connection object for communicating with a BTrDB server.
@@ -135,7 +138,7 @@ class BTrDB(object):
     def __init__(self, endpoint):
         self.ep = endpoint
         self._executor = ThreadPoolExecutor()
-        self._ARROW_ENABLED = True #_is_arrow_enabled(self.ep.info())
+        self._ARROW_ENABLED = True  # _is_arrow_enabled(self.ep.info())
         logger.debug(f"ARROW ENABLED: {self._ARROW_ENABLED}")
 
     def query(self, stmt, params=[]):

btrdb/endpoint.py

@@ -65,7 +65,7 @@ def arrowMultiValues(self, uu_list, start, end, version_list, snap_periodNS):
             start=start,
             end=end,
             versionMajor=[ver for ver in version_list],
-            snapPeriodNs=int(snap_periodNS)
+            snapPeriodNs=int(snap_periodNS),
         )
         for result in self.stub.ArrowMultiValues(params):
             check_proto_stat(result.stat)

btrdb/stream.py

@@ -24,7 +24,6 @@
 
 import pyarrow as pa
 
-import btrdb.grpcinterface.btrdb_pb2_grpc
 from btrdb.exceptions import (
     BTrDBError,
     BTRDBTypeError,
@@ -1169,7 +1168,7 @@ def versions(self):
             self._pinned_versions if self._pinned_versions else self._latest_versions()
         )
 
-    def count(self, precise:bool=False):
+    def count(self, precise: bool = False):
         """
         Compute the total number of points in the streams using filters.
 
@@ -1196,10 +1195,14 @@ def count(self, precise:bool=False):
         start = params.get("start", MINIMUM_TIME)
         end = params.get("end", MAXIMUM_TIME)
 
-        versions = self._pinned_versions if self._pinned_versions else self._latest_versions()
+        versions = (
+            self._pinned_versions if self._pinned_versions else self._latest_versions()
+        )
 
         my_counts_gen = self._btrdb._executor.map(
-            lambda s: s.count(start, end, version=versions.get(s.uuid, 0), precise=precise),
+            lambda s: s.count(
+                start, end, version=versions.get(s.uuid, 0), precise=precise
+            ),
             self._streams,
         )
 
@@ -1655,7 +1658,7 @@ def _arrow_streamset_data(self):
                     )
                 data = main_table
         return data
-  
+
     def rows(self):
         """
         Returns a materialized list of tuples where each tuple contains the
@@ -1703,7 +1706,7 @@ def rows(self):
     def arrow_rows(self):
         """Return tuples of rows from arrow table"""
         raise NotImplementedError(
-            f"arrow_rows has not been implemented yet, please use `rows` if you need this functionality."
+            "arrow_rows has not been implemented yet, please use `rows` if you need this functionality."
         )
 
     def insert(self, data_map: dict, merge: str = "never") -> dict:

btrdb/transformers.py

@@ -22,7 +22,6 @@
 
 import pandas as pd
 
-
 ##########################################################################
 ## Helper Functions
 ##########################################################################
@@ -119,7 +118,7 @@ def arrow_to_series(streamset, agg="mean", name_callable=None):
     """
     if not streamset._btrdb._ARROW_ENABLED:
         raise NotImplementedError(
-            f"arrow_to_series requires an arrow-enabled BTrDB server."
+            "arrow_to_series requires an arrow-enabled BTrDB server."
         )
     arrow_df = arrow_to_dataframe(
         streamset=streamset, agg=agg, name_callable=name_callable
@@ -150,7 +149,7 @@ def arrow_to_dataframe(
     """
     if not streamset._btrdb._ARROW_ENABLED:
         raise NotImplementedError(
-            f"arrow_to_dataframe requires an arrow-enabled BTrDB server."
+            "arrow_to_dataframe requires an arrow-enabled BTrDB server."
         )
 
     try:
@@ -182,7 +181,6 @@ def arrow_to_dataframe(
     if not callable(name_callable):
         name_callable = lambda s: s.collection + "/" + s.name
     tmp_table = streamset.arrow_values()
-    my_cols = [c for c in tmp_table.column_names]
     col_names = _stream_names(streamset, name_callable)
     cols = []
     for name in col_names:
@@ -299,7 +297,7 @@ def arrow_to_polars(streamset, agg="mean", name_callable=None):
     """
     if not streamset._btrdb._ARROW_ENABLED:
         raise NotImplementedError(
-            f"arrow_to_polars requires an arrow-enabled BTrDB server."
+            "arrow_to_polars requires an arrow-enabled BTrDB server."
         )
     try:
         import polars as pl
@@ -314,7 +312,7 @@ def arrow_to_polars(streamset, agg="mean", name_callable=None):
 def arrow_to_arrow_table(streamset):
     if not streamset._btrdb._ARROW_ENABLED:
         raise NotImplementedError(
-            f"arrow_to_arrow_table requires an arrow-enabled BTrDB server."
+            "arrow_to_arrow_table requires an arrow-enabled BTrDB server."
         )
     return streamset.arrow_values()
 
@@ -427,7 +425,7 @@ def arrow_to_numpy(streamset, agg="mean"):
     """
     if not streamset._btrdb._ARROW_ENABLED:
         raise NotImplementedError(
-            f"arrow_to_numpy requires an arrow-enabled BTrDB server."
+            "arrow_to_numpy requires an arrow-enabled BTrDB server."
         )
     arrow_df = arrow_to_dataframe(streamset=streamset, agg=agg, name_callable=None)
     return arrow_df.values
@@ -496,7 +494,7 @@ def arrow_to_dict(streamset, agg="mean", name_callable=None):
     """
     if not streamset._btrdb._ARROW_ENABLED:
         raise NotImplementedError(
-            f"arrow_to_dict requires an arrow-enabled BTrDB server."
+            "arrow_to_dict requires an arrow-enabled BTrDB server."
         )
     arrow_df = arrow_to_dataframe(
         streamset=streamset, agg=agg, name_callable=name_callable

tests/btrdb/test_arrow_streams.py

@@ -11,7 +11,7 @@
 
 @pytest.fixture
 def stream1():
-    uu = uuid.UUID('0d22a53b-e2ef-4e0a-ab89-b2d48fb2592a')
+    uu = uuid.UUID("0d22a53b-e2ef-4e0a-ab89-b2d48fb2592a")
     stream = Mock(Stream)
     stream.version = Mock(return_value=11)
     stream.uuid = Mock(return_value=uu)
@@ -28,7 +28,7 @@ def stream1():
 
 @pytest.fixture
 def stream2():
-    uu = uuid.UUID('17dbe387-89ea-42b6-864b-f505cdb483f5')
+    uu = uuid.UUID("17dbe387-89ea-42b6-864b-f505cdb483f5")
     stream = Mock(Stream)
     stream.version = Mock(return_value=22)
     stream.uuid = Mock(return_value=uu)
@@ -42,5 +42,6 @@ def stream2():
     stream._btrdb._ARROW_ENABLED = Mock(return_value=True)
     return stream
 
+
 class TestArrowStreams(object):
-    pass
+    pass

tests/btrdb/test_stream.py

@@ -11,11 +11,6 @@
 Testing package for the btrdb stream module
 """
 import concurrent.futures
-
-##########################################################################
-## Imports
-##########################################################################
-
 import datetime
 import json
 import re
@@ -35,13 +30,16 @@
     InvalidOperation,
     NoSuchPoint,
     StreamNotFoundError,
-    InvalidCollection,
-    NoSuchPoint,
 )
 from btrdb.grpcinterface import btrdb_pb2
 from btrdb.point import RawPoint, StatPoint
 from btrdb.stream import INSERT_BATCH_SIZE, Stream, StreamFilter, StreamSet
 
+##########################################################################
+## Imports
+##########################################################################
+
+
 RawPointProto = btrdb_pb2.RawPoint
 StatPointProto = btrdb_pb2.StatPoint
 EST = pytz.timezone("America/New_York")
@@ -85,6 +83,7 @@ def stream2():
     stream._btrdb._ARROW_ENABLED = Mock(return_value=False)
     return stream
 
+
 @pytest.fixture
 def arrow_stream3():
     uu = uuid.UUID("17dbe387-89ea-42b6-864b-f505cdb483f5")
@@ -101,6 +100,7 @@ def arrow_stream3():
     stream._btrdb._ARROW_ENABLED = Mock(return_value=True)
     return stream
 
+
 ##########################################################################
 ## Stream Tests
 ##########################################################################
@@ -1195,6 +1195,7 @@ def test_earliest(self, stream1, stream2):
             RawPoint(time=10, value=1),
             RawPoint(time=20, value=1),
         )
+
     def test_latest(self, stream1, stream2):
         """
         Assert latest returns correct time code