Spaces:
Sleeping
Sleeping
File size: 7,263 Bytes
8dc575c a54759f 8dc575c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """
models.py — Typed Pydantic models for the Meta Quest OpenEnv environment.
All state, actions, observations, and responses are defined here.
"""
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class SensorStatus(str, Enum):
ACTIVE = "Active"
DRIFTING = "Drifting"
FAILED = "Failed"
RECALIBRATING = "Recalibrating"
class KernelStatus(str, Enum):
ACTIVE = "Active"
DEGRADED = "Degraded"
PANIC = "Panic"
SAFE_MODE = "SafeMode"
RECOVERING = "Recovering"
class TrackingMode(str, Enum):
SLAM = "SLAM"
VIO = "VIO" # Visual-Inertial Odometry
IMU_ONLY = "IMU_ONLY"
DEGRADED = "Degraded"
class TaskID(str, Enum):
THERMAL_MITIGATION = "thermal_mitigation"
SENSOR_RECOVERY = "sensor_recovery"
KERNEL_PANIC_RECOVERY = "kernel_panic_recovery"
# ---------------------------------------------------------------------------
# Sub-models: Hardware Components
# ---------------------------------------------------------------------------
class ThermalState(BaseModel):
cpu_temp_c: float = Field(..., description="CPU temperature in Celsius")
gpu_temp_c: float = Field(..., description="GPU temperature in Celsius")
optical_pod_temp_c: float = Field(..., description="Optical pod temperature in Celsius")
throttle_active: bool = Field(False, description="Whether thermal throttle is engaged")
refresh_rate_hz: int = Field(90, description="Current display refresh rate (72/90/120 Hz)")
fan_speed_pct: float = Field(0.0, description="Fan speed as percentage (0-100)")
class SensorState(BaseModel):
imu_status: SensorStatus = Field(SensorStatus.ACTIVE, description="IMU sensor status")
imu_latency_ms: float = Field(2.0, description="IMU read latency in ms")
lidar_status: SensorStatus = Field(SensorStatus.ACTIVE, description="LiDAR sensor status")
lidar_latency_ms: float = Field(5.0, description="LiDAR read latency in ms")
hand_tracking_status: SensorStatus = Field(SensorStatus.ACTIVE, description="Hand tracking status")
hand_tracking_latency_ms: float = Field(8.0, description="Hand tracking latency in ms")
tracking_mode: TrackingMode = Field(TrackingMode.SLAM, description="Active tracking mode")
tracking_stability: float = Field(1.0, ge=0.0, le=1.0, description="Stability score 0-1")
class ResourceState(BaseModel):
battery_pct: float = Field(80.0, ge=0.0, le=100.0, description="Battery percentage")
ram_used_mb: float = Field(..., description="RAM usage in MB")
ram_total_mb: float = Field(8192.0, description="Total RAM in MB")
cpu_usage_pct: float = Field(..., description="CPU usage percentage from host")
frame_buffer_healthy: bool = Field(True, description="Frame buffer health status")
background_processes: int = Field(5, description="Number of non-essential background processes")
class SpatialAnchorState(BaseModel):
anchor_count: int = Field(12, description="Number of active spatial anchors")
anchor_stability: float = Field(1.0, ge=0.0, le=1.0, description="Average anchor stability score")
world_lock_active: bool = Field(True, description="Whether world anchor memory lock is active")
session_data_intact: bool = Field(True, description="Whether previous session data is intact")
cached_anchors_cleared: bool = Field(False, description="Whether cached anchors have been cleared")
class KernelState(BaseModel):
status: KernelStatus = Field(KernelStatus.ACTIVE, description="Kernel operational status")
crashed_app_id: Optional[int] = Field(None, description="App ID that caused crash (if any)")
safe_mode_initiated: bool = Field(False, description="Whether safe mode boot was initiated")
memory_lock_cleared: bool = Field(False, description="Whether memory lock has been cleared")
uptime_seconds: float = Field(0.0, description="System uptime in seconds")
# ---------------------------------------------------------------------------
# Full System Observation
# ---------------------------------------------------------------------------
class SystemObservation(BaseModel):
"""Complete observable state of the Meta Quest OS simulation."""
step: int = Field(0, description="Current step number")
task_id: TaskID = Field(..., description="Active task identifier")
thermals: ThermalState
sensors: SensorState
resources: ResourceState
spatial_anchors: SpatialAnchorState
kernel: KernelState
last_action: Optional[str] = Field(None, description="Last action taken by agent")
last_action_success: bool = Field(True, description="Whether last action succeeded")
last_action_message: str = Field("", description="Human-readable result of last action")
reward: float = Field(0.0, ge=0.0, le=1.0, description="Current step reward")
cumulative_reward: float = Field(0.0, description="Total reward accumulated")
done: bool = Field(False, description="Whether episode is complete")
info: Dict[str, Any] = Field(default_factory=dict, description="Extra diagnostic info")
# ---------------------------------------------------------------------------
# Action Models
# ---------------------------------------------------------------------------
class ActionRequest(BaseModel):
"""Agent action submitted to the environment."""
action: str = Field(..., description="Action command string")
task_id: Optional[TaskID] = Field(None, description="Task context (optional)")
class ActionResult(BaseModel):
"""Result returned after executing an action."""
observation: SystemObservation
reward: float = Field(..., ge=0.0, le=1.0)
done: bool
info: Dict[str, Any] = Field(default_factory=dict)
# ---------------------------------------------------------------------------
# Reset & State Models
# ---------------------------------------------------------------------------
class ResetRequest(BaseModel):
"""Request to reset the environment to a task scenario."""
task_id: TaskID = Field(TaskID.THERMAL_MITIGATION, description="Which task scenario to initialize")
seed: Optional[int] = Field(None, description="Random seed for reproducibility")
class ResetResult(BaseModel):
"""Result of a reset call."""
observation: SystemObservation
task_id: TaskID
task_description: str
available_actions: List[str]
max_steps: int
class StateResult(BaseModel):
"""Current environment state (read-only snapshot)."""
observation: SystemObservation
task_id: TaskID
step: int
done: bool
# ---------------------------------------------------------------------------
# Task Info
# ---------------------------------------------------------------------------
class TaskInfo(BaseModel):
task_id: TaskID
name: str
description: str
difficulty: str
max_steps: int
success_threshold: float
available_actions: List[str]
class HealthResponse(BaseModel):
status: str = "ok"
version: str = "1.0.0"
environment: str = "MetaQuestOSEnv" |