Merge branch 'develop' into test-image

pyobs/comm/local/__init__.py

@@ -0,0 +1,2 @@
+from .localcomm import LocalComm
+from .localnetwork import LocalNetwork

pyobs/comm/local/localcomm.py

@@ -0,0 +1,97 @@
+import pyobs.comm
+
+from typing import Optional, List, Type, Dict, Any, Callable, Coroutine
+
+from pyobs.comm import Comm
+from pyobs.events import Event
+from pyobs.interfaces import Interface
+from pyobs.utils.types import cast_response_to_real
+
+
+class LocalComm(Comm):
+    def __init__(self, name: str, *args, **kwargs):
+
+        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]:
+        """Name of this client."""
+        return self._name
+
+    @property
+    def clients(self) -> List[str]:
+        """Returns list of currently connected clients.
+
+        Returns:
+            (list) List of currently connected clients.
+        """
+        return self._network.get_client_names()
+
+    async def get_interfaces(self, client: str) -> List[Type[Interface]]:
+        """Returns list of interfaces for given client.
+
+        Args:
+            client: Name of client.
+
+        Returns:
+            List of supported interfaces.
+
+        Raises:
+            IndexError: If client cannot be found.
+        """
+
+        remote_client: LocalComm = self._network.get_client(client)
+        return remote_client.module.interfaces
+
+    async def _supports_interface(self, client: str, interface: Type[Interface]) -> bool:
+        """Checks, whether the given client supports the given interface.
+
+        Args:
+            client: Client to check.
+            interface: Interface to check.
+
+        Returns:
+            Whether or not interface is supported.
+        """
+        interfaces = await self.get_interfaces(client)
+        return interfaces 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.
+
+        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.
+        """
+
+        remote_client = self._network.get_client(client)
+        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
+            )
+        return real_results
+
+    async def send_event(self, event: Event) -> None:
+        """Send an event to other clients.
+
+        Args:
+            event (Event): Event to send
+        """
+
+        remote_clients = self._network.get_clients()
+        for client in remote_clients:
+            client._send_event_to_module(event, self.name)
+
+    async def _register_events(
+        self, events: List[Type[Event]], handler: Optional[Callable[[Event, str], Coroutine[Any, Any, bool]]] = None
+    ) -> None:
+        pass

pyobs/comm/local/localnetwork.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import Dict, List
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import pyobs.comm
+
+
+class LocalNetwork:
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            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):
+        self._clients[comm.name] = comm
+
+    def get_client(self, name: str) -> pyobs.comm.local.LocalComm:
+        return self._clients[name]
+
+    def get_clients(self) -> List[pyobs.comm.local.LocalComm]:
+        return list(self._clients.values())
+
+    def get_client_names(self) -> List[str]:
+        return list(self._clients.keys())

tests/comm/local/test_localcomm.py

@@ -0,0 +1,93 @@
+import pytest
+from typing import Any
+
+from pyobs.comm.local import LocalNetwork, LocalComm
+from pyobs.events import Event, GoodWeatherEvent
+from pyobs.interfaces import ICamera, IExposureTime, IImageType, IModule, IConfig
+from pyobs.mixins import ImageFitsHeaderMixin
+from pyobs.modules import Module
+from pyobs.modules.camera import BaseCamera
+from pyobs.utils.enums import ImageType
+
+
+def test_init():
+    comm = LocalComm("test")
+
+    assert comm._name == "test"
+    assert comm._network.get_client("test") == comm
+
+
+def test_name():
+    comm = LocalComm("test")
+
+    assert comm.name == "test"
+
+
+def test_clients(mocker):
+    comm = LocalComm("test")
+
+    clients = ["test", "telescope", "camera"]
+    mocker.patch.object(comm._network, "get_client_names", return_value=clients)
+
+    assert comm.clients == clients
+
+
+class TestModule(Module, IImageType):
+
+    async def set_image_type(self, image_type: ImageType, **kwargs: Any) -> None:
+        pass
+
+    async def get_image_type(self, **kwargs: Any) -> ImageType:
+        return ImageType.BIAS
+
+
+@pytest.mark.asyncio
+async def test_get_interfaces(mocker):
+    comm = LocalComm("test")
+
+    another_comm = LocalComm("camera")
+    another_comm.module = TestModule()
+
+    mocker.patch.object(comm._network, "get_client", return_value=another_comm)
+
+    interfaces = await comm.get_interfaces("camera")
+
+    assert set(interfaces) == {IConfig, IImageType, IModule}
+    comm._network.get_client.assert_called_once_with("camera")
+
+
+@pytest.mark.asyncio
+async def test_supports_interface(mocker):
+    comm = LocalComm("test")
+
+    mocker.patch.object(comm, "get_interfaces", return_value=[IConfig, IImageType, IModule])
+
+    assert comm._supports_interface("camera", IConfig)
+
+
+@pytest.mark.asyncio
+async def test_execute(mocker):
+    comm = LocalComm("test")
+
+    another_comm = LocalComm("camera")
+    another_comm.module = TestModule()
+
+    mocker.patch.object(comm._network, "get_client", return_value=another_comm)
+
+    assert await comm.execute("camera", "get_image_type", {"return": ImageType}) == ImageType.BIAS
+    comm._network.get_client.assert_called_once_with("camera")
+
+
+@pytest.mark.asyncio
+async def test_send_event(mocker):
+    comm = LocalComm("test")
+
+    another_comm = LocalComm("camera")
+    another_comm.module = TestModule()
+
+    mocker.patch.object(comm._network, "get_clients", return_value=[another_comm])
+    mocker.patch.object(another_comm, "_send_event_to_module")
+
+    await comm.send_event(Event())
+    another_comm._send_event_to_module.assert_called_once()
+

tests/comm/local/test_localnetwork.py

@@ -0,0 +1,44 @@
+from pyobs.comm.local import LocalNetwork, LocalComm
+
+
+def test_singleton():
+    net1 = LocalNetwork()
+    net2 = LocalNetwork()
+
+    assert net1 == net2
+
+
+def test_connect_client():
+    net = LocalNetwork()
+    client = LocalComm("test")
+
+    net.connect_client(client)
+
+    assert client == net._clients["test"]
+
+
+def test_get_client():
+    net = LocalNetwork()
+    client = LocalComm("test")
+
+    net._clients = {"test": client}
+
+    assert client == net.get_client("test")
+
+
+def test_get_clients():
+    net = LocalNetwork()
+    client = LocalComm("test")
+
+    net._clients = {"test": client}
+
+    assert [client] == net.get_clients()
+
+
+def test_get_client_names():
+    net = LocalNetwork()
+    client = LocalComm("test")
+
+    net._clients = {"test": client}
+
+    assert ["test"] == net.get_client_names()