Separate SrvFormat into ROS1 and ROS2 versions #451

src/roswire/common/srv.py

@@ -2,16 +2,19 @@
 __all__ = ("SrvFormat",)
 
 import os
-from typing import Any, Dict, List, Optional
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Generic, Optional, TypeVar
 
 import attr
 import dockerblade
 
 from .msg import MsgFormat
 
+MF = TypeVar("MF", bound=MsgFormat)
 
-@attr.s(frozen=True, auto_attribs=True, slots=True)
-class SrvFormat:
+
+@attr.s(frozen=True)
+class SrvFormat(ABC, Generic[MF]):
     """Provides an immutable definition of a given
     `ROS service format <http://wiki.ros.org/srv>`_.
 
@@ -23,23 +26,23 @@ class SrvFormat:
         The unqualified name of the service format.
     definition: str
         The plaintext contents of the associated .srv file.
-    request: Optional[MsgFormat]
+    request: Optional[MF]
         The definition of the optional request message for this service, if
         it has one.
-    response: Optional[MsgFormat]
+    response: Optional[MF]
         The definition of the optional response message for this service, if
         it has one.
     """
 
-    package: str
-    name: str
-    definition: str
-    request: Optional[MsgFormat]
-    response: Optional[MsgFormat]
+    package: str = attr.ib()
+    name: str = attr.ib()
+    definition: str = attr.ib()
+    request: Optional[MF] = attr.ib()
+    response: Optional[MF] = attr.ib()
 
-    @staticmethod
+    @classmethod
     def from_file(
-        package: str, filename: str, files: dockerblade.FileSystem
+        cls, package: str, filename: str, files: dockerblade.FileSystem
     ) -> "SrvFormat":
         """Constructs a service format from a .srv file for a given package.
 
@@ -62,56 +65,26 @@ def from_file(
         ), "service format files must end in .srv"
         name: str = os.path.basename(filename[:-4])
         contents: str = files.read(filename)
-        return SrvFormat.from_string(package, name, contents)
+        return cls.from_string(package, name, contents)
 
-    @staticmethod
-    def from_string(package: str, name: str, s: str) -> "SrvFormat":
+    @classmethod
+    @abstractmethod
+    def from_string(cls, package: str, name: str, s: str) -> "SrvFormat":
         """Constructs a service format from its definition.
 
         Raises
         ------
         ParsingError
             If the description cannot be parsed.
         """
-        req: Optional[MsgFormat] = None
-        res: Optional[MsgFormat] = None
-        name_req = f"{name}Request"
-        name_res = f"{name}Response"
-
-        sections: List[str] = [ss.strip() for ss in s.split("---")]
-        assert len(sections) < 3
-        s_req = sections[0]
-        s_res = sections[1] if len(sections) > 1 else ""
-
-        if s_req:
-            req = MsgFormat.from_string(package, name_req, s_req)
-        if s_res:
-            res = MsgFormat.from_string(package, name_res, s_res)
-
-        return SrvFormat(package, name, s, req, res)
+        ...
 
-    @staticmethod
+    @classmethod
+    @abstractmethod
     def from_dict(
-        d: Dict[str, Any], *, package: Optional[str] = None
+        cls, d: Dict[str, Any], *, package: Optional[str] = None
     ) -> "SrvFormat":
-        req: Optional[MsgFormat] = None
-        res: Optional[MsgFormat] = None
-        name: str = d["name"]
-        definition: str = d["definition"]
-        if package is None:
-            assert d["package"] is not None
-            package = d["package"]
-
-        if "request" in d:
-            req = MsgFormat.from_dict(
-                d["request"], package=package, name=f"{name}Request"
-            )
-        if "response" in d:
-            res = MsgFormat.from_dict(
-                d["response"], package=package, name=f"{name}Response"
-            )
-
-        return SrvFormat(package, name, definition, req, res)
+        ...
 
     def to_dict(self) -> Dict[str, Any]:
         d: Dict[str, Any] = {

src/roswire/ros1/__init__.py

@@ -8,4 +8,5 @@
 from .parameter import ParameterServer
 from .ros1 import ROS1
 from .service import Service, ServiceManager
+from .srv import ROS1SrvFormat
 from .state import ROS1SystemState, SystemStateProbe

src/roswire/ros1/format.py

@@ -3,22 +3,21 @@
 
 from typing import Any, Dict
 
-from . import ROS1ActionFormat
+from . import ROS1ActionFormat, ROS1SrvFormat
 from ..common import (
     FormatDatabase,
     MsgFormat,
-    SrvFormat,
 )
 
 
 class ROS1FormatDatabase(FormatDatabase[MsgFormat,
-                                        SrvFormat,
+                                        ROS1SrvFormat,
                                         ROS1ActionFormat]):
 
     @classmethod
     def from_dict(cls, d: Dict[str, Any]) -> "FormatDatabase":
         """Loads a format database from a JSON document."""
         msg = {MsgFormat.from_dict(dd) for dd in d["messages"]}
-        srv = {SrvFormat.from_dict(dd) for dd in d["services"]}
+        srv = {ROS1SrvFormat.from_dict(dd) for dd in d["services"]}
         action = {ROS1ActionFormat.from_dict(dd) for dd in d["actions"]}
         return cls(msg, srv, action)

src/roswire/ros1/package.py

@@ -10,20 +10,21 @@
 import dockerblade
 from loguru import logger
 
-from . import ROS1ActionFormat
-from ..common import MsgFormat, Package, PackageDatabase, SrvFormat
+from . import ROS1ActionFormat, ROS1SrvFormat
+from ..common import MsgFormat, Package, PackageDatabase
 from ..util import tuple_from_iterable
 
 if typing.TYPE_CHECKING:
     from .. import AppInstance
 
 
 @attr.s(frozen=True, auto_attribs=True, slots=True)
-class ROS1Package(Package[MsgFormat, SrvFormat, ROS1ActionFormat]):
+class ROS1Package(Package[MsgFormat, ROS1SrvFormat, ROS1ActionFormat]):
     name: str
     path: str
     messages: Collection[MsgFormat] = attr.ib(converter=tuple_from_iterable)
-    services: Collection[SrvFormat] = attr.ib(converter=tuple_from_iterable)
+    services: Collection[ROS1SrvFormat] = \
+        attr.ib(converter=tuple_from_iterable)
     actions: Collection[ROS1ActionFormat] = \
         attr.ib(converter=tuple_from_iterable)
 
@@ -32,7 +33,7 @@ def build(cls, path: str, app_instance: "AppInstance") -> "ROS1Package":
         """Constructs a description of a package at a given path."""
         name: str = os.path.basename(path)
         messages: List[MsgFormat] = []
-        services: List[SrvFormat] = []
+        services: List[ROS1SrvFormat] = []
         actions: List[ROS1ActionFormat] = []
         files = app_instance.files
 
@@ -51,7 +52,7 @@ def build(cls, path: str, app_instance: "AppInstance") -> "ROS1Package":
             ]
         if files.isdir(dir_srv):
             services = [
-                SrvFormat.from_file(name, f, files)
+                ROS1SrvFormat.from_file(name, f, files)
                 for f in files.listdir(dir_srv, absolute=True)
                 if f.endswith(".srv")
             ]
@@ -71,8 +72,8 @@ def from_dict(cls, d: Dict[str, Any]) -> "ROS1Package":
             MsgFormat.from_dict(dd, package=name)
             for dd in d.get("messages", [])
         ]
-        services: List[SrvFormat] = [
-            SrvFormat.from_dict(dd, package=name)
+        services: List[ROS1SrvFormat] = [
+            ROS1SrvFormat.from_dict(dd, package=name)
             for dd in d.get("services", [])
         ]
         actions: List[ROS1ActionFormat] = [

src/roswire/ros1/srv.py

@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+__all__ = ("ROS1SrvFormat", )
+
+from typing import Any, Dict, List, Optional
+
+import dockerblade
+
+from ..common import MsgFormat, SrvFormat
+
+
+class ROS1SrvFormat(SrvFormat[MsgFormat]):
+
+    @classmethod
+    def from_file(
+        cls, package: str, filename: str, files: dockerblade.FileSystem
+    ) -> "ROS1SrvFormat":
+        sf = super().from_file(package, filename, files)
+        assert isinstance(sf, ROS1SrvFormat)
+        return sf
+
+    @classmethod
+    def from_string(cls, package: str, name: str, s: str) -> "ROS1SrvFormat":
+        req: Optional[MsgFormat] = None
+        res: Optional[MsgFormat] = None
+        name_req = f"{name}Request"
+        name_res = f"{name}Response"
+
+        sections: List[str] = [ss.strip() for ss in s.split("---")]
+        assert len(sections) < 3
+        s_req = sections[0]
+        s_res = sections[1] if len(sections) > 1 else ""
+
+        if s_req:
+            req = MsgFormat.from_string(package, name_req, s_req)
+        if s_res:
+            res = MsgFormat.from_string(package, name_res, s_res)
+
+        return ROS1SrvFormat(package, name, s, req, res)
+
+    @classmethod
+    def from_dict(
+        cls, d: Dict[str, Any], *, package: Optional[str] = None
+    ) -> "ROS1SrvFormat":
+        req: Optional[MsgFormat] = None
+        res: Optional[MsgFormat] = None
+        name: str = d["name"]
+        definition: str = d["definition"]
+        if package is None:
+            assert d["package"] is not None
+            package = d["package"]
+
+        if "request" in d:
+            req = MsgFormat.from_dict(
+                d["request"], package=package, name=f"{name}Request"
+            )
+        if "response" in d:
+            res = MsgFormat.from_dict(
+                d["response"], package=package, name=f"{name}Response"
+            )
+
+        return ROS1SrvFormat(package, name, definition, req, res)

src/roswire/ros2/__init__.py

@@ -6,3 +6,4 @@
 from .package import ROS2PackageDatabase
 from .ros2 import ROS2
 from .service_manager import ROS2ServiceManager
+from .srv import ROS2SrvFormat

src/roswire/ros2/format.py

@@ -3,22 +3,22 @@
 
 from typing import Any, Dict
 
-from . import ROS2ActionFormat
+from . import ROS2ActionFormat, ROS2SrvFormat
 from ..common import (
     FormatDatabase,
     MsgFormat,
-    SrvFormat,
+
 )
 
 
 class ROS2FormatDatabase(FormatDatabase[MsgFormat,
-                                        SrvFormat,
+                                        ROS2SrvFormat,
                                         ROS2ActionFormat]):
 
     @classmethod
     def from_dict(cls, d: Dict[str, Any]) -> "FormatDatabase":
         """Loads a format database from a JSON document."""
         msg = {MsgFormat.from_dict(dd) for dd in d["messages"]}
-        srv = {SrvFormat.from_dict(dd) for dd in d["services"]}
+        srv = {ROS2SrvFormat.from_dict(dd) for dd in d["services"]}
         action = {ROS2ActionFormat.from_dict(dd) for dd in d["actions"]}
         return ROS2FormatDatabase(msg, srv, action)

src/roswire/ros2/package.py

@@ -18,11 +18,12 @@
 from loguru import logger
 from typing_extensions import Final
 
+from . import ROS2SrvFormat
 from ..common import (ActionFormat,
                       MsgFormat,
                       Package,
                       PackageDatabase,
-                      SrvFormat)
+                      )
 from ..util import tuple_from_iterable
 
 if typing.TYPE_CHECKING:
@@ -38,19 +39,20 @@
 
 
 @attr.s(frozen=True, auto_attribs=True, slots=True)
-class ROS2Package(Package[MsgFormat, SrvFormat, ActionFormat]):
+class ROS2Package(Package[MsgFormat, ROS2SrvFormat, ActionFormat]):
     name: str
     path: str
     messages: Collection[MsgFormat] = attr.ib(converter=tuple_from_iterable)
-    services: Collection[SrvFormat] = attr.ib(converter=tuple_from_iterable)
+    services: Collection[ROS2SrvFormat] = \
+        attr.ib(converter=tuple_from_iterable)
     actions: Collection[ActionFormat] = attr.ib(converter=tuple_from_iterable)
 
     @classmethod
     def build(cls, path: str, app_instance: "AppInstance") -> "ROS2Package":
         """Constructs a description of a package at a given path."""
         name: str = os.path.basename(path)
         messages: List[MsgFormat] = []
-        services: List[SrvFormat] = []
+        services: List[ROS2SrvFormat] = []
         actions: List[ActionFormat] = []
         files = app_instance.files
 
@@ -69,7 +71,7 @@ def build(cls, path: str, app_instance: "AppInstance") -> "ROS2Package":
             ]
         if files.isdir(dir_srv):
             services = [
-                SrvFormat.from_file(name, f, files)
+                ROS2SrvFormat.from_file(name, f, files)
                 for f in files.listdir(dir_srv, absolute=True)
                 if f.endswith(".srv")
             ]
@@ -89,8 +91,8 @@ def from_dict(cls, d: Dict[str, Any]) -> "ROS2Package":
             MsgFormat.from_dict(dd, package=name)
             for dd in d.get("messages", [])
         ]
-        services: List[SrvFormat] = [
-            SrvFormat.from_dict(dd, package=name)
+        services: List[ROS2SrvFormat] = [
+            ROS2SrvFormat.from_dict(dd, package=name)
             for dd in d.get("services", [])
         ]
         actions: List[ActionFormat] = [