Refactor perspective utilities into common module and add adapter test

Signed-off-by: Pascal Tomecek <[email protected]>

Author: ptomecek

.github/workflows/build.yml

@@ -672,6 +672,7 @@ jobs:
           package:
             - "sqlalchemy<2"
             - "numpy==1.19.5"
+            - "pandas<1.5"  # Any later does not support numpy==1.19.5
             - "perspective-python<3"
 
       runs-on: ${{ matrix.os }}

conda/dev-environment-unix.yml

@@ -24,7 +24,7 @@ dependencies:
  - librdkafka
  - lz4-c
  - mamba
- - mdformat>=0.7.17,<0.8
+ - mdformat=0.7.17
  - ninja
  - numpy<2
  - pandas

conda/dev-environment-win.yml

@@ -23,7 +23,7 @@ dependencies:
  - lz4-c
  - make
  - mamba
- - mdformat>=0.7.17,<0.8
+ - mdformat=0.7.17
  - ninja
  - numpy<2
  - pandas

csp/adapters/perspective.py

@@ -1,10 +1,16 @@
 import threading
 from datetime import timedelta
-from packaging import version
+from perspective import Table as Table_, View as View_
 from typing import Dict, Optional, Union
 
 import csp
 from csp import ts
+from csp.impl.perspective_common import (
+    date_to_perspective,
+    datetime_to_perspective,
+    is_perspective3,
+    perspective_type_map,
+)
 from csp.impl.wiring.delayed_node import DelayedNodeWrapperDef
 
 try:
@@ -14,19 +20,12 @@
 except ImportError:
     raise ImportError("perspective adapter requires tornado package")
 
-try:
-    from perspective import Server, Table as Table_, View as View_, __version__, set_threadpool_size
-
-    if version.parse(__version__) >= version.parse("3"):
-        _PERSPECTIVE_3 = True
-    elif version.parse(__version__) >= version.parse("0.6.2"):
-        from perspective import PerspectiveManager
 
-        _PERSPECTIVE_3 = False
-    else:
-        raise ImportError("perspective adapter requires 0.6.2 or greater of the perspective-python package")
-except ImportError:
-    raise ImportError("perspective adapter requires 0.6.2 or greater of the perspective-python package")
+_PERSPECTIVE_3 = is_perspective3()
+if _PERSPECTIVE_3:
+    from perspective import Server
+else:
+    from perspective import PerspectiveManager
 
 
 # Run perspective update in a separate tornado loop
@@ -43,12 +42,25 @@ def _apply_updates(table: object, data: {str: ts[object]}, throttle: timedelta):
 
     with csp.state():
         s_buffer = []
+        s_datetime_cols = set()
+        s_date_cols = set()
 
     with csp.start():
         csp.schedule_alarm(alarm, throttle, True)
+        if _PERSPECTIVE_3:
+            s_datetime_cols = set([c for c, t in table.schema().items() if t == "datetime"])
+            s_date_cols = set([c for c, t in table.schema().items() if t == "date"])
 
     if csp.ticked(data):
-        s_buffer.append(dict(data.tickeditems()))
+        row = dict(data.tickeditems())
+        if _PERSPECTIVE_3:
+            for col, value in row.items():
+                if col in s_datetime_cols:
+                    row[col] = datetime_to_perspective(row[col])
+                if col in s_date_cols:
+                    row[col] = date_to_perspective(row[col])
+
+        s_buffer.append(row)
 
     if csp.ticked(alarm):
         if len(s_buffer) > 0:
@@ -66,11 +78,13 @@ def _launch_application(port: int, server: object, stub: ts[object]):
         s_iothread = None
 
     with csp.start():
-        from perspective import PerspectiveTornadoHandler
-
         if _PERSPECTIVE_3:
+            from perspective.handlers.tornado import PerspectiveTornadoHandler
+
             handler_args = {"perspective_server": server, "check_origin": True}
         else:
+            from perspective import PerspectiveTornadoHandler
+
             handler_args = {"manager": server, "check_origin": True}
         s_app = tornado.web.Application(
             [
@@ -205,12 +219,14 @@ def create_table(self, name, limit=None, index=None):
         return table
 
     def _instantiate(self):
-        set_threadpool_size(self._threadpool_size)
         if _PERSPECTIVE_3:
             server = Server()
             client = server.new_local_client()
             thread = threading.Thread(target=perspective_thread, kwargs=dict(client=client))
         else:
+            from perspective import set_threadpool_size
+
+            set_threadpool_size(self._threadpool_size)
             manager = PerspectiveManager()
             thread = threading.Thread(target=perspective_thread, kwargs=dict(manager=manager))
         thread.daemon = True
@@ -221,10 +237,12 @@ def _instantiate(self):
                 k: v.tstype.typ if not issubclass(v.tstype.typ, csp.Enum) else str for k, v in table.columns.items()
             }
             if _PERSPECTIVE_3:
+                psp_type_map = perspective_type_map()
+                schema = {col: psp_type_map.get(typ, typ) for col, typ in schema.items()}
+                ptable = client.table(schema, name=table_name, limit=table.limit, index=table.index)
+            else:
                 ptable = Table(schema, limit=table.limit, index=table.index)
                 manager.host_table(table_name, ptable)
-            else:
-                ptable = client.table(schema, name=table_name, limit=table.limit, index=table.index)
 
             _apply_updates(ptable, table.columns, self._throttle)
 

csp/impl/pandas_perspective.py

@@ -1,29 +1,24 @@
 import pandas as pd
 import pytz
-from datetime import date, datetime, timedelta
-from packaging import version
+from datetime import datetime, timedelta
 from pandas.compat import set_function_name
 from typing import Optional
 
 import csp
 import csp.impl.pandas_accessor  # To ensure that the csp accessors are registered
 from csp.impl.pandas_ext_type import is_csp_type
+from csp.impl.perspective_common import (
+    PerspectiveWidget,
+    date_to_perspective,
+    datetime_to_perspective,
+    is_perspective3,
+    perspective,
+    perspective_type_map,
+)
 
 _ = csp.impl.pandas_accessor
 
-try:
-    import perspective
-
-    if version.parse(perspective.__version__) >= version.parse("3"):
-        _PERSPECTIVE_3 = True
-        from perspective.widget import PerspectiveWidget
-    else:
-        _PERSPECTIVE_3 = False
-        from perspective import PerspectiveWidget
-except ImportError:
-    raise ImportError(
-        "perspective must be installed to use this module. " "To install, run 'pip install perspective-python'."
-    )
+_PERSPECTIVE_3 = is_perspective3()
 
 
 @csp.node
@@ -43,6 +38,7 @@ def _apply_updates(
         s_buffer = []
         s_has_time_col = False
         s_datetime_cols = set()
+        s_date_cols = set()
 
     with csp.start():
         if throttle > timedelta(0):
@@ -67,26 +63,25 @@ def _apply_updates(
                 if index_col:
                     row[index_col] = idx
                 if s_has_time_col:
-                    if localize or _PERSPECTIVE_3:
+                    if localize:
                         row[time_col] = pytz.utc.localize(csp.now())
                     else:
                         row[time_col] = csp.now()
                     if _PERSPECTIVE_3:
-                        row[time_col] = int(row[time_col].timestamp() * 1000)
+                        row[time_col] = datetime_to_perspective(row[time_col])
             else:
                 row = new_rows[idx]
 
-            if (localize or _PERSPECTIVE_3) and col in s_datetime_cols and value.tzinfo is None:
+            if localize and col in s_datetime_cols and value.tzinfo is None:
                 row[col] = pytz.utc.localize(value)
             else:
                 row[col] = value
 
             if _PERSPECTIVE_3:
                 if col in s_datetime_cols:
-                    row[col] = int(row[col].timestamp() * 1000)
+                    row[col] = datetime_to_perspective(row[col])
                 if col in s_date_cols:
-                    d = row[col]
-                    row[col] = int(datetime(year=d.year, month=d.month, day=d.day, tzinfo=pytz.UTC).timestamp() * 1000)
+                    row[col] = date_to_perspective(row[col])
 
         if static_records:
             for idx, row in new_rows.items():
@@ -181,15 +176,8 @@ def __init__(
             else:
                 schema[col] = static_schema[col]
         if _PERSPECTIVE_3:
-            perspective_type_map = {
-                str: "string",
-                float: "float",
-                int: "integer",
-                date: "date",
-                datetime: "datetime",
-                bool: "boolean",
-            }
-            schema = {col: perspective_type_map.get(typ, typ) for col, typ in schema.items()}
+            psp_type_map = perspective_type_map()
+            schema = {col: psp_type_map.get(typ, typ) for col, typ in schema.items()}
 
         if self._keep_history:
             if _PERSPECTIVE_3:

csp/impl/perspective_common.py

@@ -0,0 +1,51 @@
+import pytz
+from datetime import date, datetime
+from packaging import version
+
+try:
+    import perspective
+
+    if version.parse(perspective.__version__) >= version.parse("3"):
+        _PERSPECTIVE_3 = True
+        from perspective.widget import PerspectiveWidget
+
+    elif version.parse(perspective.__version__) >= version.parse("0.6.2"):
+        from perspective import PerspectiveManager, PerspectiveWidget  # noqa F401
+
+        _PERSPECTIVE_3 = False
+    else:
+        raise ImportError("perspective adapter requires 0.6.2 or greater of the perspective-python package")
+
+except ImportError:
+    raise ImportError(
+        "perspective must be installed to use this module. " "To install, run 'pip install perspective-python'."
+    )
+
+
+def is_perspective3():
+    """Whether the perspective version is >= 3"""
+    return _PERSPECTIVE_3
+
+
+def perspective_type_map():
+    """Return the mapping of standard python types to perspective types"""
+    return {
+        str: "string",
+        float: "float",
+        int: "integer",
+        date: "date",
+        datetime: "datetime",
+        bool: "boolean",
+    }
+
+
+def datetime_to_perspective(dt: datetime) -> int:
+    """Convert a python datetime to an integer number of milliseconds for perspective >= 3"""
+    if dt.tzinfo is None:
+        dt = pytz.utc.localize(dt)
+    return int(dt.timestamp() * 1000)
+
+
+def date_to_perspective(d: date) -> int:
+    """Convert a python date to an integer number of milliseconds for perspective >= 3"""
+    return int(datetime(year=d.year, month=d.month, day=d.day, tzinfo=pytz.UTC).timestamp() * 1000)

csp/tests/adapters/test_perspective.py

@@ -0,0 +1,35 @@
+import unittest
+from datetime import date, datetime, timedelta
+
+import csp
+
+try:
+    from csp.adapters.perspective import PerspectiveAdapter
+    from csp.impl.pandas_perspective import CspPerspectiveMultiTable, CspPerspectiveTable
+    from csp.impl.perspective_common import PerspectiveWidget, is_perspective3
+except ImportError:
+    raise unittest.SkipTest("skipping perspective tests")
+
+
+class MyStruct(csp.Struct):
+    my_str: str
+    my_float: float
+    my_bool: bool
+    my_date: date
+    my_datetime: datetime
+
+
+def my_graph(output={}):
+    adapter = PerspectiveAdapter(8000)
+    table = adapter.create_table("Test")
+    data = MyStruct(
+        my_str="foo", my_float=1.0, my_bool=False, my_date=date(2020, 1, 1), my_datetime=datetime(2020, 1, 1)
+    )
+    table.publish(csp.unroll(csp.const([data, data])))
+    output["table"] = table
+
+
+class TestPerspectiveAdapter(unittest.TestCase):
+    def test_adapter(self):
+        output = {}
+        csp.run(my_graph, output, starttime=datetime.utcnow(), endtime=timedelta(seconds=1))

csp/tests/impl/test_pandas_perspective.py

@@ -13,14 +13,10 @@
     import ipywidgets
     import perspective
 
-    if version.parse(perspective.__version__) >= version.parse("3"):
-        _PERSPECTIVE_3 = True
-        from perspective.widget import PerspectiveWidget
-    else:
-        _PERSPECTIVE_3 = False
-        from perspective import PerspectiveWidget
-
     from csp.impl.pandas_perspective import CspPerspectiveMultiTable, CspPerspectiveTable
+    from csp.impl.perspective_common import PerspectiveWidget, is_perspective3
+
+    _PERSPECTIVE_3 = is_perspective3()
 except ImportError:
     raise unittest.SkipTest("skipping perspective tests")
 

pyproject.toml

@@ -67,7 +67,7 @@ develop = [
     # lint
     "codespell>=2.2.6,<2.3",
     "isort>=5,<6",
-    "mdformat>=0.7.17,<0.8",
+    "mdformat==0.7.17",  # >0.7.17 doesnot support python 3.8
     "ruff>=0.3,<0.4",
     # test
     "pytest",