Skip to content

Commit

Permalink
Revert "Adds types to Lifecycle Objects (#1338)"
Browse files Browse the repository at this point in the history
This reverts commit 2f9a771.
  • Loading branch information
fujitatomoya authored Oct 14, 2024
1 parent a09a031 commit 0c66993
Show file tree
Hide file tree
Showing 4 changed files with 56 additions and 258 deletions.
97 changes: 14 additions & 83 deletions rclpy/rclpy/impl/_rclpy_pybind11.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ from __future__ import annotations

from enum import Enum, IntEnum
from types import TracebackType
from typing import Any, Generic, Literal, overload, Sequence, TypeAlias, TypedDict
from typing import Any, Generic, Literal, overload, Sequence, TypedDict

from rclpy.clock import JumpHandle
from rclpy.clock_type import ClockType
Expand Down Expand Up @@ -101,23 +101,21 @@ class rcl_duration_t:
nanoseconds: int


class rcl_subscription_event_type_t(Enum):
_value_: int
RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED = ...
RCL_SUBSCRIPTION_LIVELINESS_CHANGED = ...
RCL_SUBSCRIPTION_REQUESTED_INCOMPATIBLE_QOS = ...
RCL_SUBSCRIPTION_MESSAGE_LOST = ...
RCL_SUBSCRIPTION_INCOMPATIBLE_TYPE = ...
RCL_SUBSCRIPTION_MATCHED = ...
class rcl_subscription_event_type_t(IntEnum):
RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED: int
RCL_SUBSCRIPTION_LIVELINESS_CHANGED: int
RCL_SUBSCRIPTION_REQUESTED_INCOMPATIBLE_QOS: int
RCL_SUBSCRIPTION_MESSAGE_LOST: int
RCL_SUBSCRIPTION_INCOMPATIBLE_TYPE: int
RCL_SUBSCRIPTION_MATCHED: int


class rcl_publisher_event_type_t(Enum):
_value_: int
RCL_PUBLISHER_OFFERED_DEADLINE_MISSED = ...
RCL_PUBLISHER_LIVELINESS_LOST = ...
RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS = ...
RCL_PUBLISHER_INCOMPATIBLE_TYPE = ...
RCL_PUBLISHER_MATCHED = ...
class rcl_publisher_event_type_t(IntEnum):
RCL_PUBLISHER_OFFERED_DEADLINE_MISSED: int
RCL_PUBLISHER_LIVELINESS_LOST: int
RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS: int
RCL_PUBLISHER_INCOMPATIBLE_TYPE: int
RCL_PUBLISHER_MATCHED: int


class EventHandle(Destroyable):
Expand All @@ -137,73 +135,6 @@ class EventHandle(Destroyable):
"""Get pending data from a ready event."""


LifecycleStateMachineState: TypeAlias = tuple[int, str]


class LifecycleStateMachine(Destroyable):

def __init__(self, node: Node, enable_com_interface: bool) -> None: ...

@property
def initialized(self) -> bool:
"""Check if state machine is initialized."""

@property
def current_state(self) -> LifecycleStateMachineState:
"""Get the current state machine state."""

@property
def available_states(self) -> list[LifecycleStateMachineState]:
"""Get the available states."""

@property
def available_transitions(self) -> list[tuple[int, str, int, str, int, str]]:
"""Get the available transitions."""

@property
def transition_graph(self) -> list[tuple[int, str, int, str, int, str]]:
"""Get the transition graph."""

def get_transition_by_label(self, label: str) -> int:
"""Get the transition id from a transition label."""

def trigger_transition_by_id(self, transition_id: int, publish_update: bool) -> None:
"""Trigger a transition by transition id."""

def trigger_transition_by_label(self, label: str, publish_update: bool) -> None:
"""Trigger a transition by label."""

@property
def service_change_state(self) -> Service:
"""Get the change state service."""

@property
def service_get_state(self) -> Service:
"""Get the get state service."""

@property
def service_get_available_states(self) -> Service:
"""Get the get available states service."""

@property
def service_get_available_transitions(self) -> Service:
"""Get the get available transitions service."""

@property
def service_get_transition_graph(self) -> Service:
"""Get the get transition graph service."""


class TransitionCallbackReturnType(Enum):
_value_: int
SUCCESS = ...
FAILURE = ...
ERROR = ...

def to_label(self) -> str:
"""Convert the transition callback return code to a transition label."""


class GuardCondition(Destroyable):

def __init__(self, context: Context) -> None: ...
Expand Down
50 changes: 14 additions & 36 deletions rclpy/rclpy/lifecycle/managed_entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,85 +13,63 @@
# limitations under the License.

from functools import wraps
from typing import Any, Callable, Dict, List, Optional, overload, TYPE_CHECKING, Union

from ..impl.implementation_singleton import rclpy_implementation as _rclpy

if TYPE_CHECKING:
from typing import TypeAlias
from rclpy.lifecycle.node import LifecycleState


TransitionCallbackReturn: 'TypeAlias' = _rclpy.TransitionCallbackReturnType
TransitionCallbackReturn = _rclpy.TransitionCallbackReturnType


class ManagedEntity:

def on_configure(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_configure(self, state) -> TransitionCallbackReturn:
"""Handle configure transition request."""
return TransitionCallbackReturn.SUCCESS

def on_cleanup(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_cleanup(self, state) -> TransitionCallbackReturn:
"""Handle cleanup transition request."""
return TransitionCallbackReturn.SUCCESS

def on_shutdown(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_shutdown(self, state) -> TransitionCallbackReturn:
"""Handle shutdown transition request."""
return TransitionCallbackReturn.SUCCESS

def on_activate(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_activate(self, state) -> TransitionCallbackReturn:
"""Handle activate transition request."""
return TransitionCallbackReturn.SUCCESS

def on_deactivate(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_deactivate(self, state) -> TransitionCallbackReturn:
"""Handle deactivate transition request."""
return TransitionCallbackReturn.SUCCESS

def on_error(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_error(self, state) -> TransitionCallbackReturn:
"""Handle error transition request."""
return TransitionCallbackReturn.SUCCESS


class SimpleManagedEntity(ManagedEntity):
"""A simple managed entity that only sets a flag when activated/deactivated."""

def __init__(self) -> None:
def __init__(self):
self._enabled = False

def on_activate(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_activate(self, state) -> TransitionCallbackReturn:
self._enabled = True
return TransitionCallbackReturn.SUCCESS

def on_deactivate(self, state: 'LifecycleState') -> TransitionCallbackReturn:
def on_deactivate(self, state) -> TransitionCallbackReturn:
self._enabled = False
return TransitionCallbackReturn.SUCCESS

@property
def is_activated(self) -> bool:
def is_activated(self):
return self._enabled

@staticmethod
@overload
def when_enabled(wrapped: None, *,
when_not_enabled: Optional[Callable[..., None]] = None
) -> Callable[[Callable[..., None]], Callable[..., None]]: ...

@staticmethod
@overload
def when_enabled(wrapped: Callable[..., None], *,
when_not_enabled: Optional[Callable[..., None]] = None
) -> Callable[..., None]: ...

@staticmethod
def when_enabled(wrapped: Optional[Callable[..., None]] = None, *,
when_not_enabled: Optional[Callable[..., None]] = None) -> Union[
Callable[..., None],
Callable[[Callable[..., None]], Callable[..., None]]
]:
def decorator(wrapped: Callable[..., None]) -> Callable[..., None]:
def when_enabled(wrapped=None, *, when_not_enabled=None):
def decorator(wrapped):
@wraps(wrapped)
def only_when_enabled_wrapper(self: SimpleManagedEntity, *args: List[Any],
**kwargs: Dict[str, Any]) -> None:
def only_when_enabled_wrapper(self: SimpleManagedEntity, *args, **kwargs):
if not self._enabled:
if when_not_enabled is not None:
when_not_enabled()
Expand Down
Loading

0 comments on commit 0c66993

Please sign in to comment.