308 big text prompt

src/ephys_link/__about__.py

@@ -1 +1 @@
-__version__ = "1.2.5"
+__version__ = "1.2.6.dev0"

src/ephys_link/common.py

@@ -12,6 +12,18 @@
 # Debugging flag
 DEBUG = False
 
+# Ephys Link ASCII
+ASCII = r"""
+  ______       _                 _      _       _
+ |  ____|     | |               | |    (_)     | |
+ | |__   _ __ | |__  _   _ ___  | |     _ _ __ | | __
+ |  __| | '_ \| '_ \| | | / __| | |    | | '_ \| |/ /
+ | |____| |_) | | | | |_| \__ \ | |____| | | | |   <
+ |______| .__/|_| |_|\__, |___/ |______|_|_| |_|_|\_\
+        | |           __/ |
+        |_|          |___/
+"""
+
 
 def dprint(message: str) -> None:
     """Print message if debug is enabled.

src/ephys_link/server.py

@@ -11,17 +11,30 @@
 
 from __future__ import annotations
 
-import json
-import sys
+from json import loads
 from signal import SIGINT, SIGTERM, signal
+from sys import exit
 from typing import TYPE_CHECKING, Any
 
-import socketio
 from aiohttp import web
 from aiohttp.web_runner import GracefulExit
-
-from ephys_link import common as com
-from ephys_link.__about__ import __version__ as version
+from packaging import version
+from requests import get
+from requests.exceptions import ConnectionError
+from socketio import AsyncServer
+
+from ephys_link.__about__ import __version__
+from ephys_link.common import (
+    ASCII,
+    CanWriteInputDataFormat,
+    DriveToDepthInputDataFormat,
+    DriveToDepthOutputData,
+    GotoPositionInputDataFormat,
+    InsideBrainInputDataFormat,
+    PositionalOutputData,
+    StateOutputData,
+    dprint,
+)
 from ephys_link.platforms.new_scale_handler import NewScaleHandler
 from ephys_link.platforms.new_scale_pathfinder_handler import NewScalePathfinderHandler
 from ephys_link.platforms.sensapex_handler import SensapexHandler
@@ -34,7 +47,7 @@
 class Server:
     def __init__(self):
         # Server and Socketio
-        self.sio = socketio.AsyncServer()
+        self.sio = AsyncServer()
         self.app = web.Application()
 
         # Is there a client connected?
@@ -127,7 +140,7 @@ async def get_manipulators(self, _) -> str:
         :return: :class:`ephys_link.common.GetManipulatorsOutputData` as JSON formatted string.
         :rtype: str
         """
-        com.dprint("[EVENT]\t\t Get discoverable manipulators")
+        dprint("[EVENT]\t\t Get discoverable manipulators")
 
         return self.platform.get_manipulators().json()
 
@@ -141,7 +154,7 @@ async def register_manipulator(self, _, manipulator_id: str) -> str:
         :return: Error message on error, empty string otherwise.
         :rtype: str
         """
-        com.dprint(f"[EVENT]\t\t Register manipulator: {manipulator_id}")
+        dprint(f"[EVENT]\t\t Register manipulator: {manipulator_id}")
 
         return self.platform.register_manipulator(manipulator_id)
 
@@ -155,7 +168,7 @@ async def unregister_manipulator(self, _, manipulator_id: str) -> str:
         :return: Error message on error, empty string otherwise.
         :rtype: str
         """
-        com.dprint(f"[EVENT]\t\t Unregister manipulator: {manipulator_id}")
+        dprint(f"[EVENT]\t\t Unregister manipulator: {manipulator_id}")
 
         return self.platform.unregister_manipulator(manipulator_id)
 
@@ -169,7 +182,7 @@ async def get_pos(self, _, manipulator_id: str) -> str:
         :return: :class:`ephys_link.common.PositionalOutputData` as JSON formatted string.
         :rtype: str
         """
-        # com.dprint(f"[EVENT]\t\t Get position of manipulator" f" {manipulator_id}")
+        # dprint(f"[EVENT]\t\t Get position of manipulator" f" {manipulator_id}")
 
         return self.platform.get_pos(manipulator_id).json()
 
@@ -210,18 +223,18 @@ async def goto_pos(self, _, data: str) -> str:
         :rtype: str
         """
         try:
-            parsed_data: com.GotoPositionInputDataFormat = json.loads(data)
+            parsed_data: GotoPositionInputDataFormat = loads(data)
             manipulator_id = parsed_data["manipulator_id"]
             pos = parsed_data["pos"]
             speed = parsed_data["speed"]
         except KeyError:
             print(f"[ERROR]\t\t Invalid goto_pos data: {data}\n")
-            return com.PositionalOutputData([], "Invalid data format").json()
+            return PositionalOutputData([], "Invalid data format").json()
         except Exception as e:
             print(f"[ERROR]\t\t Error in goto_pos: {e}\n")
-            return com.PositionalOutputData([], "Error in goto_pos").json()
+            return PositionalOutputData([], "Error in goto_pos").json()
         else:
-            com.dprint(f"[EVENT]\t\t Move manipulator {manipulator_id} " f"to position {pos}")
+            dprint(f"[EVENT]\t\t Move manipulator {manipulator_id} " f"to position {pos}")
             goto_result = await self.platform.goto_pos(manipulator_id, pos, speed)
             return goto_result.json()
 
@@ -236,18 +249,18 @@ async def drive_to_depth(self, _, data: str) -> str:
         :rtype: str
         """
         try:
-            parsed_data: com.DriveToDepthInputDataFormat = json.loads(data)
+            parsed_data: DriveToDepthInputDataFormat = loads(data)
             manipulator_id = parsed_data["manipulator_id"]
             depth = parsed_data["depth"]
             speed = parsed_data["speed"]
         except KeyError:
             print(f"[ERROR]\t\t Invalid drive_to_depth data: {data}\n")
-            return com.DriveToDepthOutputData(-1, "Invalid data " "format").json()
+            return DriveToDepthOutputData(-1, "Invalid data " "format").json()
         except Exception as e:
             print(f"[ERROR]\t\t Error in drive_to_depth: {e}\n")
-            return com.DriveToDepthOutputData(-1, "Error in drive_to_depth").json()
+            return DriveToDepthOutputData(-1, "Error in drive_to_depth").json()
         else:
-            com.dprint(f"[EVENT]\t\t Drive manipulator {manipulator_id} to depth {depth}")
+            dprint(f"[EVENT]\t\t Drive manipulator {manipulator_id} to depth {depth}")
             drive_result = await self.platform.drive_to_depth(manipulator_id, depth, speed)
             return drive_result.json()
 
@@ -262,17 +275,17 @@ async def set_inside_brain(self, _, data: str) -> str:
         :rtype: str
         """
         try:
-            parsed_data: com.InsideBrainInputDataFormat = json.loads(data)
+            parsed_data: InsideBrainInputDataFormat = loads(data)
             manipulator_id = parsed_data["manipulator_id"]
             inside = parsed_data["inside"]
         except KeyError:
             print(f"[ERROR]\t\t Invalid set_inside_brain data: {data}\n")
-            return com.StateOutputData(False, "Invalid data format").json()
+            return StateOutputData(False, "Invalid data format").json()
         except Exception as e:
             print(f"[ERROR]\t\t Error in inside_brain: {e}\n")
-            return com.StateOutputData(False, "Error in set_inside_brain").json()
+            return StateOutputData(False, "Error in set_inside_brain").json()
         else:
-            com.dprint(f"[EVENT]\t\t Set manipulator {manipulator_id} inside brain to {inside}")
+            dprint(f"[EVENT]\t\t Set manipulator {manipulator_id} inside brain to {inside}")
             return self.platform.set_inside_brain(manipulator_id, inside).json()
 
     async def calibrate(self, _, manipulator_id: str) -> str:
@@ -285,7 +298,7 @@ async def calibrate(self, _, manipulator_id: str) -> str:
         :return: Error message on error, empty string otherwise.
         :rtype: str
         """
-        com.dprint(f"[EVENT]\t\t Calibrate manipulator" f" {manipulator_id}")
+        dprint(f"[EVENT]\t\t Calibrate manipulator" f" {manipulator_id}")
 
         return await self.platform.calibrate(manipulator_id, self.sio)
 
@@ -299,7 +312,7 @@ async def bypass_calibration(self, _, manipulator_id: str) -> str:
         :return: Error message on error, empty string otherwise.
         :rtype: str
         """
-        com.dprint(f"[EVENT]\t\t Bypass calibration of manipulator" f" {manipulator_id}")
+        dprint(f"[EVENT]\t\t Bypass calibration of manipulator" f" {manipulator_id}")
 
         return self.platform.bypass_calibration(manipulator_id)
 
@@ -314,18 +327,18 @@ async def set_can_write(self, _, data: str) -> str:
         :rtype: str
         """
         try:
-            parsed_data: com.CanWriteInputDataFormat = json.loads(data)
+            parsed_data: CanWriteInputDataFormat = loads(data)
             manipulator_id = parsed_data["manipulator_id"]
             can_write = parsed_data["can_write"]
             hours = parsed_data["hours"]
         except KeyError:
             print(f"[ERROR]\t\t Invalid set_can_write data: {data}\n")
-            return com.StateOutputData(False, "Invalid data " "format").json()
+            return StateOutputData(False, "Invalid data " "format").json()
         except Exception as e:
             print(f"[ERROR]\t\t Error in inside_brain: {e}\n")
-            return com.StateOutputData(False, "Error in set_can_write").json()
+            return StateOutputData(False, "Error in set_can_write").json()
         else:
-            com.dprint(f"[EVENT]\t\t Set manipulator {manipulator_id} can_write state to {can_write}")
+            dprint(f"[EVENT]\t\t Set manipulator {manipulator_id} can_write state to {can_write}")
             return self.platform.set_can_write(manipulator_id, can_write, hours, self.sio).json()
 
     def stop(self, _) -> bool:
@@ -336,7 +349,7 @@ def stop(self, _) -> bool:
         :return: True if successful, False otherwise.
         :rtype: bool
         """
-        com.dprint("[EVENT]\t\t Stop all manipulators")
+        dprint("[EVENT]\t\t Stop all manipulators")
 
         return self.platform.stop()
 
@@ -379,16 +392,33 @@ def launch(self, platform_type: str, server_port: int, pathfinder_port: int | No
         elif platform_type == "new_scale_pathfinder":
             self.platform = NewScalePathfinderHandler(pathfinder_port)
         else:
-            sys.exit(f"[ERROR]\t\t Invalid manipulator type: {platform_type}")
+            exit(f"[ERROR]\t\t Invalid manipulator type: {platform_type}")
+
+        # Preamble.
+        print(ASCII)
+        print(f"v{__version__}")
+
+        # Check for newer version.
+        try:
+            version_request = get("https://api.github.com/repos/VirtualBrainLab/ephys-link/tags", timeout=10)
+            latest_version = version_request.json()[0]["name"]
+            if version.parse(latest_version) > version.parse(__version__):
+                print(f"New version available: {latest_version}")
+                print("Download at: https://github.com/VirtualBrainLab/ephys-link/releases/latest")
+        except ConnectionError:
+            pass
 
-        # Preamble
-        print(f"=== Ephys Link v{version} ===")
+        # Explain window.
+        print()
+        print("This is the Ephys Link server window.")
+        print("You may safely leave it running in the background.")
+        print("To stop the it, close this window or press CTRL + Pause/Break.")
+        print()
 
         # List available manipulators
         print("Available Manipulators:")
         print(self.platform.get_manipulators()["manipulators"])
-
-        print("\n(Shutdown server with CTRL+Pause/Break)\n")
+        print()
 
         # Mark that server is running
         self.is_running = True