diff --git a/src/rosdiscover/acme/acme.py b/src/rosdiscover/acme/acme.py index 082d80e2..b8906e52 100644 --- a/src/rosdiscover/acme/acme.py +++ b/src/rosdiscover/acme/acme.py @@ -5,7 +5,8 @@ The main class provided by this module is :class:`AcmeGenerator` """ -__all__ = ("AcmeGenerator", ) +__all__ = ("AcmeGenerator",) + import json import os import subprocess @@ -21,6 +22,7 @@ # Constants for Acme generation from ..interpreter.context import Provenance +from ..interpreter.summary import NodeletManagerSummary, NodeletSummary TOPIC_CONNECTOR = """ connector {conn_name} : TopicConnectorT = new TopicConnectorT extended with {{ {roles} @@ -49,6 +51,20 @@ property launchedBy = "{filename}"; }}; """ +NODELET_COMPONENT = """ component {comp_name} : ROSNodeletCompT = new ROSNodeletCompT extended with {{ + {ports} + property name = "{node_name}"; + property node_manager = "{manager_name}"; + property launchedBy = "{filename}"; + }}; +""" + +NODELET_MANAGER = """ groups {comp_name}: ROSNodeletManagerGroup = new ROSNodeletManagerGroup extended with {{ + members {{{members}}} + property name = "{node_name}"; + property launchedBy = "{filename}"; +""" + ATTACHMENT = " attachment {comp}.{port} to {conn}.{role};" SERVICE_ATTACHMENT = " attachment {qualified_port} to {conn}.{role};" SUBSCRIBER_ROLE = """ role {role_name} : ROSTopicSubscriberRoleT = new ROSTopicSubscriberRoleT; @@ -160,10 +176,10 @@ def __init__(self, self.__to_ignore = things_to_ignore if things_to_ignore is not None else [] def get_components_and_connectors(self) \ - -> Tuple[List[NodeSummary], - Dict[str, _TopicInformation], - Dict[str, _ServiceInformation], - Dict[str, _ActionInformation]]: + -> Tuple[List[NodeSummary], + Dict[str, _TopicInformation], + Dict[str, _ServiceInformation], + Dict[str, _ActionInformation]]: components: List[NodeSummary] = [] topics: Dict[str, _TopicInformation] = {} services: Dict[str, _ServiceInformation] = {} @@ -254,10 +270,13 @@ def generate_acme(self) -> str: service_conns: Dict[str, dict] = {} action_conns: Dict[str, dict] = {} attachments_to_topic: Dict[str, List[str]] = {} - for c in components: + nodelet_manager_to_nodelet: Dict[str, List[str]] = {} + for c in [c for c in components if not isinstance(c, NodeletManagerSummary)]: ports = [] comp_name = self.to_acme_name(c.name) + is_nodelet = isinstance(c, NodeletSummary) + for pub in [t for t in c.pubs if not t.implicit and not self._ignore(t.name)]: if pub.name not in attachments_to_topic: attachments_to_topic[pub.name] = [] @@ -323,12 +342,26 @@ def generate_acme(self) -> str: name, f"{comp_name}.{pname}", False) - component_template: str = NODE_COMPONENT \ - if c.provenance != Provenance.PLACEHOLDER else NODE_PLACEHOLDER_COMPONENT - comp = component_template.format(comp_name=comp_name, - ports='\n'.join(ports), - node_name=c.name, - filename=c.filename) + if is_nodelet: + assert isinstance(c, NodeletSummary) + comp = NODELET_COMPONENT.format(comp_name=comp_name, + ports='\n'.join(ports), + node_name=c.name, + filename=c.filename, + manager=c.nodelet_manager) + if c.nodelet_manager in nodelet_manager_to_nodelet: + nodelets = nodelet_manager_to_nodelet[c.nodelet_manager] + else: + nodelets = [] + nodelet_manager_to_nodelet[c.nodelet_manager] = nodelets + nodelets += comp_name + else: + component_template: str = NODE_COMPONENT \ + if c.provenance != Provenance.PLACEHOLDER else NODE_PLACEHOLDER_COMPONENT + comp = component_template.format(comp_name=comp_name, + ports='\n'.join(ports), + node_name=c.name, + filename=c.filename) component_strs.append(comp) acme = acme + "\n".join(component_strs) @@ -338,6 +371,15 @@ def generate_acme(self) -> str: self._process_services(service_conns, attachments, connector_strs) self._process_actions(action_conns, attachments, connector_strs) + group_strs: List[str] = [] + for c in [c for c in components if isinstance(c, NodeletManagerSummary)]: + comp_name = self.to_acme_name(c.name) + members = ", ".join(nodelet_manager_to_nodelet.get(c.name, [])) + group = NODELET_MANAGER.format(comp_name=comp_name, members=members, node_name=c.name, filename=c.filename) + group_strs.append(group) + + acme = acme + "\n".join(group_strs) + acme = acme + "\n".join(connector_strs) acme = acme + "\n".join(attachments) + "}" self.generate_acme_file(acme) diff --git a/src/rosdiscover/interpreter/__init__.py b/src/rosdiscover/interpreter/__init__.py index d3915b33..021a4d32 100644 --- a/src/rosdiscover/interpreter/__init__.py +++ b/src/rosdiscover/interpreter/__init__.py @@ -7,7 +7,7 @@ The main class within this module is :class:`Interpreter`, which acts as a model evaluator / virtual machine for a ROS architecture. """ -from .context import NodeContext +from .context import NodeContext, NodeletContext, NodeletManagerContext from .model import model, NodeModel from .parameter import ParameterServer from .plugin import ModelPlugin diff --git a/src/rosdiscover/interpreter/context.py b/src/rosdiscover/interpreter/context.py index 86a897af..d0874ea9 100644 --- a/src/rosdiscover/interpreter/context.py +++ b/src/rosdiscover/interpreter/context.py @@ -1,7 +1,6 @@ # -*- coding: utf-8 -*- import re -import typing -from typing import Any, List, Mapping, Optional, Set, Tuple, Union +import typing as t import attr import dockerblade @@ -14,7 +13,7 @@ from .summary import NodeSummary from ..core import Action, Service, Topic -if typing.TYPE_CHECKING: +if t.TYPE_CHECKING: from .plugin import ModelPlugin from ..recover.symbolic import SymbolicUnknown @@ -28,24 +27,23 @@ class NodeContext: kind: str package: str args: str - remappings: Mapping[str, str] + remappings: t.Mapping[str, str] launch_filename: str app: "AppInstance" = attr.ib(repr=False) _params: ParameterServer = attr.ib(repr=False) _files: dockerblade.files.FileSystem = attr.ib(repr=False) - _nodelet: bool = attr.ib(default=False, repr=False) _provenance: "Provenance" = attr.ib(default=Provenance.UNKNOWN, repr=False) - _uses: Set[Service] = attr.ib(factory=set, repr=False) - _provides: Set[Service] = attr.ib(factory=set, repr=False) - _subs: Set[Topic] = attr.ib(factory=set, repr=False) - _pubs: Set[Topic] = attr.ib(factory=set, repr=False) - _action_servers: Set[Action] = attr.ib(factory=set, repr=False) - _action_clients: Set[Action] = attr.ib(factory=set, repr=False) + _uses: t.Set[Service] = attr.ib(factory=set, repr=False) + _provides: t.Set[Service] = attr.ib(factory=set, repr=False) + _subs: t.Set[Topic] = attr.ib(factory=set, repr=False) + _pubs: t.Set[Topic] = attr.ib(factory=set, repr=False) + _action_servers: t.Set[Action] = attr.ib(factory=set, repr=False) + _action_clients: t.Set[Action] = attr.ib(factory=set, repr=False) # The tuple is (name, dynamic) where name is the name of the parameter # and dynamic is whether the node reacts to updates to the parameter via reconfigure - _reads: Set[Tuple[str, bool]] = attr.ib(factory=set, repr=False) - _writes: Set[str] = attr.ib(factory=set, repr=False) - _plugins: List['ModelPlugin'] = attr.ib(factory=list) + _reads: t.Set[t.Tuple[str, bool]] = attr.ib(factory=set, repr=False) + _writes: t.Set[str] = attr.ib(factory=set, repr=False) + _plugins: t.List['ModelPlugin'] = attr.ib(factory=list) def merge(self, context: 'NodeContext') -> None: self._params.update(context._params) @@ -95,7 +93,6 @@ def summarise(self) -> NodeSummary: namespace=self.namespace, kind=self.kind, package=self.package, - nodelet=self._nodelet, provenance=self._provenance, reads=self._reads, writes=self._writes, @@ -121,7 +118,7 @@ def _resolve_without_remapping(self, name: str) -> str: else: return rosname.namespace_join(self.namespace, name) - def resolve(self, name: Union[str, 'SymbolicUnknown']) -> str: + def resolve(self, name: t.Union[str, 'SymbolicUnknown']) -> str: """Resolves a given name within the context of this node. Returns @@ -140,26 +137,26 @@ def resolve(self, name: Union[str, 'SymbolicUnknown']) -> str: logger.warning(f"Unable to resolve unknown name in NodeContext [{self.name}]") return UNKNOWN_NAME - def _name_str(self, name: Union[str, 'SymbolicUnknown']) -> str: + def _name_str(self, name: t.Union[str, 'SymbolicUnknown']) -> str: if isinstance(name, str): return name return "Unknown Symbol" - def provide(self, service: Union[str, 'SymbolicUnknown'], fmt: str) -> None: + def provide(self, service: t.Union[str, 'SymbolicUnknown'], fmt: str) -> None: """Instructs the node to provide a service.""" logger.debug(f"node [{self.name}] provides service [{self._name_str(service)}] " f"using format [{fmt}]") service_name_full = self.resolve(service) self._provides.add(Service(name=service_name_full, format=fmt)) - def use(self, service: Union[str, 'SymbolicUnknown'], fmt: str) -> None: + def use(self, service: t.Union[str, 'SymbolicUnknown'], fmt: str) -> None: """Instructs the node to use a given service.""" logger.debug(f"node [{self.name}] uses a service [{self._name_str(service)}] " f"with format [{fmt}]") service_name_full = self.resolve(service) self._uses.add(Service(name=service_name_full, format=fmt)) - def sub(self, topic_name: Union[str, 'SymbolicUnknown'], fmt: str, implicit: bool = False) -> None: + def sub(self, topic_name: t.Union[str, 'SymbolicUnknown'], fmt: str, implicit: bool = False) -> None: """Subscribes the node to a given topic. Parameters @@ -177,7 +174,7 @@ def sub(self, topic_name: Union[str, 'SymbolicUnknown'], fmt: str, implicit: boo f"[{self._name_str(topic_name)}] with format [{fmt}]") self._subs.add(Topic(name=topic_name_full, format=fmt, implicit=implicit)) - def pub(self, topic_name: Union[str, 'SymbolicUnknown'], fmt: str, implicit: bool = False) -> None: + def pub(self, topic_name: t.Union[str, 'SymbolicUnknown'], fmt: str, implicit: bool = False) -> None: """Instructs the node to publish to a given topic. Parameters @@ -196,17 +193,17 @@ def pub(self, topic_name: Union[str, 'SymbolicUnknown'], fmt: str, implicit: boo self._pubs.add(Topic(name=topic_name_full, format=fmt, implicit=implicit)) def read(self, - param: Union[str, 'SymbolicUnknown'], - default: Optional[Any] = None, + param: t.Union[str, 'SymbolicUnknown'], + default: t.Optional[t.Any] = None, dynamic: bool = False - ) -> Any: + ) -> t.Any: """Obtains the value of a given parameter from the parameter server.""" logger.debug(f"node [{self.name}] reads parameter [{self._name_str(param)}]") param = self.resolve(param) self._reads.add((param, dynamic)) return self._params.get(param, default) - def write(self, param: Union[str, 'SymbolicUnknown'], val: Any) -> None: + def write(self, param: t.Union[str, 'SymbolicUnknown'], val: t.Any) -> None: logger.debug(f"node [{self.name}] writes [{val}] to " f"parameter [{self._name_str(param)}]") param = self.resolve(param) @@ -214,16 +211,16 @@ def write(self, param: Union[str, 'SymbolicUnknown'], val: Any) -> None: self._params[param] = val # FIXME we _may_ want to record this interaction in our summary - def has_param(self, param: Union[str, 'SymbolicUnknown']) -> bool: + def has_param(self, param: t.Union[str, 'SymbolicUnknown']) -> bool: """Determines whether a given parameter has been defined.""" logger.debug(f"node [{self.name}] checks for existence of parameter [{param}]") param = self.resolve(param) return param in self._params - def delete_param(self, param: Union[str, 'SymbolicUnknown']) -> None: + def delete_param(self, param: t.Union[str, 'SymbolicUnknown']) -> None: raise NotImplementedError("parameter deletion is not implemented") - def read_file(self, fn: Union[str, 'SymbolicUnknown']) -> str: + def read_file(self, fn: t.Union[str, 'SymbolicUnknown']) -> str: """Reads the contents of a text file.""" if isinstance(fn, str): if not self._files.exists(fn): @@ -234,11 +231,11 @@ def read_file(self, fn: Union[str, 'SymbolicUnknown']) -> str: logger.warning(f"Unable to resolve unknown parameter filename in NodeContext [{self.name}]") return UNKNOWN_NAME - def parameter_keys(self, prefix: str) -> typing.Iterable[str]: + def parameter_keys(self, prefix: str) -> t.Iterable[str]: prefix = self.resolve(prefix) return (key for key in self._params.keys() if key.startswith(prefix)) - def action_server(self, ns: Union[str, 'SymbolicUnknown'], fmt: str) -> None: + def action_server(self, ns: t.Union[str, 'SymbolicUnknown'], fmt: str) -> None: """Creates a new action server. Parameters @@ -262,7 +259,7 @@ def action_server(self, ns: Union[str, 'SymbolicUnknown'], fmt: str) -> None: self.pub(f'{ns}/feedback', f'{fmt}Feedback', implicit=True) self.pub(f'{ns}/result', f'{fmt}Result', implicit=True) - def action_client(self, ns: Union[str, 'SymbolicUnknown'], fmt: str) -> None: + def action_client(self, ns: t.Union[str, 'SymbolicUnknown'], fmt: str) -> None: """Creates a new action client. Parameters @@ -294,29 +291,11 @@ def actions_have_topics(self): else: return distribution < ROSDistribution.FOXY - def load_nodelet(self, nodelet_context: 'NodeContext'): - self.merge(nodelet_context) - # In the recovered architecture, the nodelets themselves don't - # report what they publish etc. - # TODO: Fix this when we have NodeletManagerContexts and NodeletContexts - nodelet_context._params.clear() - nodelet_context._uses.clear() - nodelet_context._provides.clear() - nodelet_context._subs.clear() - nodelet_context._pubs.clear() - nodelet_context._action_servers.clear() - nodelet_context._action_clients.clear() - nodelet_context._reads.clear() - nodelet_context._writes.clear() - def load_plugin(self, plugin: 'ModelPlugin') -> None: """Loads a given dynamic plugin.""" logger.debug(f'loading plugin in node [{self.name}]: {plugin}') self._plugins.append(plugin) - def mark_nodelet(self) -> None: - self._nodelet = True - def mark_placeholder(self) -> None: self._provenance = Provenance.PLACEHOLDER @@ -325,3 +304,24 @@ def mark_handwritten(self) -> None: def mark_recovered(self) -> None: self._provenance = Provenance.RECOVERED + + +@attr.s(slots=True, auto_attribs=True) +class NodeletManagerContext(NodeContext): + _nodelets: t.Collection['NodeletContext'] = attr.ib(factory=set, repr=False) + + def load_nodelet(self, nodelet_context: 'NodeletContext') -> None: + self.merge(nodelet_context) + # In the recovered architecture, the nodelets themselves don't + # report what they publish etc. + nodelets = set(self._nodelets) + nodelets.add(nodelet_context) + self.__setattr__("_nodelets", nodelets) + + +@attr.s(slots=True, auto_attribs=True) +class NodeletContext(NodeContext): + _nodelet_manager: 'NodeletManagerContext' = attr.ib(default=None) + + def set_nodelet_manager(self, manager: 'NodeletManagerContext') -> None: + self._nodelet_manager = manager diff --git a/src/rosdiscover/interpreter/interpreter.py b/src/rosdiscover/interpreter/interpreter.py index 50ab9d5e..1901be55 100644 --- a/src/rosdiscover/interpreter/interpreter.py +++ b/src/rosdiscover/interpreter/interpreter.py @@ -11,7 +11,7 @@ from roswire.ros1.launch.reader import ROS1LaunchFileReader from roswire.ros2.launch.reader import ROS2LaunchFileReader -from .context import NodeContext +from .context import NodeContext, NodeletContext, NodeletManagerContext from .model import PlaceholderModel from .summary import SystemSummary from .parameter import ParameterServer @@ -27,6 +27,7 @@ class Interpreter: params: ParameterServer The simulated parameter server for this interpreter. """ + @classmethod @contextlib.contextmanager def for_config(cls, @@ -133,16 +134,16 @@ def _create_nodelet_manager(self, remappings: t.Mapping[str, str]) -> None: """Creates a nodelet manager with a given name.""" logger.info(f'launched nodelet manager: {manager} as {name}') - ctx = NodeContext(name=name, - namespace=namespace, - kind="nodelet", - package="nodelet", - launch_filename=launch_filename, - remappings=remappings, - files=self._app.files, - params=self.params, - app=self._app, - args='') + ctx = NodeletManagerContext(name=name, + namespace=namespace, + kind="nodelet", + package="nodelet", + launch_filename=launch_filename, + remappings=remappings, + files=self._app.files, + params=self.params, + app=self._app, + args='') self.nodes[ctx.fullname] = ctx def _load_nodelet(self, @@ -186,15 +187,31 @@ def _load_nodelet(self, if manager: logger.info(f'launching nodelet [{name}] ' f'inside manager [{manager}] from {launch_filename}') - - return self._load(pkg=pkg, - nodetype=nodetype, - name=name, - namespace=namespace, - launch_filename=launch_filename, - remappings=remappings, - args=f'manager {manager}' - ) + # This is being loaded into an existing manager, so find that as the context + if namespace: + manager_name = f"{namespace}/{manager}" + manager_name = manager.replace('//', '/') + if manager in self.nodes: + manager_context = self.nodes[manager_name] + elif f"/{manager_name}" in self.nodes: + manager_context = self.nodes[f"/{manager_name}"] + else: + raise ValueError(f"The nodelet manager {manager_name} has not been launched") + assert isinstance(manager_context, NodeletManagerContext) + # Create a context for the nodelet + ctx = NodeletContext(name=name, + namespace=namespace, + kind=nodetype, + package=pkg, + args='', + launch_filename=launch_filename, + remappings=remappings, + files=self._app.files, + params=self.params, + app=self._app) + model = self.fetch_model(name, nodetype, pkg) + model.eval(ctx) + manager_context.load_nodelet(ctx) else: logger.info(f'launching standalone nodelet [{name}]') return self._load(pkg=pkg, @@ -276,6 +293,22 @@ def _load(self, if remappings: logger.info(f"using remappings: {remappings}") + model = self.fetch_model(name, nodetype, pkg) + + ctx = NodeContext(name=name, + namespace=namespace, + kind=nodetype, + package=pkg, + args=args, + launch_filename=launch_filename, + remappings=remappings, + files=self._app.files, + params=self.params, + app=self._app) + self.nodes[ctx.fullname] = ctx + model.eval(ctx) + + def fetch_model(self, name, nodetype, pkg): try: model = self.models.fetch(pkg, nodetype) # This is to handle nodelet strangness @@ -287,46 +320,4 @@ def _load(self, f"in package [{pkg}]") logger.warning(m) raise - if args.startswith('manager'): - # This is being loaded into an existing manager, so find that as the context - manager_name = args.split(" ")[1] - if namespace: - manager_name = f"{namespace}/{manager_name}" - manager_name = manager_name.replace('//', '/') - if manager_name in self.nodes: - manager_context = self.nodes[manager_name] - elif f"/{manager_name}" in self.nodes: - manager_context = self.nodes[f"/{manager_name}"] - else: - raise ValueError(f"The nodelet manager {manager_name} has not been launched") - # Create a context for the nodelet - ctx = NodeContext(name=name, - namespace=namespace, - kind=nodetype, - package=pkg, - args=args, - launch_filename=launch_filename, - remappings=remappings, - files=self._app.files, - params=self.params, - app=self._app) - model.eval(ctx) - manager_context.load_nodelet(ctx) - # Place the nodelet as a node, which is observed - # TODO: This needs to be rethought -- we should have a separate NodeletManagerContext - # that con contain NodeletContexts. This would better map the NodeletManager/ - # Nodelet mapping, and would actually contain traceability between topics - self.nodes[ctx.fullname] = ctx - else: - ctx = NodeContext(name=name, - namespace=namespace, - kind=nodetype, - package=pkg, - args=args, - launch_filename=launch_filename, - remappings=remappings, - files=self._app.files, - params=self.params, - app=self._app) - self.nodes[ctx.fullname] = ctx - model.eval(ctx) + return model diff --git a/src/rosdiscover/interpreter/summary.py b/src/rosdiscover/interpreter/summary.py index 5e57a6a7..32220720 100644 --- a/src/rosdiscover/interpreter/summary.py +++ b/src/rosdiscover/interpreter/summary.py @@ -18,7 +18,6 @@ class NodeSummary: namespace: str kind: str package: str - nodelet: bool filename: str # Provenance indicates where the model comes from. # PLACEHOLDER indicates whether the node was not really discovered, but @@ -31,43 +30,45 @@ class NodeSummary: # model. # RECOVERED indicates that the node was recovered through static analysis provenance: Provenance - pubs: Collection[Topic] - subs: Collection[Topic] + _pubs: Collection[Topic] + _subs: Collection[Topic] # The tuple is (name, dynamic) where name is the name of the parameter # and dynamic is whether the node reacts to updates to the parameter via reconfigure - reads: Collection[Tuple[str, bool]] - writes: Collection[str] - uses: Collection[Service] - provides: Collection[Service] - action_servers: Collection[Action] - action_clients: Collection[Action] + _reads: Collection[Tuple[str, bool]] + _writes: Collection[str] + _uses: Collection[Service] + _provides: Collection[Service] + _action_servers: Collection[Action] + _action_clients: Collection[Action] def __attrs_post_init__(self) -> None: - object.__setattr__(self, 'pubs', frozenset(self.pubs)) - object.__setattr__(self, 'subs', frozenset(self.subs)) - object.__setattr__(self, 'reads', frozenset(self.reads)) - object.__setattr__(self, 'writes', frozenset(self.writes)) - object.__setattr__(self, 'uses', frozenset(self.uses)) - object.__setattr__(self, 'provides', frozenset(self.provides)) - object.__setattr__(self, 'action_servers', frozenset(self.action_servers)) - object.__setattr__(self, 'action_clients', frozenset(self.action_clients)) + object.__setattr__(self, '_pubs', frozenset(self._pubs)) + object.__setattr__(self, '_subs', frozenset(self._subs)) + object.__setattr__(self, '_reads', frozenset(self._reads)) + object.__setattr__(self, '_writes', frozenset(self._writes)) + object.__setattr__(self, '_uses', frozenset(self._uses)) + object.__setattr__(self, '_provides', frozenset(self._provides)) + object.__setattr__(self, '_action_servers', frozenset(self._action_servers)) + object.__setattr__(self, '_action_clients', frozenset(self._action_clients)) + + @classmethod + def _merge_collections(cls, s1: Collection[Any], s2: Collection[Any]) -> Collection[Any]: + s: Set[Any] = set() + s.update(s1) + s.update(s2) + return s @classmethod def merge(cls, lhs: 'NodeSummary', rhs: 'NodeSummary') -> 'NodeSummary': - def merge_collections(s1: Collection[Any], s2: Collection[Any]) -> Collection[Any]: - s: Set[Any] = set() - s.update(s1) - s.update(s2) - return s - - reads = merge_collections(lhs.reads, rhs.reads) - writes = merge_collections(lhs.writes, rhs.writes) - pubs = merge_collections(lhs.pubs, rhs.pubs) - subs = merge_collections(lhs.subs, rhs.subs) - uses = merge_collections(lhs.uses, rhs.uses) - provides = merge_collections(lhs.provides, rhs.provides) - actions_servers = merge_collections(lhs.action_servers, rhs.action_servers) - action_clients = merge_collections(lhs.action_clients, rhs.action_clients) + + reads = cls._merge_collections(lhs._reads, rhs._reads) + writes = cls._merge_collections(lhs._writes, rhs._writes) + pubs = cls._merge_collections(lhs._pubs, rhs._pubs) + subs = cls._merge_collections(lhs._subs, rhs._subs) + uses = cls._merge_collections(lhs._uses, rhs._uses) + provides = cls._merge_collections(lhs._provides, rhs._provides) + actions_servers = cls._merge_collections(lhs._action_servers, rhs._action_servers) + action_clients = cls._merge_collections(lhs._action_clients, rhs._action_clients) if lhs.name != rhs.name and lhs.name: logger.warning(f"Merging two nodes that are named differently: {lhs.name} & {rhs.name}") @@ -103,7 +104,6 @@ def merge_collections(s1: Collection[Any], s2: Collection[Any]) -> Collection[An namespace=namespace, kind=kind, package=package, - nodelet=lhs.nodelet, filename=filename, provenance=provenance, reads=reads, @@ -117,23 +117,22 @@ def merge_collections(s1: Collection[Any], s2: Collection[Any]) -> Collection[An ) def to_dict(self) -> Dict[str, Any]: - pubs = [t.to_dict() for t in self.pubs] - subs = [t.to_dict() for t in self.subs] - provides = [s.to_dict() for s in self.provides] - uses = [s.to_dict() for s in self.uses] - action_servers = [a.to_dict() for a in self.action_servers] - action_clients = [a.to_dict() for a in self.action_clients] - reads = [{'name': n, 'dynamic': d} for (n, d) in self.reads] + pubs = [t.to_dict() for t in self._pubs] + subs = [t.to_dict() for t in self._subs] + provides = [s.to_dict() for s in self._provides] + uses = [s.to_dict() for s in self._uses] + action_servers = [a.to_dict() for a in self._action_servers] + action_clients = [a.to_dict() for a in self._action_clients] + reads = [{'name': n, 'dynamic': d} for (n, d) in self._reads] return {'name': self.name, 'fullname': self.fullname, 'namespace': self.namespace, 'kind': self.kind, 'package': self.package, - 'nodelet': self.nodelet, 'filename': self.filename, 'provenance': self.provenance.value, 'reads': reads, - 'writes': list(self.writes), + 'writes': list(self._writes), 'provides': provides, 'uses': uses, 'action-servers': action_servers, @@ -148,7 +147,6 @@ def from_dict(cls, dict: Dict[str, Any]) -> 'NodeSummary': namepsace = dict.get('namespace', '') kind = dict.get('kind', '') package = dict.get('package', '') - nodelet = dict.get('nodelet', False) filename = dict.get('filename', '') provenance = Provenance(dict.get('provenance', Provenance.PLACEHOLDER)) reads = [(p['name'], p['dynamic']) for p in dict.get('reads', [])] @@ -170,7 +168,6 @@ def from_dict(cls, dict: Dict[str, Any]) -> 'NodeSummary': namespace=namepsace, kind=kind, package=package, - nodelet=nodelet, filename=filename, provenance=provenance, reads=reads, @@ -182,6 +179,230 @@ def from_dict(cls, dict: Dict[str, Any]) -> 'NodeSummary': action_servers=action_servers, action_clients=action_clients) + @property + def pubs(self) -> Collection[Topic]: + return self._pubs + + @property + def subs(self) -> Collection[Topic]: + return self._subs + + @property + def provides(self) -> Collection[Service]: + return self._provides + + @property + def uses(self) -> Collection[Service]: + return self._uses + + @property + def reads(self) -> Collection[Tuple[str, bool]]: + return self._reads + + @property + def writes(self) -> Collection[str]: + return self._writes + + @property + def action_clients(self) -> Collection[Action]: + return self._action_clients + + @property + def action_servers(self) -> Collection[Action]: + return self._action_servers + + +MERGE_NODELETS_TO_NODELET_MANAGERS = True + + +@attr.s(frozen=True, slots=True, auto_attribs=True) +class NodeletSummary(NodeSummary): + """Summarises the architectural effects of a given node.""" + nodelet_manager: str + + @classmethod + def merge(cls, lhs: 'NodeSummary', rhs: 'NodeSummary') -> 'NodeletSummary': + assert isinstance(lhs, NodeletSummary) + assert isinstance(rhs, NodeletSummary) + nc = NodeSummary.merge(lhs, rhs) + + if lhs.nodelet_manager != rhs.nodelet_manager: + logger.warning(f"{lhs.fullname} nodelet manager {rhs.fullname} retaining original nodelet manager" + f" {lhs.fullname}") + nodelet_manager = lhs.nodelet_manager if lhs.nodelet_manager else rhs.nodelet_manager + return NodeletSummary( + name=nc.name, + fullname=nc.fullname, + namespace=nc.namespace, + kind=nc.kind, + package=nc.package, + filename=nc.filename, + provenance=nc.provenance, + reads=nc.reads, + writes=nc.writes, + pubs=nc.pubs, + subs=nc.subs, + provides=nc.provides, + uses=nc.uses, + action_servers=nc.action_servers, + action_clients=nc.action_clients, + nodelet_manager=nodelet_manager + ) + + def to_dict(self) -> Dict[str, Any]: + dict_ = super().to_dict() + dict_['nodelet_manager'] = self.nodelet_manager + dict_['nodekind'] = 'nodelet' + return dict_ + + @classmethod + def from_dict(cls, dict_: Dict[str, Any]) -> 'NodeletSummary': + nc = NodeSummary.from_dict(dict_) + return NodeletSummary( + name=nc.name, + fullname=nc.fullname, + namespace=nc.namespace, + kind=nc.kind, + package=nc.package, + filename=nc.filename, + provenance=nc.provenance, + reads=nc.reads, + writes=nc.writes, + pubs=nc.pubs, + subs=nc.subs, + provides=nc.provides, + uses=nc.uses, + action_servers=nc.action_servers, + action_clients=nc.action_clients, + nodelet_manager=dict_.get('nodelet_manager', '') + ) + + @property + def pubs(self) -> Collection[Topic]: + return self._pubs if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def subs(self) -> Collection[Topic]: + return self._subs if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def provides(self) -> Collection[Service]: + return self._provides if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def uses(self) -> Collection[Service]: + return self._uses if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def reads(self) -> Collection[Tuple[str, bool]]: + return self._reads if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def writes(self) -> Collection[str]: + return self._writes if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def action_clients(self) -> Collection[Action]: + return self._action_clients if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def action_servers(self) -> Collection[Action]: + return self._action_servers if not MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + +@attr.s(frozen=True, slots=True, auto_attribs=True) +class NodeletManagerSummary(NodeSummary): + nodelets: Collection[NodeletSummary] + + def __attrs_post_init__(self) -> None: + super().__attrs_post_init__() + object.__setattr__(self, 'nodelets', frozenset(self.writes)) + + @classmethod + def merge(cls, lhs: 'NodeSummary', rhs: 'NodeSummary') -> 'NodeletManagerSummary': + assert isinstance(lhs, NodeletManagerSummary) + assert isinstance(rhs, NodeletManagerSummary) + nc = NodeSummary.merge(lhs, rhs) + nodelets = NodeSummary._merge_collections(lhs.nodelets, rhs.nodelets) + return NodeletManagerSummary( + name=nc.name, + fullname=nc.fullname, + namespace=nc.namespace, + kind=nc.kind, + package=nc.package, + filename=nc.filename, + provenance=nc.provenance, + reads=nc.reads, + writes=nc.writes, + pubs=nc.pubs, + subs=nc.subs, + provides=nc.provides, + uses=nc.uses, + action_servers=nc.action_servers, + action_clients=nc.action_clients, + nodelets=nodelets + ) + + @classmethod + def from_dict(cls, dict_: Dict[str, Any]) -> 'NodeletManagerSummary': + nc = NodeSummary.from_dict(dict_) + return NodeletManagerSummary( + name=nc.name, + fullname=nc.fullname, + namespace=nc.namespace, + kind=nc.kind, + package=nc.package, + filename=nc.filename, + provenance=nc.provenance, + reads=nc.reads, + writes=nc.writes, + pubs=nc.pubs, + subs=nc.subs, + provides=nc.provides, + uses=nc.uses, + action_servers=nc.action_servers, + action_clients=nc.action_clients, + nodelets=list(NodeletSummary.from_dict(n) for n in dict_.get('nodelets', [])) + ) + + def to_dict(self) -> Dict[str, Any]: + dict_ = super().to_dict() + dict_['nodekind'] = 'nodelet_manager' + dict_['nodelets'] = list(n.to_dict() for n in self.nodelets) + return dict_ + + @property + def pubs(self) -> Collection[Topic]: + return list(p for n in self.nodelets for p in n.pubs) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def subs(self) -> Collection[Topic]: + return list(p for n in self.nodelets for p in n.subs) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def provides(self) -> Collection[Service]: + return list(p for n in self.nodelets for p in n.provides) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def uses(self) -> Collection[Service]: + return list(p for n in self.nodelets for p in n.uses) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def reads(self) -> Collection[Tuple[str, bool]]: + return list(p for n in self.nodelets for p in n.reads) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def writes(self) -> Collection[str]: + return list(p for n in self.nodelets for p in n.writes) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def action_clients(self) -> Collection[Action]: + return list(p for n in self.nodelets for p in n.action_clients) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + + @property + def action_servers(self) -> Collection[Action]: + return list(p for n in self.nodelets for p in n.action_servers) if MERGE_NODELETS_TO_NODELET_MANAGERS else {} + @attr.s(frozen=True, slots=True, auto_attribs=True) class SystemSummary(Mapping[str, NodeSummary]): @@ -191,20 +412,46 @@ class SystemSummary(Mapping[str, NodeSummary]): _node_to_summary: Mapping[str, NodeSummary] def __len__(self) -> int: - return len(self._node_to_summary) + length = len(self._node_to_summary) + # Because nodelet managers contain nodelets include them too + for n in self._node_to_summary.values(): + if isinstance(n, NodeletManagerSummary): + assert isinstance(n, NodeletManagerSummary) + length += len(n.nodelets) + return length def __iter__(self) -> Iterator[str]: - yield from self._node_to_summary + # Because nodelet managers contain nodelets include them too + for key, item in self._node_to_summary.items(): + yield key + if isinstance(item, NodeletManagerSummary): + assert isinstance(item, NodeletManagerSummary) + for nodelet in item.nodelets: + yield nodelet.fullname def __getitem__(self, name: str) -> NodeSummary: - return self._node_to_summary[name] + item = self._node_to_summary.get(name, None) + # Because nodelet managers contain nodelets include them too + if not item: + items = [n for nm in self._node_to_summary.values() if isinstance(nm, NodeletManagerSummary) for + n in nm.nodelets] + if len(items) == 1: + item = items[0] + else: + raise KeyError(f'More than one nodelet with name {name}') + if item: + return item + raise KeyError(f"'{name}' not found in summary") def to_dict(self) -> List[Dict[str, Any]]: return [n.to_dict() for n in self.values()] @classmethod def from_dict(cls, arr: Collection[Any]) -> 'SystemSummary': - summaries = [NodeSummary.from_dict(s) for s in arr] + summaries = [NodeSummary.from_dict(s) for s in arr if 'nodekind' not in s] + summaries.extend( + NodeletManagerSummary.from_dict(s) for s in arr if s.get('nodekind', None) == 'nodelet_manager' + ) return SystemSummary(node_to_summary={summary.name: summary for summary in summaries}) @property @@ -220,7 +467,26 @@ def merge(cls, lhs: 'SystemSummary', rhs: 'SystemSummary') -> 'SystemSummary': if key not in rhs: node_summaries[key] = summary else: - node_summaries[key] = NodeSummary.merge(summary, rhs[key]) + rhs_sum = rhs[key] + # if isinstance(summary, NodeletSummary): + # if isinstance(rhs, NodeletSummary): + # assert isinstance(summary, NodeletSummary) + # assert isinstance(rhs, NodeletSummary) + # node_summaries[key] = NodeletSummary.merge(summary, rhs) + # else: + # logger.error(f"{rhs.fullname} is not a nodelet in rhs") + # node_summaries[key] = summary + + if isinstance(summary, NodeletManagerSummary): + if isinstance(summary, NodeletManagerSummary): + assert isinstance(summary, NodeletManagerSummary) + assert isinstance(rhs, NodeletManagerSummary) + node_summaries[key] = NodeletManagerSummary.merge(summary, rhs_sum) + else: + logger.error(f"{rhs_sum.fullname} is not a nodelet manager in rhs") + node_summaries[key] = summary + else: + node_summaries[key] = NodeSummary.merge(summary, rhs_sum) for key, summary in rhs.items(): if key not in lhs: node_summaries[key] = summary diff --git a/src/rosdiscover/models/depth_image_proc_nodelets.py b/src/rosdiscover/models/depth_image_proc_nodelets.py index 945862f9..5d0c6fa3 100644 --- a/src/rosdiscover/models/depth_image_proc_nodelets.py +++ b/src/rosdiscover/models/depth_image_proc_nodelets.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -from ..interpreter import model +from ..interpreter import model, NodeletContext POINTCLOUD2 = 'sensor_msgs/PointCloud2' MSGS_CAMERA_INFO = 'sensor_msgs/CameraInfo' @@ -9,7 +9,7 @@ @model(DEPTH_IMAGE_PROC_PKG, 'convert_metric') def convert_metric(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{DEPTH_IMAGE_PROC_PKG}/convert_metric not being loaded as a nodelet' c.sub('~image_raw', IMAGE_TOPIC_TYPE) c.pub('~image', IMAGE_TOPIC_TYPE) @@ -17,7 +17,7 @@ def convert_metric(c): @model(DEPTH_IMAGE_PROC_PKG, 'disparity') def disparity(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{DEPTH_IMAGE_PROC_PKG}/disparity not being loaded as a nodelet' c.sub('~left/image_rect', IMAGE_TOPIC_TYPE) c.sub('~right/camera_info', MSGS_CAMERA_INFO) @@ -30,7 +30,7 @@ def disparity(c): @model(DEPTH_IMAGE_PROC_PKG, 'point_cloud_xyz') def point_cloud_xyz(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{DEPTH_IMAGE_PROC_PKG}/point_cloud_xyz not being loaded as a nodelet' c.sub('~camera_info', MSGS_CAMERA_INFO) c.sub('~image_rect', IMAGE_TOPIC_TYPE) @@ -42,7 +42,7 @@ def point_cloud_xyz(c): @model(DEPTH_IMAGE_PROC_PKG, 'point_cloud_xyzrgb') def point_cloud_xyzrgb(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{DEPTH_IMAGE_PROC_PKG/point_cloud_xyzrgb} not being loaded as a nodelet' c.sub('~rgb/camera_info', MSGS_CAMERA_INFO) c.sub('~rgb/image_rect_color', IMAGE_TOPIC_TYPE) @@ -55,7 +55,7 @@ def point_cloud_xyzrgb(c): @model(DEPTH_IMAGE_PROC_PKG, 'register') def register(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{DEPTH_IMAGE_PROC_PKG}/register not being loaded as a nodelet' c.sub('~rgb/camera_info', MSGS_CAMERA_INFO) c.sub('~depth/camera_info', MSGS_CAMERA_INFO) diff --git a/src/rosdiscover/models/image_proc_nodelets.py b/src/rosdiscover/models/image_proc_nodelets.py index 2626e429..94e999b1 100644 --- a/src/rosdiscover/models/image_proc_nodelets.py +++ b/src/rosdiscover/models/image_proc_nodelets.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -from ..interpreter import model +from ..interpreter import model, NodeletContext IMAGE_TOPIC_TYPE = 'sensor_msgs/Image' IMAGE_PROC_PKG = 'image_proc' @@ -7,6 +7,8 @@ @model(IMAGE_PROC_PKG, "image_proc") def image_proc(c): + assert isinstance(c, NodeletContext), f'{IMAGE_PROC_PKG}/image_proc not being loaded as a nodelet' + c.sub('image_raw', IMAGE_TOPIC_TYPE) c.sub('camera_info', 'sensor_msgs/CameraInfo') @@ -20,7 +22,7 @@ def image_proc(c): @model(IMAGE_PROC_PKG, "debayer") def debayer(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{IMAGE_PROC_PKG}/debayer not being loaded as a nodelet' c.pub('image_mono', IMAGE_TOPIC_TYPE) c.pub('image_color', IMAGE_TOPIC_TYPE) @@ -31,7 +33,7 @@ def debayer(c): @model(IMAGE_PROC_PKG, 'rectify') def rectify(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{IMAGE_PROC_PKG}/rectify not being loaded as a nodelet' c.sub('image_mono', IMAGE_TOPIC_TYPE) c.sub('camera_info', 'sensor_msgs/CameraInfo') @@ -44,7 +46,7 @@ def rectify(c): @model(IMAGE_PROC_PKG, 'crop_decimate') def crop_decimate(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{IMAGE_PROC_PKG}/crop_decimate not being loaded as a nodelet' c.sub('camera/image_raw', IMAGE_TOPIC_TYPE) c.sub('camera/camera_info', 'sensor_msgs/CameraInfo') @@ -64,7 +66,7 @@ def crop_decimate(c): @model(IMAGE_PROC_PKG, 'resize') def resize(c): - c.mark_nodelet() + assert isinstance(c, NodeletContext), f'{IMAGE_PROC_PKG}/resize not being loaded as a nodelet' c.sub('image', IMAGE_TOPIC_TYPE) c.sub('camera_info', 'sensor_msgs/CameraInfo') diff --git a/src/rosdiscover/models/pointgrey_camera_driver.py b/src/rosdiscover/models/pointgrey_camera_driver.py index c7aa65ce..afa34142 100644 --- a/src/rosdiscover/models/pointgrey_camera_driver.py +++ b/src/rosdiscover/models/pointgrey_camera_driver.py @@ -1,11 +1,10 @@ # -*- coding: utf-8 -*- -from ..interpreter import model, NodeContext +from ..interpreter import model, NodeletContext @model("pointgrey_camera_driver", "PointGreyCameraNodelet") -def pointgrey_camera_driver(c: NodeContext) -> None: - c.mark_nodelet() - +def pointgrey_camera_driver(c: NodeletContext) -> None: + assert isinstance(c, NodeletContext), 'pointgrey_camera_driver not being loaded as a nodelet' # How to derive dynamic parameters from: # https://github.com/ros-drivers/pointgrey_camera_driver/blob/1c71a654bea94f59396361cd735ef718f8f07011/pointgrey_camera_driver/src/nodelet.cpp#L270 c.read('serial', 0)