WIP: Support connecting to LiveSplitOne

pyproject.toml

@@ -2,32 +2,34 @@
 requires = [
     "setuptools>=77.0",
     "setuptools-scm>=8",
-    "pyside6-essentials~=6.9",  # Needed to build uic files
+    "pyside6-essentials~=6.9", # Needed to build uic files
 ]
 build-backend = "setuptools.build_meta"
 
 [project]
-name="splitguides"
-description="A tool for speedrunners to display notes that advance automatically with their splits in Livesplit."
+name = "splitguides"
+description = "A tool for speedrunners to display notes that advance automatically with their splits in Livesplit."
 authors = [
-  { name = "David C Ellis" },
+    { name = "David C Ellis" },
 ]
-readme="README.md"
-requires-python = ">=3.12"  # 3.12 should function and build now
+readme = "README.md"
+requires-python = ">=3.12"
 dependencies = [
     "pyside6~=6.9",
     "jinja2~=3.1",
-    "bleach[css]==6.0",  # Each upgrade to bleach has broken something so pin it.
+    "bleach[css]==6.0",              # Each upgrade to bleach has broken something so pin it.
     "flask~=3.0",
     "markdown~=3.6",
     "keyboard~=0.13.5",
     "ducktools-classbuilder>=0.7.4",
     "waitress~=3.0",
+    "platformdirs~=4.3",
+    "websockets>=15.0.1",
 ]
 classifiers = [
     "Development Status :: 4 - Beta",
     "Operating System :: Microsoft :: Windows",
-    "Operating System :: POSIX :: Linux",  # Should also work on Linux, even if livesplit doesn't
+    "Operating System :: POSIX :: Linux",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
     "Topic :: Games/Entertainment",
@@ -59,14 +61,7 @@ __version_tuple__ = {version_tuple}
 """
 
 [tool.pytest.ini_options]
-addopts= "--cov=src/ --cov-report=term-missing"
+addopts = "--cov=src/ --cov-report=term-missing"
 testpaths = [
     "tests",
 ]
-
-[tool.uv]
-# Only x86_64 / AMD64 is supported
-environments = [
-    "platform_system == 'Windows' and platform_machine == 'AMD64' and implementation_name == 'cpython'",
-    "platform_system == 'Linux' and platform_machine == 'x86_64' and implementation_name == 'cpython'",
-]

src/splitguides/livesplit_client.py

@@ -44,6 +44,7 @@ class LivesplitConnection(Prefab):
     port: int = 16834
     timeout: int = 1
     sock: socket.socket | None = attribute(default=None, init=False, repr=False)
+    timer: str = "LiveSplit"
 
     def connect(self) -> bool:
         """

src/splitguides/livesplitone_client.py

@@ -0,0 +1,244 @@
+import json
+from websockets.sync.client import connect, ClientConnection
+from datetime import timedelta
+import typing
+
+from ducktools.classbuilder.prefab import Prefab, attribute
+
+from .livesplit_client import BUFFER_SIZE, parse_time
+
+class LivesplitoneConnection(Prefab):
+    """
+    Socket based livesplit one connection model
+    """
+    server: str = "localhost"
+    port: int = 16834
+    timeout: int = 1
+    sock: ClientConnection | None = attribute(default=None, init=False, repr=False)
+    timer: str = "LiveSplitOne"
+
+    def connect(self) -> bool:
+        """
+        Attempt to connect to the livesplit one server
+        :return: True if connected, otherwise False
+        """
+        self.sock = connect(f"ws://{self.server}:{self.port}")
+        return True
+
+    def close(self) -> None:
+        if self.sock:
+            self.sock.close()
+            self.sock = None
+
+    def send(self, msg: bytes) -> None:
+        """
+        Send a message to the livesplit one server - connect if not already connected.
+        If the connection is aborted (ie: if livesplit one server has been closed)
+        raise a ConnectionAbortedError
+
+        :param msg: bytes message to send (should end with "\r\n")
+        :return:
+        """
+        if not self.sock:
+            self.connect()
+
+        if self.sock:  # Check again in case connection failed
+            try:
+                self.sock.send(msg)
+            except ConnectionAbortedError:
+                self.sock.close()
+                self.sock = None
+                raise ConnectionAbortedError("The connection has been closed by the host")
+
+    def receive(self) -> bytes:
+        """
+        Attempt to receive a message from the livesplit one server
+        raise ConnectionError if the connection has been terminated.
+
+        :return: bytes received from the server
+        """
+        if not self.sock:
+            self.connect()
+
+        if self.sock:
+            try:
+                data_received = self.sock.recv(BUFFER_SIZE, False)
+            except OSError:
+                self.sock.close()
+                self.sock = None
+                raise ConnectionError("The connection has been closed by the host")
+
+            if data_received == b"":
+                self.sock.close()
+                self.sock = None
+                raise ConnectionError("The connection has been closed by the host")
+
+            return data_received
+
+        return b""
+
+
+class LivesplitoneMessaging(Prefab):
+    connection: LivesplitoneConnection
+
+    def connect(self) -> bool:
+        return self.connection.connect()
+
+    def close(self) -> None:
+        self.connection.close()
+
+    def send(self, message) -> None:
+        m = json.dumps(message).encode("UTF8")
+        self.connection.send(m + b"\r\n")
+
+    @typing.overload
+    def receive(self, datatype: typing.Literal["time"]) -> timedelta: ...
+    @typing.overload
+    def receive(self, datatype: typing.Literal["int"]) -> int: ...
+    @typing.overload
+    def receive(self, datatype: typing.Literal["state"]) -> dict: ...
+    @typing.overload
+    def receive(self, datatype: typing.Literal["text"] = "text") -> str: ...
+
+    def receive(self, datatype="text"):
+        result = self.connection.receive()
+        result = result.strip().decode("UTF8")
+        result = json.loads(result)["success"]
+        if datatype == "time":
+            result = parse_time(result["string"])
+        elif datatype == "int":
+            return int(result["string"])
+        elif datatype == "state":
+            return result
+
+        return result
+
+    def start_timer(self) -> None:
+        """
+        Start the timer
+        """
+        self.send({ "command": "start" })
+
+    def start_or_split(self) -> None:
+        """
+        Start the timer or split a running timer
+        """
+        self.send({ "command": "splitOrStart" })
+
+    def split(self) -> None:
+        """
+        Split
+        """
+        self.send({ "command": "split" })
+
+    def unsplit(self) -> None:
+        """
+        Undo the previous split
+        """
+        self.send({ "command": "undoSplit" })
+
+    def skip_split(self) -> None:
+        """
+        Skip the current split
+        """
+        self.send({ "command": "skipSplit" })
+
+    def pause(self) -> None:
+        """
+        Pause the timer
+        """
+        self.send({ "command": "pause" })
+
+    def resume(self) -> None:
+        """
+        Resume a paused timer
+        """
+        self.send({ "command": "resume" })
+
+    def reset(self) -> None:
+        """
+        Reset the timer
+        """
+        self.send({ "command": "reset" })
+
+    def init_game_time(self) -> None:
+        """
+        Activate the game timer
+        """
+        self.send({ "command": "initializeGameTime" })
+
+    def set_game_time(self, t: str) -> None:
+        """
+        Set the game timer
+        :param t:
+        :return:
+        """
+        self.send({ "command": "setGameTime", "time": t })
+
+    def set_loading_times(self, t: str) -> None:
+        """
+
+        :param t:
+        """
+        self.send({ "command": "setLoadingTimes", "time": t })
+
+    def pause_game_time(self) -> None:
+        """
+        Pause the game timer
+        """
+        self.send({ "command": "pauseGameTime" })
+
+    def unpause_game_time(self) -> None:
+        """
+        Unpause the game timer
+        """
+        self.send({ "command": "resumeGameTime" })
+
+    def set_comparison(self, comparison) -> None:
+        """
+        Change the comparison method
+
+        :param comparison: Time to compare against eg 'Personal Best' or 'Best Segments'
+        """
+        self.send({ "command": "setCurrentComparison", "comparison": comparison })
+
+    def get_last_split_time(self) -> timedelta:
+        self.send({ "command": "getCurrentRunSplitTime" })
+        return self.receive("time")
+
+    def get_comparison_split_time(self) -> timedelta:
+        self.send({ "command": "getComparisonTime" })
+        return self.receive("time")
+
+    def get_current_time(self) -> timedelta:
+        self.send({ "command": "getCurrentTime" })
+        return self.receive("time")
+
+    def get_split_index(self) -> int:
+        self.send({ "command": "getCurrentState" })
+        s = self.receive("state")
+        match s["state"]:
+            case "Running" | "Paused":
+                return s["index"]
+            case _:
+                return -1
+
+    def get_current_split_name(self) -> str:
+        self.send({ "command": "getSegmentName" })
+        return self.receive()
+
+    def get_previous_split_name(self) -> str:
+        self.send({ "command": "getSegmentName",  "index": -1, "relative": True })
+        return self.receive()
+
+    def get_current_timer_phase(self) -> str:
+        self.send({ "command": "getCurrentState" })
+        return self.receive("state")["state"]
+
+
+def get_livesplitone_client(
+        server: str = "localhost",
+        port: int = 16834,
+        timeout: int = 1
+) -> LivesplitoneMessaging:
+    return LivesplitoneMessaging(connection=LivesplitoneConnection(server, port, timeout))

src/splitguides/server/split_server.py

@@ -9,6 +9,7 @@
 
 from ..settings import ServerSettings
 from ..livesplit_client import get_client
+from ..livesplitone_client import get_livesplitone_client
 from ..note_parser import Notes
 
 KEEP_ALIVE = 10
@@ -58,7 +59,11 @@ def event_stream():
 
         current_note_index = None
         last_update = 0
-        client = get_client(settings.hostname, settings.port)
+        match settings.timer:
+            case "LiveSplitOne":
+                client = get_livesplitone_client(settings.hostname, settings.port)
+            case _: # "LiveSplit"
+                client = get_client(settings.hostname, settings.port)
         connected = client.connect()
         # Note if the previous state was not connected
         disconnected = True
@@ -73,8 +78,8 @@ def event_stream():
                 except (ConnectionError, TimeoutError):
                     connected = client.connect()
                     yield (
-                        f"data: <h2>Trying to connect to livesplit.</h2>"
-                        f"<h3>Make sure Livesplit server is running.</h3>{data}\n\n"
+                        f"data: <h2>Trying to connect to {settings.timer}.</h2>"
+                        f"<h3>Make sure {settings.timer} server is running.</h3>{data}\n\n"
                     )
                 else:
                     if current_note_index != new_index or disconnected:
@@ -99,8 +104,8 @@ def event_stream():
                 disconnected = True
                 connected = client.connect()
                 yield (
-                    f"data: <h2>Trying to connect to livesplit.</h2>"
-                    f"<h3>Make sure Livesplit server is running.</h3>{data}\n\n"
+                    f"data: <h2>Trying to connect to {settings.timer}.</h2>"
+                    f"<h3>Make sure {settings.timer} server is running.</h3>{data}\n\n"
                 )
             time.sleep(0.5)
 

src/splitguides/settings.py

@@ -15,6 +15,8 @@
 PROJECT_NAME = "splitguides"
 
 match sys.platform:
+    # Don't change existing win32/linux config locations
+    # but let platformdirs handle this for previously unsupported platforms
     case "win32":
         if _local_app_folder := os.environ.get("LOCALAPPDATA"):
             if not os.path.isdir(_local_app_folder):
@@ -27,9 +29,8 @@
     case "linux":
         SETTINGS_FOLDER = Path(os.path.expanduser(os.path.join("~", f".{PROJECT_NAME}")))
     case other:
-        raise UnsupportedPlatformError(
-            f"Platform {other!r} is not currently supported."
-        )
+        import platformdirs
+        SETTINGS_FOLDER = Path(platformdirs.user_config_dir(PROJECT_NAME))
 
 SETTINGS_FOLDER.mkdir(exist_ok=True)
 
@@ -70,6 +71,7 @@ class BaseSettings(metaclass=ABCMeta):
     output_file: Path = attribute(serialize=False)
 
     # Networking Settings
+    timer: str = "LiveSplit"
     hostname: str = "localhost"
     port: int = 16834