Upgrade csp to perspective 3.x (#392)

* Upgrade csp to perspective 3.x

Signed-off-by: Pascal Tomecek <pascal.tomecek@cubistsystematic.com>
Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com>

Author: ptomecek

.github/workflows/build.yml

@@ -670,9 +670,9 @@ jobs:
           python-version:
             - 3.9
           package:
-            - "sqlalchemy>=2"
             - "sqlalchemy<2"
-            - "numpy==1.19.5"
+            - "numpy==1.22.4"  # Min supported version of pandas 2.2
+            - "perspective-python<3"
 
       runs-on: ${{ matrix.os }}
 
@@ -709,11 +709,15 @@ jobs:
 
         - name: Python Test Steps
           run: make test TEST_ARGS="-k TestDBReader"
-          if: ${{ contains( 'sqlalchemy', matrix.package )}}
+          if: ${{ contains( matrix.package, 'sqlalchemy' )}}
 
         - name: Python Test Steps
           run: make test
-          if: ${{ contains( 'numpy', matrix.package )}}
+          if: ${{ contains( matrix.package, 'numpy' )}}
+
+        - name: Python Test Steps
+          run: make test TEST_ARGS="-k Perspective"
+          if: ${{ contains( matrix.package, 'perspective' )}}
 
     ###########################################################################################################
     #.........................................................................................................#

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,9 +1,16 @@
 import threading
 from datetime import timedelta
+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,20 +21,17 @@
     raise ImportError("perspective adapter requires tornado package")
 
 
-try:
-    from perspective import PerspectiveManager, Table as Table_, View as View_, __version__, set_threadpool_size
-
-    MAJOR, MINOR, PATCH = map(int, __version__.split("."))
-    if (MAJOR, MINOR, PATCH) < (0, 6, 2):
-        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
-def perspective_thread(manager):
+def perspective_thread(client):
     loop = tornado.ioloop.IOLoop()
-    manager.set_loop_callback(loop.add_callback)
+    client.set_loop_callback(loop.add_callback)
     loop.start()
 
 
@@ -38,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:
@@ -54,19 +71,25 @@ def _apply_updates(table: object, data: {str: ts[object]}, throttle: timedelta):
 
 
 @csp.node
-def _launch_application(port: int, manager: object, stub: ts[object]):
+def _launch_application(port: int, server: object, stub: ts[object]):
     with csp.state():
         s_app = None
         s_ioloop = None
         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(
             [
                 # create a websocket endpoint that the client Javascript can access
-                (r"/websocket", PerspectiveTornadoHandler, {"manager": manager, "check_origin": True})
+                (r"/websocket", PerspectiveTornadoHandler, handler_args)
             ],
             websocket_ping_interval=15,
         )
@@ -196,21 +219,34 @@ def create_table(self, name, limit=None, index=None):
         return table
 
     def _instantiate(self):
-        set_threadpool_size(self._threadpool_size)
-
-        manager = PerspectiveManager()
+        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
 
-        thread = threading.Thread(target=perspective_thread, kwargs=dict(manager=manager))
+            set_threadpool_size(self._threadpool_size)
+            manager = PerspectiveManager()
+            thread = threading.Thread(target=perspective_thread, kwargs=dict(manager=manager))
         thread.daemon = True
         thread.start()
 
         for table_name, table in self._tables.items():
             schema = {
                 k: v.tstype.typ if not issubclass(v.tstype.typ, csp.Enum) else str for k, v in table.columns.items()
             }
-            ptable = Table(schema, limit=table.limit, index=table.index)
-            manager.host_table(table_name, ptable)
+            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)
 
             _apply_updates(ptable, table.columns, self._throttle)
 
-        _launch_application(self._port, manager, csp.const("stub"))
+        if _PERSPECTIVE_3:
+            _launch_application(self._port, server, csp.const("stub"))
+        else:
+            _launch_application(self._port, manager, csp.const("stub"))

csp/dataframe.py

@@ -1,4 +1,5 @@
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta
+from packaging import version
 from typing import Dict, Optional
 
 import csp.baselib
@@ -12,6 +13,7 @@ class DataFrame:
     def __init__(self, data: Optional[Dict] = None):
         self._data = data or {}
         self._columns = list(self._data.keys())
+        self._psp_client = None
 
     @property
     def columns(self):
@@ -204,10 +206,17 @@ def to_perspective(self, starttime: datetime, endtime: datetime = None, realtime
         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
+
             global RealtimePerspectiveWidget
             if RealtimePerspectiveWidget is None:
 
-                class RealtimePerspectiveWidget(perspective.PerspectiveWidget):
+                class RealtimePerspectiveWidget(PerspectiveWidget):
                     def __init__(self, engine_runner, *args, **kwargs):
                         super().__init__(*args, **kwargs)
                         self._runner = engine_runner
@@ -222,14 +231,14 @@ def join(self):
                         self._runner.join()
 
         except ImportError:
-            raise ImportError("eval_perspective requires perspective-python installed")
+            raise ImportError("to_perspective requires perspective-python installed")
 
         if not realtime:
             df = self.to_pandas(starttime, endtime)
-            return perspective.PerspectiveWidget(df.ffill(), plugin="Y Line", columns=self._columns, group_by="index")
+            return PerspectiveWidget(df.ffill(), plugin="Y Line", columns=self._columns, group_by="index")
 
         @csp.node
-        def apply_updates(table: object, data: {str: csp.ts[object]}, timecol: str, throttle: timedelta):
+        def apply_updates(table: object, data: Dict[str, csp.ts[object]], timecol: str, throttle: timedelta):
             with csp.alarms():
                 alarm = csp.alarm(bool)
             with csp.state():
@@ -240,7 +249,10 @@ def apply_updates(table: object, data: {str: csp.ts[object]}, timecol: str, thro
 
             if csp.ticked(data):
                 s_buffer.append(dict(data.tickeditems()))
-                s_buffer[-1][timecol] = csp.now()
+                if _PERSPECTIVE_3:
+                    s_buffer[-1][timecol] = int(csp.now().timestamp() * 1000)
+                else:
+                    s_buffer[-1][timecol] = csp.now()
 
             if csp.ticked(alarm):
                 if len(s_buffer) > 0:
@@ -252,7 +264,21 @@ def apply_updates(table: object, data: {str: csp.ts[object]}, timecol: str, thro
         timecol = "time"
         schema = {k: v.tstype.typ for k, v in self._data.items()}
         schema[timecol] = datetime
-        table = perspective.Table(schema)
+        if _PERSPECTIVE_3:
+            perspective_type_map = {
+                str: "string",
+                float: "float",
+                int: "integer",
+                date: "date",
+                datetime: "datetime",
+                bool: "boolean",
+            }
+            schema = {col: perspective_type_map[typ] for col, typ in schema.items()}
+            if self._psp_client is None:
+                self._psp_client = perspective.Server().new_local_client()
+            table = self._psp_client.table(schema)
+        else:
+            table = perspective.Table(schema)
         runner = csp.run_on_thread(
             apply_updates,
             table,