Skip to content

Commit 857d506

Browse files
authored
Update schemas.py
1 parent 1053982 commit 857d506

1 file changed

Lines changed: 82 additions & 16 deletions

File tree

src/ohip/schemas.py

Lines changed: 82 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@ class ConsentSource(str, Enum):
5252
GESTURE = "gesture"
5353
UI = "ui"
5454
PROFILE = "profile"
55+
BENCHMARK = "benchmark"
56+
REPLAY = "replay"
5557

5658

5759
class NudgeLevel(str, Enum):
@@ -235,18 +237,85 @@ class ImpedanceProfile:
235237
tangential_N_per_mm: Tuple[float, float] # (min, max)
236238

237239

238-
@dataclass
240+
@dataclass(init=False)
239241
class ContactPlan:
240242
"""Concrete contact execution plan (bounded by safety envelopes)."""
243+
241244
target: Pose
242245
contact_normal: Vector3
243246
peak_force_N: float
244247
dwell_ms: int
245248
approach_speed_mps: float
246249
release_speed_mps: float
247250
impedance: ImpedanceProfile
248-
rationale: str = ""
249-
consent_mode: ConsentMode = ConsentMode.NONE
251+
rationale: str
252+
consent_mode: ConsentMode
253+
contact_zone: str
254+
255+
def __init__(
256+
self,
257+
*,
258+
target: Pose,
259+
contact_normal: Optional[Vector3] = None,
260+
normal: Optional[Vector3] = None,
261+
peak_force_N: Optional[float] = None,
262+
max_force_N: Optional[float] = None,
263+
dwell_ms: int | Tuple[int, int] | List[int] = 1500,
264+
approach_speed_mps: float = 0.15,
265+
release_speed_mps: float = 0.20,
266+
impedance: ImpedanceProfile | Dict[str, Any] | None = None,
267+
rationale: str = "",
268+
consent_mode: ConsentMode | str = ConsentMode.NONE,
269+
contact_zone: str = "",
270+
) -> None:
271+
self.target = target
272+
self.contact_normal = self._coerce_vector(contact_normal if contact_normal is not None else normal)
273+
chosen_force = peak_force_N if peak_force_N is not None else max_force_N
274+
self.peak_force_N = float(1.0 if chosen_force is None else chosen_force)
275+
self.dwell_ms = self._coerce_dwell_ms(dwell_ms)
276+
self.approach_speed_mps = float(approach_speed_mps)
277+
self.release_speed_mps = float(release_speed_mps)
278+
self.impedance = self._coerce_impedance(impedance)
279+
self.rationale = str(rationale)
280+
self.consent_mode = consent_mode if isinstance(consent_mode, ConsentMode) else ConsentMode(str(consent_mode))
281+
self.contact_zone = str(contact_zone)
282+
self.validate()
283+
284+
@staticmethod
285+
def _coerce_vector(value: Optional[Vector3]) -> Vector3:
286+
if value is None:
287+
return Vector3(0.0, 0.0, 1.0)
288+
if isinstance(value, Vector3):
289+
return value
290+
raise TypeError("contact normal must be a Vector3")
291+
292+
@staticmethod
293+
def _coerce_dwell_ms(value: int | Tuple[int, int] | List[int]) -> int:
294+
if isinstance(value, (tuple, list)):
295+
assert len(value) == 2, "dwell_ms range must contain exactly 2 elements"
296+
lo = int(value[0])
297+
hi = int(value[1])
298+
assert lo <= hi, "dwell_ms range must satisfy min <= max"
299+
return int(round((lo + hi) / 2.0))
300+
return int(value)
301+
302+
@staticmethod
303+
def _coerce_impedance(value: ImpedanceProfile | Dict[str, Any] | None) -> ImpedanceProfile:
304+
if isinstance(value, ImpedanceProfile):
305+
return value
306+
data = value or {}
307+
return ImpedanceProfile(
308+
normal_N_per_mm=tuple(map(float, data.get("normal_N_per_mm", [0.3, 0.6]))),
309+
tangential_N_per_mm=tuple(map(float, data.get("tangential_N_per_mm", [0.1, 0.3]))),
310+
)
311+
312+
@property
313+
def normal(self) -> Vector3:
314+
return self.contact_normal
315+
316+
@property
317+
def max_force_N(self) -> float:
318+
return self.peak_force_N
250319

251320
def validate(self) -> None:
252321
assert self.peak_force_N >= 0.0, "peak_force_N must be >= 0"
@@ -259,7 +328,7 @@ def validate(self) -> None:
259328
assert 0.0 <= lo <= hi, f"impedance {name} invalid (min <= max required)"
260329

261330
def to_dict(self) -> Dict[str, Any]:
262-
return {
331+
data = {
263332
"target": self.target.to_dict(),
264333
"normal": self.contact_normal.as_list(),
265334
"peak_force_N": float(self.peak_force_N),
@@ -273,27 +342,24 @@ def to_dict(self) -> Dict[str, Any]:
273342
"rationale": self.rationale,
274343
"consent_mode": self.consent_mode.value,
275344
}
345+
if self.contact_zone:
346+
data["contact_zone"] = self.contact_zone
347+
return data
276348

277349
@staticmethod
278350
def from_dict(d: Dict[str, Any]) -> "ContactPlan":
279-
imp = d.get("impedance", {})
280-
profile = ImpedanceProfile(
281-
normal_N_per_mm=tuple(map(float, imp.get("normal_N_per_mm", [0.3, 0.6]))), # spec defaults
282-
tangential_N_per_mm=tuple(map(float, imp.get("tangential_N_per_mm", [0.1, 0.3]))),
283-
)
284-
plan = ContactPlan(
351+
return ContactPlan(
285352
target=Pose.from_dict(d["target"]),
286-
contact_normal=Vector3.from_list(d.get("normal", [0.0, 0.0, 1.0])),
287-
peak_force_N=float(d.get("peak_force_N", 1.0)),
288-
dwell_ms=int(d.get("dwell_ms", 1500)),
353+
contact_normal=Vector3.from_list(d.get("contact_normal", d.get("normal", [0.0, 0.0, 1.0]))),
354+
peak_force_N=float(d.get("peak_force_N", d.get("max_force_N", 1.0))),
355+
dwell_ms=d.get("dwell_ms", 1500),
289356
approach_speed_mps=float(d.get("approach_speed_mps", 0.15)),
290357
release_speed_mps=float(d.get("release_speed_mps", 0.2)),
291-
impedance=profile,
358+
impedance=d.get("impedance", {}),
292359
rationale=str(d.get("rationale", "")),
293360
consent_mode=ConsentMode(d.get("consent_mode", "none")),
361+
contact_zone=str(d.get("contact_zone", "")),
294362
)
295-
plan.validate()
296-
return plan
297363

298364

299365
@dataclass

0 commit comments

Comments
 (0)