MCPcopy Create free account
hub / github.com/BeingBeyond/BeTTER / evaluate_node_condition

Function evaluate_node_condition

src/task/conditions/predicates.py:70–102  ·  view source on GitHub ↗
(ctx: ConditionContext, condition: dict[str, Any])

Source from the content-addressed store, hash-verified

68
69
70def evaluate_node_condition(ctx: ConditionContext, condition: dict[str, Any]) -> bool:
71 object_id = condition.get("subject_id") or condition.get("subject")
72 if object_id is None:
73 raise ValueError(f"Node condition missing subject_id: {condition}")
74
75 attribute = condition.get("attribute")
76 if attribute is None:
77 raise ValueError(f"Node condition missing attribute: {condition}")
78
79 axis_name = str(condition.get("axis", "z"))
80 axis_to_idx = {"x": 0, "y": 1, "z": 2}
81 if axis_name not in axis_to_idx:
82 raise ValueError(f"Unsupported axis '{axis_name}'. Expected one of x/y/z.")
83 axis_idx = axis_to_idx[axis_name]
84
85 obj = ctx.get_object(str(object_id))
86 pos, quat_wxyz = obj.get_world_pose()
87
88 min_value = condition.get("min")
89 max_value = condition.get("max")
90 if min_value is None or max_value is None:
91 raise ValueError(f"Node condition requires both min and max: {condition}")
92
93 if attribute == "position":
94 value = float(np.asarray(pos, dtype=float)[axis_idx])
95 elif attribute in {"rotation_deg", "euler_deg"}:
96 quat_xyzw = wxyz_to_xyzw(quat_wxyz)
97 euler_xyz_deg = R.from_quat(quat_xyzw).as_euler("xyz", degrees=True)
98 value = float(euler_xyz_deg[axis_idx])
99 else:
100 raise ValueError(f"Unsupported node attribute '{attribute}'")
101
102 return float(min_value) <= value <= float(max_value)
103
104
105def evaluate_edge_condition(ctx: ConditionContext, condition: dict[str, Any]) -> bool:

Callers 1

_eval_nodeMethod · 0.85

Calls 3

wxyz_to_xyzwFunction · 0.90
get_objectMethod · 0.80
get_world_poseMethod · 0.45

Tested by

no test coverage detected