nova.actions

12class Action(pydantic.BaseModel, ABC):
13    _registry: ClassVar[dict[str, type[Action]]] = {}
14
15    def __init_subclass__(cls, **kwargs):
16        super().__init_subclass__(**kwargs)
17
18        action_type = getattr(cls, "type", None)
19        # when no type is found -> skip
20        if not isinstance(action_type, str):
21            logger.warning(f"Action class '{cls.__name__}' does not have a valid type")
22            return
23
24        if action_type in Action._registry:
25            logger.warning(f"Duplicate action type '{action_type}'")
26            return
27        Action._registry[action_type] = cls
28        logger.debug(f"Registered action type: {action_type}")
29
30    @classmethod
31    def from_dict(cls, data: dict) -> Action:
32        """
33        Pick the correct concrete Action from the `_registry`
34        and let Pydantic validate against that class.
35        """
36        if not isinstance(data, dict):
37            raise TypeError("`data` must be a dict")
38
39        action_type = data.get("type")
40        if action_type is None:
41            raise ValueError("Missing required key `type`")
42
43        try:
44            concrete_cls = Action._registry[action_type]
45        except KeyError:
46            raise ValueError(f"Unknown action type '{action_type}'")
47
48        return concrete_cls.model_validate(data)
49
50    @abstractmethod
51    def is_motion(self) -> bool:
52        """Return whether the action is a motion"""
53
54    @abstractmethod
55    def to_api_model(self):
56        """Convert the action to an API model"""

148def cartesian_ptp(
149    target: PoseOrVectorTuple,
150    settings: MotionSettings = MotionSettings(),
151    collision_scene: wb.models.CollisionScene | None = None,
152) -> CartesianPTP:
153    """Convenience function to create a point-to-point motion
154
155    Args:
156        target: the target pose or vector. If the target is a vector, the orientation is set to (0, 0, 0).
157        settings: the motion settings
158
159    Returns: the point-to-point motion
160
161    Examples:
162    >>> ms = MotionSettings(tcp_acceleration_limit=10)
163    >>> assert cartesian_ptp((1, 2, 3, 4, 5, 6), settings=ms) == CartesianPTP(target=Pose((1, 2, 3, 4, 5, 6)), settings=ms)
164    >>> assert cartesian_ptp((1, 2, 3)) == cartesian_ptp((1, 2, 3, 0, 0, 0))
165    >>> assert cartesian_ptp(Pose((1, 2, 3, 4, 5, 6)), settings=ms) == cartesian_ptp((1, 2, 3, 4, 5, 6), settings=ms)
166    >>> Action.from_dict(cartesian_ptp((1, 2, 3, 4, 5, 6), MotionSettings()).model_dump())
167    CartesianPTP(type='cartesian_ptp', target=Pose(position=Vector3d(x=1, y=2, z=3), orientation=Vector3d(x=4, y=5, z=6)), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None)
168
169    """
170    if not isinstance(target, Pose):
171        t = (*target, 0.0, 0.0, 0.0) if len(target) == 3 else target
172        target = Pose(t)
173
174    return CartesianPTP(target=target, settings=settings, collision_scene=collision_scene)

148def cartesian_ptp(
149    target: PoseOrVectorTuple,
150    settings: MotionSettings = MotionSettings(),
151    collision_scene: wb.models.CollisionScene | None = None,
152) -> CartesianPTP:
153    """Convenience function to create a point-to-point motion
154
155    Args:
156        target: the target pose or vector. If the target is a vector, the orientation is set to (0, 0, 0).
157        settings: the motion settings
158
159    Returns: the point-to-point motion
160
161    Examples:
162    >>> ms = MotionSettings(tcp_acceleration_limit=10)
163    >>> assert cartesian_ptp((1, 2, 3, 4, 5, 6), settings=ms) == CartesianPTP(target=Pose((1, 2, 3, 4, 5, 6)), settings=ms)
164    >>> assert cartesian_ptp((1, 2, 3)) == cartesian_ptp((1, 2, 3, 0, 0, 0))
165    >>> assert cartesian_ptp(Pose((1, 2, 3, 4, 5, 6)), settings=ms) == cartesian_ptp((1, 2, 3, 4, 5, 6), settings=ms)
166    >>> Action.from_dict(cartesian_ptp((1, 2, 3, 4, 5, 6), MotionSettings()).model_dump())
167    CartesianPTP(type='cartesian_ptp', target=Pose(position=Vector3d(x=1, y=2, z=3), orientation=Vector3d(x=4, y=5, z=6)), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None)
168
169    """
170    if not isinstance(target, Pose):
171        t = (*target, 0.0, 0.0, 0.0) if len(target) == 3 else target
172        target = Pose(t)
173
174    return CartesianPTP(target=target, settings=settings, collision_scene=collision_scene)

209def circular(
210    target: PoseOrVectorTuple,
211    intermediate: PoseOrVectorTuple,
212    settings: MotionSettings = MotionSettings(),
213    collision_scene: wb.models.CollisionScene | None = None,
214) -> Circular:
215    """Convenience function to create a circular motion
216
217    Args:
218        target: the target pose or vector. If the target is a vector, the orientation is set to (0, 0, 0).
219        intermediate: the intermediate pose or vector. If the intermediate is a vector, the orientation is set to
220            (0, 0, 0).
221        settings: the motion settings
222
223    Returns: the circular motion
224
225    Examples:
226    >>> ms = MotionSettings(tcp_acceleration_limit=10)
227    >>> assert circular((1, 2, 3, 4, 5, 6), (7, 8, 9, 10, 11, 12), settings=ms) == Circular(target=Pose((1, 2, 3, 4, 5, 6)), intermediate=Pose((7, 8, 9, 10, 11, 12)), settings=ms)
228    >>> assert circular((1, 2, 3), (4, 5, 6)) == circular((1, 2, 3, 0, 0, 0), (4, 5, 6, 0, 0, 0))
229    >>> Action.from_dict(circular((1, 2, 3, 4, 5, 6), (7, 8, 9, 10, 11, 12), MotionSettings()).model_dump())
230    Circular(type='circular', target=Pose(position=Vector3d(x=1, y=2, z=3), orientation=Vector3d(x=4, y=5, z=6)), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None, intermediate=Pose(position=Vector3d(x=7, y=8, z=9), orientation=Vector3d(x=10, y=11, z=12)))
231
232    """
233    if not isinstance(target, Pose):
234        t = (*target, 0.0, 0.0, 0.0) if len(target) == 3 else target
235        target = Pose(t)
236
237    if not isinstance(intermediate, Pose):
238        i = (*intermediate, 0.0, 0.0, 0.0) if len(intermediate) == 3 else intermediate
239        intermediate = Pose(i)
240
241    return Circular(
242        target=target, intermediate=intermediate, settings=settings, collision_scene=collision_scene
243    )

209def circular(
210    target: PoseOrVectorTuple,
211    intermediate: PoseOrVectorTuple,
212    settings: MotionSettings = MotionSettings(),
213    collision_scene: wb.models.CollisionScene | None = None,
214) -> Circular:
215    """Convenience function to create a circular motion
216
217    Args:
218        target: the target pose or vector. If the target is a vector, the orientation is set to (0, 0, 0).
219        intermediate: the intermediate pose or vector. If the intermediate is a vector, the orientation is set to
220            (0, 0, 0).
221        settings: the motion settings
222
223    Returns: the circular motion
224
225    Examples:
226    >>> ms = MotionSettings(tcp_acceleration_limit=10)
227    >>> assert circular((1, 2, 3, 4, 5, 6), (7, 8, 9, 10, 11, 12), settings=ms) == Circular(target=Pose((1, 2, 3, 4, 5, 6)), intermediate=Pose((7, 8, 9, 10, 11, 12)), settings=ms)
228    >>> assert circular((1, 2, 3), (4, 5, 6)) == circular((1, 2, 3, 0, 0, 0), (4, 5, 6, 0, 0, 0))
229    >>> Action.from_dict(circular((1, 2, 3, 4, 5, 6), (7, 8, 9, 10, 11, 12), MotionSettings()).model_dump())
230    Circular(type='circular', target=Pose(position=Vector3d(x=1, y=2, z=3), orientation=Vector3d(x=4, y=5, z=6)), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None, intermediate=Pose(position=Vector3d(x=7, y=8, z=9), orientation=Vector3d(x=10, y=11, z=12)))
231
232    """
233    if not isinstance(target, Pose):
234        t = (*target, 0.0, 0.0, 0.0) if len(target) == 3 else target
235        target = Pose(t)
236
237    if not isinstance(intermediate, Pose):
238        i = (*intermediate, 0.0, 0.0, 0.0) if len(intermediate) == 3 else intermediate
239        intermediate = Pose(i)
240
241    return Circular(
242        target=target, intermediate=intermediate, settings=settings, collision_scene=collision_scene
243    )

35def collision_free(
36    target: Pose | tuple[float, ...],
37    settings: MotionSettings = MotionSettings(),
38    collision_scene: wb.models.CollisionScene | None = None,
39) -> CollisionFreeMotion:
40    return CollisionFreeMotion(target=target, settings=settings, collision_scene=collision_scene)

 26class CombinedActions(pydantic.BaseModel):
 27    """A trajectory of motions and actions"""
 28
 29    # See: https://docs.pydantic.dev/latest/concepts/serialization/#serialize_as_any-runtime-setting
 30    items: tuple[
 31        Annotated[
 32            pydantic.SerializeAsAny[ActionContainerItem], pydantic.Field(discriminator="type")
 33        ],
 34        ...,
 35    ] = ()
 36
 37    def __len__(self):
 38        return len(self.items)
 39
 40    def __getitem__(self, item):
 41        return self.items[item]
 42
 43    def __setattr__(self, key, value):
 44        if key == "items":
 45            raise TypeError("Cannot set items directly")
 46        super().__setattr__(key, value)
 47
 48    def __iter__(self):
 49        return iter(self.items)
 50
 51    def append(self, item: ActionContainerItem):
 52        super().__setattr__("items", self.items + (item,))
 53
 54    def _generate_trajectory(
 55        self,
 56    ) -> tuple[list[Motion | CollisionFreeMotion], list[ActionLocation]]:
 57        """Generate two lists: one of Motion objects and another of ActionContainer objects,
 58        where each ActionContainer wraps a non-Motion action with its path parameter.
 59
 60        The path parameter is the index of the last Motion object in the list of Motion objects.
 61        S - M - M - A - A - M - M - A - M - M
 62        0 - 1 - 2 - 3 - 3 - 3 - 4 - 5 - 5 - 6
 63
 64        Returns:
 65            tuple: A tuple containing:
 66                - list of Motion objects from self.items.
 67                - list of ActionContainer objects with indexed path parameters.
 68        """
 69        motions = []
 70        actions = []
 71        last_motion_index = 0
 72
 73        for item in self.items:
 74            if isinstance(item, WaitAction):
 75                continue  # Skip WaitAction items
 76            if isinstance(item, Motion) or isinstance(item, CollisionFreeMotion):
 77                motions.append(item)
 78                last_motion_index += 1  # Increment the motion index for each new Motion
 79            else:
 80                # Assign the current value of last_motion_index as path_parameter for actions
 81                actions.append(ActionLocation(path_parameter=last_motion_index, action=item))
 82
 83        return motions, actions
 84
 85    @property
 86    def motions(self) -> list[Motion | CollisionFreeMotion]:
 87        motions, _ = self._generate_trajectory()
 88        return motions
 89
 90    @property
 91    def actions(self) -> list[ActionLocation]:
 92        _, actions = self._generate_trajectory()
 93        return actions
 94
 95    @property
 96    def start(self) -> ActionContainerItem | None:
 97        return self.motions[0] if self.motions else None
 98
 99    @property
100    def end(self) -> ActionContainerItem | None:
101        return self.motions[-1] if self.motions else None
102
103    def poses(self) -> list[Pose]:
104        """Returns the positions of all motions. If a motion is not a cartesian motion, the position is ignored
105
106        Returns: the positions
107
108        """
109        motions, _ = self._generate_trajectory()
110        return [
111            Pose(position=motion.target.position, orientation=motion.target.orientation)
112            for motion in motions
113            if isinstance(motion.target, Pose)
114        ]
115
116    def positions(self):
117        """Returns the positions of all motions. If a motion is not a cartesian motion, the position is ignored
118
119        Returns: the positions
120
121        """
122        return [pose.position for pose in self.poses()]
123
124    def orientations(self):
125        """Returns the orientations of all motions. If a motion is not a cartesian motion, the orientation is ignored
126
127        Returns: the orientations
128
129        """
130        return [pose.orientation for pose in self.poses()]
131
132    def __add__(self, other: CombinedActions) -> CombinedActions:
133        return CombinedActions(items=self.items + other.items)
134
135    def to_motion_command(self) -> list[api.models.MotionCommand]:
136        motion_commands = []
137        for motion in self.motions:
138            path = api.models.MotionCommandPath.from_dict(motion.to_api_model().model_dump())
139            blending = (
140                motion.settings.as_blending_setting()
141                if motion.settings.has_blending_settings()
142                else None
143            )
144            limits_override = (
145                motion.settings.as_limits_settings()
146                if motion.settings.has_limits_override()
147                else None
148            )
149            motion_commands.append(
150                api.models.MotionCommand(
151                    path=path, blending=blending, limits_override=limits_override
152                )
153            )
154        return motion_commands
155
156    def to_set_io(self) -> list[api.models.SetIO]:
157        return [
158            api.models.SetIO(
159                io=api.models.IOValue(**action.action.to_api_model()),
160                location=action.path_parameter,
161            )
162            for action in self.actions
163        ]

21def io_write(key: str, value: Any, device_id: str | None = None) -> WriteAction:
22    """Create a WriteAction
23
24    Args:
25        key: The key to write
26        value: The value to write
27        device_id: The device id
28
29    Returns:
30        The WriteAction
31
32    """
33    return WriteAction(key=key, value=value, device_id=device_id)

274def joint_ptp(
275    target: tuple[float, ...],
276    settings: MotionSettings = MotionSettings(),
277    collision_scene: wb.models.CollisionScene | None = None,
278) -> JointPTP:
279    """Convenience function to create a joint PTP motion
280
281    Args:
282        target: the target joint configuration
283        settings: the motion settings
284
285    Returns: the joint PTP motion
286
287    Examples:
288    >>> ms = MotionSettings(tcp_acceleration_limit=10)
289    >>> assert joint_ptp((1, 2, 3, 4, 5, 6), settings=ms) == JointPTP(target=(1, 2, 3, 4, 5, 6), settings=ms)
290    >>> Action.from_dict(joint_ptp((1, 2, 3, 4, 5, 6), MotionSettings()).model_dump())
291    JointPTP(type='joint_ptp', target=(1.0, 2.0, 3.0, 4.0, 5.0, 6.0), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None)
292
293    """
294    return JointPTP(target=target, settings=settings, collision_scene=collision_scene)

274def joint_ptp(
275    target: tuple[float, ...],
276    settings: MotionSettings = MotionSettings(),
277    collision_scene: wb.models.CollisionScene | None = None,
278) -> JointPTP:
279    """Convenience function to create a joint PTP motion
280
281    Args:
282        target: the target joint configuration
283        settings: the motion settings
284
285    Returns: the joint PTP motion
286
287    Examples:
288    >>> ms = MotionSettings(tcp_acceleration_limit=10)
289    >>> assert joint_ptp((1, 2, 3, 4, 5, 6), settings=ms) == JointPTP(target=(1, 2, 3, 4, 5, 6), settings=ms)
290    >>> Action.from_dict(joint_ptp((1, 2, 3, 4, 5, 6), MotionSettings()).model_dump())
291    JointPTP(type='joint_ptp', target=(1.0, 2.0, 3.0, 4.0, 5.0, 6.0), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None)
292
293    """
294    return JointPTP(target=target, settings=settings, collision_scene=collision_scene)

 90def linear(
 91    target: PoseOrVectorTuple,
 92    settings: MotionSettings = MotionSettings(),
 93    collision_scene: wb.models.CollisionScene | None = None,
 94) -> Linear:
 95    """Convenience function to create a linear motion
 96
 97    Args:
 98        target: the target pose or vector. If the target is a vector, the orientation is set to (0, 0, 0).
 99        settings: the motion settings
100
101    Returns: the linear motion
102
103    Examples:
104    >>> ms = MotionSettings(tcp_velocity_limit=10)
105    >>> assert linear((1, 2, 3, 4, 5, 6), settings=ms) == Linear(target=Pose((1, 2, 3, 4, 5, 6)), settings=ms)
106    >>> assert linear((1, 2, 3)) == linear((1, 2, 3, 0, 0, 0))
107    >>> assert linear(Pose((1, 2, 3, 4, 5, 6)), settings=ms) == linear((1, 2, 3, 4, 5, 6), settings=ms)
108    >>> Action.from_dict(linear((1, 2, 3, 4, 5, 6), MotionSettings()).model_dump())
109    Linear(type='linear', target=Pose(position=Vector3d(x=1, y=2, z=3), orientation=Vector3d(x=4, y=5, z=6)), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None)
110
111    """
112    if not isinstance(target, Pose):
113        t = (*target, 0.0, 0.0, 0.0) if len(target) == 3 else target
114        target = Pose(t)
115
116    return Linear(target=target, settings=settings, collision_scene=collision_scene)

 90def linear(
 91    target: PoseOrVectorTuple,
 92    settings: MotionSettings = MotionSettings(),
 93    collision_scene: wb.models.CollisionScene | None = None,
 94) -> Linear:
 95    """Convenience function to create a linear motion
 96
 97    Args:
 98        target: the target pose or vector. If the target is a vector, the orientation is set to (0, 0, 0).
 99        settings: the motion settings
100
101    Returns: the linear motion
102
103    Examples:
104    >>> ms = MotionSettings(tcp_velocity_limit=10)
105    >>> assert linear((1, 2, 3, 4, 5, 6), settings=ms) == Linear(target=Pose((1, 2, 3, 4, 5, 6)), settings=ms)
106    >>> assert linear((1, 2, 3)) == linear((1, 2, 3, 0, 0, 0))
107    >>> assert linear(Pose((1, 2, 3, 4, 5, 6)), settings=ms) == linear((1, 2, 3, 4, 5, 6), settings=ms)
108    >>> Action.from_dict(linear((1, 2, 3, 4, 5, 6), MotionSettings()).model_dump())
109    Linear(type='linear', target=Pose(position=Vector3d(x=1, y=2, z=3), orientation=Vector3d(x=4, y=5, z=6)), settings=MotionSettings(min_blending_velocity=None, position_zone_radius=None, joint_velocity_limits=None, joint_acceleration_limits=None, tcp_velocity_limit=50.0, tcp_acceleration_limit=None, tcp_orientation_velocity_limit=None, tcp_orientation_acceleration_limit=None), collision_scene=None)
110
111    """
112    if not isinstance(target, Pose):
113        t = (*target, 0.0, 0.0, 0.0) if len(target) == 3 else target
114        target = Pose(t)
115
116    return Linear(target=target, settings=settings, collision_scene=collision_scene)

24def wait(wait_for_in_seconds: float) -> WaitAction:
25    return WaitAction(wait_for_in_seconds=wait_for_in_seconds)

167class MovementControllerContext(pydantic.BaseModel):
168    combined_actions: CombinedActions
169    motion_id: str