Cache HTTP data at 60 FPS

src/ephys_link/bindings/mpm_binding.py

@@ -1,8 +1,5 @@
 """Bindings for New Scale Pathfinder MPM HTTP server platform.
 
-MPM works slightly differently than the other platforms since it operates in stereotactic coordinates.
-This means exceptions need to be made for its API.
-
 Usage: Instantiate MPMBindings to interact with the New Scale Pathfinder MPM HTTP server platform.
 """
 
@@ -64,6 +61,9 @@ class MPMBinding(BaseBinding):
         "AM",
         "AN",
     )
+
+    # Server cache lifetime (60 FPS).
+    CACHE_LIFETIME = 1 / 60
 
     # Movement polling preferences.
     UNCHANGED_COUNTER_LIMIT = 10
@@ -81,6 +81,10 @@ def __init__(self, port: int = 8080) -> None:
         """
         self._url = f"http://localhost:{port}"
         self._movement_stopped = False
+
+        # Data cache.
+        self.cache = {}
+        self.cache_time = 0
 
     @staticmethod
     @override
@@ -279,8 +283,14 @@ def unified_space_to_platform_space(self, unified_space: Vector4) -> Vector4:
     # Helper functions.
     async def _query_data(self) -> dict[str, Any]:  # pyright: ignore [reportExplicitAny]
         try:
-            # noinspection PyTypeChecker
-            return (await get_running_loop().run_in_executor(None, get, self._url)).json()  # pyright: ignore [reportAny]
+            # Update cache if it's expired.
+            if get_running_loop().time() - self.cache_time > self.CACHE_LIFETIME:
+                # noinspection PyTypeChecker
+                self.cache = (await get_running_loop().run_in_executor(None, get, self._url)).json()  # pyright: ignore [reportAny]
+                self.cache_time = get_running_loop().time()
+
+            # Return cached data.
+            return self.cache
         except ConnectionError as connectionError:
             error_message = f"Unable to connect to MPM HTTP server: {connectionError}"
             raise RuntimeError(error_message) from connectionError