Spaces:
Sleeping
Sleeping
File size: 2,006 Bytes
cb330aa | 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 | """
Core models and shared type aliases for SYNAPSE-X.
"""
from typing import Any, Literal
from pydantic import BaseModel, Field, PrivateAttr
ActionType = Literal["execute", "delay", "reallocate"]
ActionPayload = dict[str, Any]
class Task(BaseModel):
id: int
name: str
dependencies: list[int] = Field(default_factory=list, description="IDs of tasks that must complete first")
release_time: int = Field(0, ge=0, description="Step when the task becomes actionable")
released: bool = True
priority: float = Field(..., ge=0.0, le=1.0, description="Task priority 0-1")
risk: float = Field(..., ge=0.0, le=1.0, description="Execution risk 0-1")
uncertainty: float = Field(..., ge=0.0, le=1.0, description="PRISM uncertainty 0-1")
deadline: float = Field(..., description="Steps remaining until deadline")
completed: bool = False
failed: bool = False
future_risk: float = Field(0.0, ge=0.0, description="ECHO predicted future risk")
deadline_pressure: float = Field(0.0, ge=0.0, le=1.0, description="ECHO deadline pressure")
resources_required: float = Field(0.1, ge=0.0, le=1.0, description="Resource cost to execute")
delay_count: int = 0
_history_risk: float = PrivateAttr(default=0.0)
@property
def is_terminal(self) -> bool:
return self.completed or self.failed
@property
def is_active(self) -> bool:
return self.released and not self.is_terminal
class Observation(BaseModel):
tasks: list[Task]
time: int
resources: float = Field(..., ge=0.0, le=1.0, description="Available resources 0-1")
episode_done: bool = False
class Action(BaseModel):
action_type: ActionType
task_id: int
class StepResult(BaseModel):
observation: Observation
reward: float
done: bool
info: dict[str, Any]
class GradeResult(BaseModel):
score: float = Field(..., ge=0.0, le=1.0)
completion_rate: float
efficiency: float
reward_score: float
details: dict[str, Any]
|