v1.9.0

version 1.9.0

CHANGELOG.rst

@@ -1,3 +1,7 @@
+v1.9.0 (2023-12-23)
+*******************
+* Added getters and safe getters for Image class.
+
 v1.3.0 (2023-02-04)
 *******************
 * Adopted LCO default task to new LCO portal.

mypy.ini

@@ -7,6 +7,9 @@ implicit_reexport = True
 # Minimum version supported
 python_version = 3.9
 
+packages=pyobs
+exclude = docs/|test/
+
 [mypy-pyobs.comm.xmpp.xep_0009.*]
 ignore_errors = True
 

pyobs/application.py

@@ -6,7 +6,7 @@
 import threading
 from io import StringIO
 from logging.handlers import TimedRotatingFileHandler
-from typing import Optional, Any, Dict
+from typing import Optional, Any, Dict, List
 import yaml
 
 from pyobs.object import get_object, get_class_from_string
@@ -35,7 +35,7 @@ def __init__(self, config: str, log_file: Optional[str] = None, log_level: str =
 
         # formatter for logging, and list of logging handlers
         formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(filename)s:%(lineno)d %(message)s")
-        handlers = []
+        handlers: List[logging.Handler] = []
 
         # create stdout logging handler
         stream_handler = logging.StreamHandler()
@@ -110,7 +110,7 @@ def run(self) -> None:
         log.info("Closing loop...")
         self._loop.close()
 
-    def _signal_handler(self, sig) -> None:
+    def _signal_handler(self, sig: int) -> None:
         """React to signals and quit module."""
 
         # stop loop

pyobs/cli/pyobs.py

@@ -1,11 +1,14 @@
 import argparse
 import os
-from typing import Type, Any
+from typing import Type, Any, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pyobs.application import Application
 
 from pyobs import version
 
 
-def init_cli():
+def init_cli() -> argparse.ArgumentParser:
     # init argument parsing
     # for all command line parameters we set the default to an environment variable,
     # so they can also be specified that way
@@ -30,7 +33,7 @@ def init_cli():
     return parser
 
 
-def parse_cli(parser: argparse.ArgumentParser):
+def parse_cli(parser: argparse.ArgumentParser) -> dict[str, Any]:
     from pyobs.utils.time import Time
 
     # parse args
@@ -50,7 +53,7 @@ def parse_cli(parser: argparse.ArgumentParser):
     return vars(args)
 
 
-def start_daemon(app_class, pid_file=None, **kwargs: Any) -> None:
+def start_daemon(app_class: Type["Application"], pid_file: str, **kwargs: Any) -> None:
     """Start process as a daemon.
 
     Args:
@@ -70,7 +73,7 @@ def start_daemon(app_class, pid_file=None, **kwargs: Any) -> None:
         run(app_class, **kwargs)
 
 
-def run(app_class: Type, **kwargs: Any) -> None:
+def run(app_class: Type["Application"], **kwargs: Any) -> None:
     """Run a pyobs application with the given options.
 
     Args:

pyobs/cli/pyobsw.py

@@ -1,7 +1,7 @@
 from .pyobs import init_cli, parse_cli, run
 
 
-def main():
+def main() -> None:
     from pyobs.application import GuiApplication
 
     # init argument parsing and parse it

pyobs/comm/dbus/patch.py

@@ -14,7 +14,7 @@
 from dbus_next import introspection as intr
 
 
-def patch():
+def patch() -> None:
     """
     Patches dbus-next to allow for a providing the sender name. Probably a bad idea, but better than using
     a fork of the project.

pyobs/comm/dummy/dummycomm.py

@@ -1,6 +1,6 @@
 import inspect
 import logging
-from typing import Any, List, Type
+from typing import Any, List, Type, Dict
 
 from pyobs.comm import Comm
 from pyobs.events import Event
@@ -22,16 +22,26 @@ def clients(self) -> List[str]:
         """Always return zero clients."""
         return []
 
-    def get_interfaces(self, client: str) -> List[Type[Interface]]:
+    async def get_interfaces(self, client: str) -> List[Type[Interface]]:
         """No interfaces implemented."""
         return []
 
-    def _supports_interface(self, client: str, interface: Type[Interface]) -> bool:
+    async def _supports_interface(self, client: str, interface: Type[Interface]) -> bool:
         """Interfaces are never supported."""
         return False
 
-    def execute(self, client: str, method: str, signature: inspect.Signature, *args: Any) -> Future:
-        """Always fake a successful execution of a method."""
+    async def execute(self, client: str, method: str, annotation: Dict[str, Any], *args: Any) -> Any:
+        """Execute a given method on a remote client.
+
+        Args:
+            client (str): ID of client.
+            method (str): Method to call.
+            annotation: Method annotation.
+            *args: List of parameters for given method.
+
+        Returns:
+            Passes through return from method call.
+        """
         return Future(empty=True)
 
     @property

pyobs/comm/local/localcomm.py

@@ -9,16 +9,15 @@
 
 
 class LocalComm(Comm):
-    def __init__(self, name: str, *args, **kwargs):
-
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         Comm.__init__(self, *args, **kwargs)
 
         self._name = name
         self._network = pyobs.comm.local.LocalNetwork()
         self._network.connect_client(self)
 
     @property
-    def name(self) -> Optional[str]:
+    def name(self) -> str:
         """Name of this client."""
         return self._name
 
@@ -45,7 +44,7 @@ async def get_interfaces(self, client: str) -> List[Type[Interface]]:
         """
 
         remote_client: LocalComm = self._network.get_client(client)
-        return remote_client.module.interfaces
+        return [] if remote_client.module is None else remote_client.module.interfaces
 
     async def _supports_interface(self, client: str, interface: Type[Interface]) -> bool:
         """Checks, whether the given client supports the given interface.
@@ -55,10 +54,10 @@ async def _supports_interface(self, client: str, interface: Type[Interface]) ->
             interface: Interface to check.
 
         Returns:
-            Whether or not interface is supported.
+            Whether interface is supported.
         """
         interfaces = await self.get_interfaces(client)
-        return interfaces in interfaces
+        return interface in interfaces
 
     async def execute(self, client: str, method: str, annotation: Dict[str, Any], *args: Any) -> Any:
         """Execute a given method on a remote client.
@@ -74,10 +73,12 @@ async def execute(self, client: str, method: str, annotation: Dict[str, Any], *a
         """
 
         remote_client = self._network.get_client(client)
+        if remote_client.module is None:
+            raise ValueError
         simple_results = await remote_client.module.execute(method, *args)
         real_results = cast_response_to_real(
-                simple_results, annotation["return"], self.cast_to_real_pre, self.cast_to_real_post
-            )
+            simple_results, annotation["return"], self.cast_to_real_pre, self.cast_to_real_post
+        )
         return real_results
 
     async def send_event(self, event: Event) -> None:
@@ -94,4 +95,4 @@ async def send_event(self, event: Event) -> None:
     async def _register_events(
         self, events: List[Type[Event]], handler: Optional[Callable[[Event, str], Coroutine[Any, Any, bool]]] = None
     ) -> None:
-        pass
+        pass

pyobs/comm/local/localnetwork.py

@@ -1,25 +1,25 @@
 from __future__ import annotations
 
-from typing import Dict, List
+from typing import Dict, List, Optional
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     import pyobs.comm
 
 
 class LocalNetwork:
-    _instance = None
+    _instance: Optional["LocalNetwork"] = None
 
-    def __new__(cls):
+    def __new__(cls) -> "LocalNetwork":
         if cls._instance is None:
-            print('Creating the object')
+            print("Creating the object")
             cls._instance = super(LocalNetwork, cls).__new__(cls)
 
             cls._clients: Dict[str, pyobs.comm.local.LocalComm] = {}
 
         return cls._instance
 
-    def connect_client(self, comm: pyobs.comm.local.LocalComm):
+    def connect_client(self, comm: pyobs.comm.local.LocalComm) -> None:
         self._clients[comm.name] = comm
 
     def get_client(self, name: str) -> pyobs.comm.local.LocalComm: