267 change communication to use pure json strings (#274)

* Update python version in env

* Converted server into a class

* Fixed server class

* Respond with JSON strings

* Fix documentation

* Make catch all return "UNKNOWN_EVENT"

* Default configure for sensapex

* Read and parse input as strings

pyproject.toml

@@ -55,6 +55,7 @@ path = "src/ephys_link/__about__.py"
 exclude = ["/.github", "/.idea"]
 
 [tool.hatch.envs.default]
+python = "3.12"
 dependencies = [
     "coverage[toml]>=6.5",
     "pytest",

src/ephys_link/__main__.py

@@ -1,20 +1,23 @@
 import argparse
 import signal
 import time
-from importlib import metadata
 from threading import Event, Thread
 
 import serial
 import serial.tools.list_ports as ports
 
 from ephys_link import common as com
-from ephys_link import server
+from ephys_link.__about__ import __version__ as version
+from ephys_link.server import Server
 
 # Setup Arduino serial port (emergency stop)
 poll_rate = 0.05
 kill_serial_event = Event()
 poll_serial_thread: Thread
 
+# Create Server
+server = Server()
+
 
 def poll_serial(kill_event: Event, serial_port: str) -> None:
     """Continuously poll serial port for data
@@ -100,7 +103,7 @@ def close_serial(_, __) -> None:
     "-v",
     "--version",
     action="version",
-    version=f"Electrophysiology Manipulator Link v{metadata.version('ephys_link')}",
+    version=f"Electrophysiology Manipulator Link v{version}",
     help="Print version and exit",
 )
 

src/ephys_link/common.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import TypedDict
 
 # Debugging flag
@@ -81,6 +82,10 @@ def __init__(self, manipulators: list, num_axes: int, dimensions: list, error: s
             error=error,
         )
 
+    def json(self) -> str:
+        """Return JSON string"""
+        return json.dumps(self)
+
 
 class PositionalOutputData(dict):
     """Output format for (position, error)
@@ -98,6 +103,10 @@ def __init__(self, position: list, error: str) -> None:
         """Constructor"""
         super().__init__(position=position, error=error)
 
+    def json(self) -> str:
+        """Return JSON string"""
+        return json.dumps(self)
+
 
 class AngularOutputData(dict):
     """Output format for (angles, error)
@@ -112,6 +121,10 @@ def __init__(self, angles: list, error: str) -> None:
         """Constructor"""
         super().__init__(angles=angles, error=error)
 
+    def json(self) -> str:
+        """Return JSON string"""
+        return json.dumps(self)
+
 
 class ShankCountOutputData(dict):
     """Output format for (num_shanks, error)
@@ -126,6 +139,10 @@ def __init__(self, shank_count: int, error: str) -> None:
         """Constructor"""
         super().__init__(shank_count=shank_count, error=error)
 
+    def json(self) -> str:
+        """Return JSON string"""
+        return json.dumps(self)
+
 
 class DriveToDepthOutputData(dict):
     """Output format for depth driving (depth, error)
@@ -142,6 +159,10 @@ def __init__(self, depth: float, error: str) -> None:
         """Create drive to depth output data dictionary"""
         super().__init__(depth=depth, error=error)
 
+    def json(self) -> str:
+        """Return JSON string"""
+        return json.dumps(self)
+
 
 class StateOutputData(dict):
     """Output format for (state, error)
@@ -157,3 +178,7 @@ class StateOutputData(dict):
     def __init__(self, state: bool, error: str) -> None:
         """Create state output data dictionary"""
         super().__init__(state=state, error=error)
+
+    def json(self) -> str:
+        """Return JSON string"""
+        return json.dumps(self)

src/ephys_link/gui.py

@@ -1,4 +1,3 @@
-# pylama:skip=1
 import socket
 from argparse import Namespace
 from threading import Event, Thread

src/ephys_link/platform_handler.py

@@ -98,7 +98,6 @@ def register_manipulator(self, manipulator_id: str) -> str:
         try:
             # Register manipulator
             self._register_manipulator(manipulator_id)
-            com.dprint(f"[SUCCESS]\t Registered manipulator: {manipulator_id}\n")
         except ValueError as ve:
             # Manipulator not found in UMP
             print(f"[ERROR]\t\t Manipulator not found: {manipulator_id}: {ve}\n")
@@ -109,6 +108,7 @@ def register_manipulator(self, manipulator_id: str) -> str:
             print(f"{type(e)}: {e}\n")
             return "Error registering manipulator"
         else:
+            com.dprint(f"[SUCCESS]\t Registered manipulator: {manipulator_id}\n")
             return ""
 
     def unregister_manipulator(self, manipulator_id: str) -> str: