Add overwatcher_is_running function

pyproject.toml

@@ -65,7 +65,6 @@ sphinx-autobuild = ">=2021.3.14"
 pyds9 = "^1.8.1"
 sphinx-autodoc-typehints = "^1.23.2"
 ruff = ">=0.5.0"
-types-redis = "^4.6.0.20240425"
 
 [tool.ruff]
 line-length = 88

src/gort/overwatcher/calibration.py

@@ -73,8 +73,8 @@ def time_to_calibration(self):
     async def get_last_taken(self):
         """Gets the last time the observation was taken from Redis."""
 
-        client = redis_client()
-        data = await client.hgetall(f"gort:overwatcher:calibrations:{self.name}")
+        async with redis_client() as client:
+            data = await client.hgetall(f"gort:overwatcher:calibrations:{self.name}")
 
         if data == {}:
             return None
@@ -86,11 +86,11 @@ async def mark_done(self):
 
         self.done = True
 
-        client = redis_client()
-        await client.hset(
-            f"gort:overwatcher:calibrations:{self.name}",
-            mapping={"last_taken": time.Time.now().isot},
-        )
+        async with redis_client() as client:
+            await client.hset(
+                f"gort:overwatcher:calibrations:{self.name}",
+                mapping={"last_taken": time.Time.now().isot},
+            )
 
 
 class CalibrationSchedule:

src/gort/overwatcher/observer.py

@@ -22,7 +22,7 @@
 
 
 if TYPE_CHECKING:
-    from gort.observer import GortObserver
+    pass
 
 
 __all__ = ["ObserverOverwatcher"]
@@ -265,8 +265,9 @@ async def is_enabled(self):
         """Is observing enabled?"""
 
         try:
-            client = redis_client()
-            enabled: str | None = await client.get("gort:overwatcher:enabled")
+            async with redis_client() as client:
+                enabled: str | None = await client.get("gort:overwatcher:enabled")
+
             if not enabled:
                 raise ValueError("cannot determine if observing is enabled.")
             return bool(int(enabled))

src/gort/tools.py

@@ -18,10 +18,10 @@
 import re
 import tempfile
 import warnings
-from contextlib import suppress
+from contextlib import asynccontextmanager, suppress
 from functools import partial
 
-from typing import TYPE_CHECKING, Any, Callable, Coroutine, Sequence
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Callable, Coroutine, Sequence
 
 import httpx
 import numpy
@@ -75,6 +75,7 @@
     "handle_signals",
     "get_lvmapi_route",
     "GuiderMonitor",
+    "overwatcher_is_running",
 ]
 
 AnyPath = str | os.PathLike
@@ -557,13 +558,16 @@ def get_db_connection():
     return conn
 
 
-def redis_client():
+@asynccontextmanager
+async def redis_client() -> AsyncGenerator[aioredis.Redis, None]:
     """Returns a Redis connection from the configuration file parameters."""
 
     redis_config = config["services"]["redis"]
     client = aioredis.from_url(redis_config["url"], decode_responses=True)
 
-    return client
+    yield client
+
+    await client.aclose()
 
 
 async def run_in_executor(fn, *args, catch_warnings=False, executor="thread", **kwargs):
@@ -1000,3 +1004,12 @@ def to_header(self):
                     continue
 
         return header
+
+
+async def overwatcher_is_running():
+    """Returns :obj:`True` if the overwatcher is running."""
+
+    async with redis_client() as client:
+        enabled: str | None = await client.get("gort:overwatcher:enabled")
+
+    return bool(int(enabled)) if enabled else False