event handlers: Allow to match the target action with a callable and not only with an object instance

launch/launch/event_handlers/on_action_event_base.py

@@ -0,0 +1,130 @@
+# Copyright 2021 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Module for OnActionEventBase class."""
+
+import collections.abc
+from typing import Callable
+from typing import List # noqa
+from typing import Optional
+from typing import Text
+from typing import Type
+from typing import TYPE_CHECKING
+from typing import Union
+
+from ..event import Event
+from ..event_handler import BaseEventHandler
+from ..launch_context import LaunchContext
+from ..launch_description_entity import LaunchDescriptionEntity
+from ..some_actions_type import SomeActionsType
+
+if TYPE_CHECKING:
+ from ..action import Action # noqa: F401
+
+
+class OnActionEventBase(BaseEventHandler):
+ """Base event handler for events that have a source action."""
+
+ def __init__(
+ self,
+ *,
+ action_matcher: Optional[Union[Callable[['Action'], bool], 'Action']],
+ on_event: Union[
+ SomeActionsType,
+ Callable[[Event, LaunchContext], Optional[SomeActionsType]]
+ ],
+ target_event_cls: Type[Event],
+ target_action_cls: Type['Action'],
+ **kwargs
+ ) -> None:
+ """
+ Construct a `OnActionEventBase` instance.
+
+ :param action_matcher: `ExecuteProcess` instance or callable to filter events
+ from which proces/processes to handle.
+ :param on_event: Action to be done to handle the event.
+ :param target_event_cls: A subclass of `Event`, indicating which events
+ should be handled.
+ :param target_action_cls: A subclass of `Action`, indicating which kind of action can
+ generate the event.
+ """
+ if not issubclass(target_event_cls, Event):
+ raise TypeError("'target_event_cls' must be a subclass of 'Event'")
+ if (
+ not isinstance(action_matcher, (target_action_cls, type(None)))
+ and not callable(action_matcher)
+ ):
+ raise TypeError(
+ f"action_matcher must be an '{target_action_cls.__name__}' instance or a callable"
+ )
+ self.__target_event_cls = target_event_cls
+ self.__action_matcher = action_matcher
+
+ def event_matcher(event):
+ if not isinstance(event, target_event_cls):
+ return False
+ if callable(action_matcher):
+ return action_matcher(event.action)
+ if isinstance(action_matcher, target_action_cls):
+ return event.action is action_matcher
+ assert action_matcher is None
+ return True
+ super().__init__(matcher=event_matcher, **kwargs)
+ self.__actions_on_event: List[LaunchDescriptionEntity] = []
+ # TODO(wjwwood) check that it is not only callable, but also a callable that matches
+ # the correct signature for a handler in this case
+ if callable(on_event):
+ # Then on_exit is a function or lambda, so we can just call it, but
+ # we don't put anything in self.__actions_on_event because we cannot
+ # know what the function will return.
+ self.__on_event = on_event
+ else:
+ # Otherwise, setup self.__actions_on_event
+ if isinstance(on_event, collections.abc.Iterable):
+ for entity in on_event:
+ if not isinstance(entity, LaunchDescriptionEntity):
+ raise TypeError(
+ "expected all items in 'on_event' iterable to be of type "
+ "'LaunchDescriptionEntity' but got '{}'".format(type(entity)))
+ self.__actions_on_event = list(on_event) # Outside list is to ensure type is List
+ else:
+ self.__actions_on_event = [on_event]
+
+ def handle(self, event: Event, context: LaunchContext) -> Optional[SomeActionsType]:
+ """Handle the given event."""
+ super().handle(event, context)
+
+ if self.__actions_on_event:
+ return self.__actions_on_event
+ return self.__on_event(event, context)
+
+ @property
+ def handler_description(self) -> Text:
+ """Return the string description of the handler."""
+ # TODO(jacobperron): revisit how to describe known actions that are passed in.
+ # It would be nice if the parent class could output their description
+ # via the 'entities' property.
+ if self.__actions_on_event:
+ return '<actions>'
+ return '{}'.format(self.__on_event)
+
+ @property
+ def matcher_description(self) -> Text:
+ """Return the string description of the matcher."""
+ if self.__target_action is None:
+ return f'event == {self.__target_event_cls.__name__}'
+ return (
+ f'event == {self.__target_event_cls.__name__} and'
+ f' {self.__action_matcher.__name__}(event.action)'
+ )

launch/launch/event_handlers/on_execution_complete.py

@@ -12,27 +12,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import collections.abc
 from typing import Callable
 from typing import cast
-from typing import List # noqa
 from typing import Optional
-from typing import Text
 from typing import TYPE_CHECKING
 from typing import Union
 
+from .on_action_event_base import OnActionEventBase
 from ..event import Event
-from ..event_handler import EventHandler
 from ..events import ExecutionComplete
 from ..launch_context import LaunchContext
-from ..launch_description_entity import LaunchDescriptionEntity
 from ..some_actions_type import SomeActionsType
 
 if TYPE_CHECKING:
- from .. import Action # noqa
+ from .. import Action # noqa: F401
 
 
-class OnExecutionComplete(EventHandler):
+class OnExecutionComplete(OnActionEventBase):
  """
  Convenience class for handling an action completion event.
 
@@ -43,69 +39,25 @@ class OnExecutionComplete(EventHandler):
  def __init__(
  self,
  *,
- target_action: Optional['Action'] = None,
- on_completion: Union[SomeActionsType, Callable[[int], Optional[SomeActionsType]]],
+ target_action:
+ Optional[Union[Callable[['Action'], bool], 'Action']] = None,
+ on_completion:
+ Union[
+ SomeActionsType,
+ Callable[[ExecutionComplete, LaunchContext], Optional[SomeActionsType]]],
  **kwargs
  ) -> None:
  """Create an OnExecutionComplete event handler."""
- from ..action import Action # noqa
- if not isinstance(target_action, (Action, type(None))):
- raise ValueError("OnExecutionComplete requires an 'Action' as the target")
+ from ..action import Action # noqa: F811
+ on_completion = cast(
+ Union[
+ SomeActionsType,
+ Callable[[Event, LaunchContext], Optional[SomeActionsType]]],
+ on_completion)
  super().__init__(
- matcher=(
- lambda event: (
- isinstance(event, ExecutionComplete) and (
- target_action is None or
- event.action == target_action
- )
- )
- ),
- entities=None,
+ action_matcher=target_action,
+ on_event=on_completion,
+ target_event_cls=ExecutionComplete,
+ target_action_cls=Action,
  **kwargs,
  )
- self.__target_action = target_action
- # TODO(wjwwood) check that it is not only callable, but also a callable that matches
- # the correct signature for a handler in this case
- self.__on_completion = on_completion
- self.__actions_on_completion = [] # type: List[LaunchDescriptionEntity]
- if callable(on_completion):
- # Then on_completion is a function or lambda, so we can just call it, but
- # we don't put anything in self.__actions_on_completion because we cannot
- # know what the function will return.
- pass
- else:
- # Otherwise, setup self.__actions_on_completion
- if isinstance(on_completion, collections.abc.Iterable):
- for entity in on_completion:
- if not isinstance(entity, LaunchDescriptionEntity):
- raise ValueError(
- "expected all items in 'on_completion' iterable to be of type "
- "'LaunchDescriptionEntity' but got '{}'".format(type(entity)))
- self.__actions_on_completion = list(on_completion)
- else:
- self.__actions_on_completion = [on_completion]
- # Then return it from a lambda and use that as the self.__on_completion callback.
- self.__on_completion = lambda event, context: self.__actions_on_completion
-
- def handle(self, event: Event, context: LaunchContext) -> Optional[SomeActionsType]:
- """Handle the given event."""
- return self.__on_completion(cast(ExecutionComplete, event), context)
-
- @property
- def handler_description(self) -> Text:
- """Return the string description of the handler."""
- # TODO(jacobperron): revisit how to describe known actions that are passed in.
- # It would be nice if the parent class could output their description
- # via the 'entities' property.
- if self.__actions_on_completion:
- return '<actions>'
- return '{}'.format(self.__on_completion)
-
- @property
- def matcher_description(self) -> Text:
- """Return the string description of the matcher."""
- if self.__target_action is None:
- return 'event == ExecutionComplete'
- return 'event == ExecutionComplete and event.action == Action({})'.format(
- hex(id(self.__target_action))
- )

launch/launch/event_handlers/on_process_exit.py

@@ -1,4 +1,4 @@
-# Copyright 2018 Open Source Robotics Foundation, Inc.
+# Copyright 2018-2021 Open Source Robotics Foundation, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,27 +14,25 @@
 
 """Module for OnProcessExit class."""
 
-import collections.abc
 from typing import Callable
 from typing import cast
-from typing import List # noqa
 from typing import Optional
-from typing import Text
 from typing import TYPE_CHECKING
 from typing import Union
 
+from .on_action_event_base import OnActionEventBase
 from ..event import Event
-from ..event_handler import BaseEventHandler
 from ..events.process import ProcessExited
 from ..launch_context import LaunchContext
-from ..launch_description_entity import LaunchDescriptionEntity
 from ..some_actions_type import SomeActionsType
 
+
 if TYPE_CHECKING:
+ from ..actions import Action # noqa: F401
  from ..actions import ExecuteProcess # noqa: F401
 
 
-class OnProcessExit(BaseEventHandler):
+class OnProcessExit(OnActionEventBase):
  """
  Convenience class for handling a process exited event.
 
@@ -45,70 +43,29 @@ class OnProcessExit(BaseEventHandler):
  def __init__(
  self,
  *,
- target_action: 'ExecuteProcess' = None,
- on_exit: Union[SomeActionsType,
- Callable[[ProcessExited, LaunchContext], Optional[SomeActionsType]]],
+ target_action:
+ Optional[Union[Callable[['ExecuteProcess'], bool], 'ExecuteProcess']] = None,
+ on_exit:
+ Union[
+ SomeActionsType,
+ Callable[[ProcessExited, LaunchContext], Optional[SomeActionsType]]
+ ],
  **kwargs
  ) -> None:
  """Create an OnProcessExit event handler."""
- from ..actions import ExecuteProcess # noqa
- if not isinstance(target_action, (ExecuteProcess, type(None))):
- raise TypeError("OnProcessExit requires an 'ExecuteProcess' action as the target")
+ from ..actions import ExecuteProcess # noqa: F811
+ target_action = cast(
+ Optional[Union[Callable[['Action'], bool], 'Action']],
+ target_action)
+ on_exit = cast(
+ Union[
+ SomeActionsType,
+ Callable[[Event, LaunchContext], Optional[SomeActionsType]]],
+ on_exit)
  super().__init__(
- matcher=(
- lambda event: (
- isinstance(event, ProcessExited) and (
- target_action is None or
- event.action == target_action
- )
- )
- ),
+ action_matcher=target_action,
+ on_event=on_exit,
+ target_event_cls=ProcessExited,
+ target_action_cls=ExecuteProcess,
  **kwargs,
  )
- self.__target_action = target_action
- self.__actions_on_exit = [] # type: List[LaunchDescriptionEntity]
- # TODO(wjwwood) check that it is not only callable, but also a callable that matches
- # the correct signature for a handler in this case
- if callable(on_exit):
- # Then on_exit is a function or lambda, so we can just call it, but
- # we don't put anything in self.__actions_on_exit because we cannot
- # know what the function will return.
- self.__on_exit = on_exit
- else:
- # Otherwise, setup self.__actions_on_exit
- if isinstance(on_exit, collections.abc.Iterable):
- for entity in on_exit:
- if not isinstance(entity, LaunchDescriptionEntity):
- raise ValueError(
- "expected all items in 'on_exit' iterable to be of type "
- "'LaunchDescriptionEntity' but got '{}'".format(type(entity)))
- self.__actions_on_exit = list(on_exit) # Outside list is to ensure type is List
- else:
- self.__actions_on_exit = [on_exit]
-
- def handle(self, event: Event, context: LaunchContext) -> Optional[SomeActionsType]:
- """Handle the given event."""
- super().handle(event, context)
-
- if self.__actions_on_exit:
- return self.__actions_on_exit
- return self.__on_exit(cast(ProcessExited, event), context)
-
- @property
- def handler_description(self) -> Text:
- """Return the string description of the handler."""
- # TODO(jacobperron): revisit how to describe known actions that are passed in.
- # It would be nice if the parent class could output their description
- # via the 'entities' property.
- if self.__actions_on_exit:
- return '<actions>'
- return '{}'.format(self.__on_exit)
-
- @property
- def matcher_description(self) -> Text:
- """Return the string description of the matcher."""
- if self.__target_action is None:
- return 'event == ProcessExited'
- return 'event == ProcessExited and event.action == ExecuteProcess({})'.format(
- hex(id(self.__target_action))
- )