diff --git a/desdeo/api/models/generic_states.py b/desdeo/api/models/generic_states.py index 7baeb2a81..26054a1ce 100644 --- a/desdeo/api/models/generic_states.py +++ b/desdeo/api/models/generic_states.py @@ -17,6 +17,9 @@ from desdeo.problem import Tensor, VariableType +from .nautilus import ( + NautilusState, +) from .state import ( EMOFetchState, EMOIterateState, @@ -64,6 +67,8 @@ class StateKind(str, Enum): GENERIC_INTERMEDIATE = "generic.solve_intermediate" ENAUTILUS_STEP = "e-nautilus.stepping" ENAUTILUS_FINAL = "e-nautilus.final" + NAUTILUS_NAVIGATE = "nautilus.navigate" + NAUTILUS_INITIALIZE = "nautilus.initialize" class State(SQLModel, table=True): @@ -155,6 +160,63 @@ def create( return row + # @classmethod + # def create( + # cls, + # database_session: Session, + # *, + # kind: StateKind, + # problem_id: int | None = None, + # session_id: int | None = None, + # parent_id: int | None = None, + # state: SQLModel | None = None, + # ) -> "StateDB": + # """Create a new StateDB entry and corresponding substate entry. + + # Args: + # database_session: Database session + # kind: Explicit StateKind of this state. + # problem_id: Required for root states. + # session_id: Required for child states. + # parent_id: Parent state ID if this is a child. + # state: The substate SQLModel instance. + # """ + # if state is None: + # raise ValueError("State (substate) must be provided.") + + # if parent_id is None: + # # Root state + # if problem_id is None: + # raise ValueError("Root state requires problem_id.") + # state_db = cls( + # problem_id=problem_id, + # session_id=None, + # parent_id=None, + # kind=kind, + # ) + # else: + # # Child state + # if session_id is None: + # raise ValueError("Child state requires session_id.") + # state_db = cls( + # problem_id=None, + # session_id=session_id, + # parent_id=parent_id, + # kind=kind, + # ) + + # database_session.add(state_db) + # database_session.commit() + # database_session.refresh(state_db) + + # # Link substate + # state.state_id = state_db.id + # database_session.add(state) + # database_session.commit() + # database_session.refresh(state_db) + + # return state_db + @property def state(self) -> SQLModel | None: """Return the concrete substate instance (e.g., NIMBUSSaveState)... @@ -194,6 +256,8 @@ def state(self) -> SQLModel | None: StateKind.GENERIC_INTERMEDIATE: IntermediateSolutionState, StateKind.ENAUTILUS_STEP: ENautilusState, StateKind.ENAUTILUS_FINAL: ENautilusFinalState, + StateKind.NAUTILUS_NAVIGATE: NautilusState, + StateKind.NAUTILUS_INITIALIZE: NautilusState } SUBSTATE_TO_KIND: dict[SQLModel, StateKind] = { @@ -212,6 +276,8 @@ def state(self) -> SQLModel | None: IntermediateSolutionState: StateKind.GENERIC_INTERMEDIATE, ENautilusState: StateKind.ENAUTILUS_STEP, ENautilusFinalState: StateKind.ENAUTILUS_FINAL, + # NautilusState: StateKind.NAUTILUS_NAVIGATE, + # NautilusState: StateKind.NAUTILUS_INITIALIZE } diff --git a/desdeo/api/models/nautilus.py b/desdeo/api/models/nautilus.py new file mode 100644 index 000000000..4db5e6281 --- /dev/null +++ b/desdeo/api/models/nautilus.py @@ -0,0 +1,63 @@ +from sqlmodel import JSON, Column, Field, SQLModel + + +class NautilusState(SQLModel, table=True): + """Concrete NAUTILUS Navigator state stored for a single interaction step. + + This model stores the full algorithmic state returned by the + NAUTILUS Navigator after either initialization or a navigation step. + + The instance is linked to a base `StateDB` entry, which defines the + problem context and state type (e.g. "nautilus.initialize" or + "nautilus.navigate"). This table contains only the algorithm-specific + data required to reconstruct the navigation process at that step. + + Attributes: + id (int | None): Primary key of this NAUTILUS state entry. + objective_symbols (list[str]): Short symbolic names of the objectives. + objective_long_names (list[str]): Descriptive names of the objectives. + units (list[str] | None): Units of the objectives, if defined. None if unitless. + is_maximized (list[bool]): Boolean flags indicating whether each objective + is to be maximized (True) or minimized (False). + ideal (list[float]): Ideal objective values of the problem. + nadir (list[float]): Nadir objective values of the problem. + lower_bounds (dict[str, list[float]]): Lower bounds of the reachable region per objective + across navigation steps. + upper_bounds (dict[str, list[float]]): Upper bounds of the reachable region per objective + across navigation steps. + preferences (dict[str, list[float]]): Preference values provided by the decision maker + for each navigation step. + bounds (dict[str, list[float]]): Bound preferences provided by the decision maker + for each navigation step. + total_steps (int): Total number of steps allowed in this NAUTILUS session. + current_step (int): Current navigation step index. + remaining_steps (int): Number of steps remaining in the navigation process. + reachable_solution (dict[str, float]): The objective values of the currently reachable solution. + """ + + __tablename__ = "nautilus_states" + + id: int | None = Field(default=None, primary_key=True) + + # Problem meta + objective_symbols: list[str] = Field(sa_column=Column(JSON)) + objective_long_names: list[str] = Field(sa_column=Column(JSON)) + units: list[str] | None = Field(default=None, sa_column=Column(JSON)) + is_maximized: list[bool] = Field(sa_column=Column(JSON)) + + # Problem bounds + ideal: list[float] = Field(sa_column=Column(JSON)) + nadir: list[float] = Field(sa_column=Column(JSON)) + + # Navigation data + lower_bounds: dict[str, list[float]] = Field(sa_column=Column(JSON)) + upper_bounds: dict[str, list[float]] = Field(sa_column=Column(JSON)) + preferences: dict[str, list[float]] = Field(sa_column=Column(JSON)) + bounds: dict[str, list[float]] = Field(sa_column=Column(JSON)) + + total_steps: int + current_step: int + remaining_steps: int + + # Final reachable solution + reachable_solution: dict[str, float] = Field(sa_column=Column(JSON)) diff --git a/desdeo/api/schemas/__init__.py b/desdeo/api/schemas/__init__.py new file mode 100644 index 000000000..36827e27c --- /dev/null +++ b/desdeo/api/schemas/__init__.py @@ -0,0 +1,6 @@ +from .nautilus import ( + NautilusInitialResponse, + NautilusInitRequest, + NautilusNavigateRequest, + NautilusResponse, +) diff --git a/desdeo/api/schemas/nautilus.py b/desdeo/api/schemas/nautilus.py new file mode 100644 index 000000000..49543fa7d --- /dev/null +++ b/desdeo/api/schemas/nautilus.py @@ -0,0 +1,59 @@ +from typing import Literal + +from pydantic import BaseModel, Field + + +# Requests +class NautilusInitRequest(BaseModel): + """Request to initialize a NAUTILUS Navigator session for a specific problem.""" + + problem_id: int = Field(..., description="The ID of the problem to navigate.") + total_steps: int = Field(100, description="The total number of steps in the NAUTILUS Navigator.") + +class NautilusNavigateRequest(BaseModel): + """Request to navigate a NAUTILUS Navigator session.""" + + problem_id: int = Field(..., description="The ID of the problem to navigate.") + preference: dict[str, float] = Field(..., description="The preference of the Decision Maker (DM) for each objective.") + bounds: dict[str, float] = Field(..., description="The bounds preference of the DM for each objective.") + go_back_step: int = Field(..., description="The step index to go back in the navigation history.") + steps_remaining: int = Field(..., description="The number of steps remaining in the navigation process.") + +# Responses +class NautilusInitialResponse(BaseModel): + """Response returned by the NAUTILUS Navigator when initialized.""" + + state_id: int = Field(..., description="The ID of this navigation state.") + response_type: Literal["nautilus.initialize"] = "nautilus.initialize" + parent_state_id: int | None = Field(None, description="Parent state ID, if this is a child step.") + + objective_symbols: list[str] = Field(..., description="The symbols of the objectives.") + objective_long_names: list[str] = Field(..., description="Long/descriptive names of the objectives.") + units: list[str] | None = Field(None, description="The units of the objectives, empty if unitless.") + is_maximized: list[bool] = Field(..., description="Whether each objective is to be maximized.") + ideal: list[float] = Field(..., description="The ideal values of the objectives.") + nadir: list[float] = Field(..., description="The nadir values of the objectives.") + total_steps: int = Field(..., description="The total number of steps in this NAUTILUS session.") + + +class NautilusNavigateResponse(BaseModel): + """Response returned by the NAUTILUS Navigator during navigation (modern ENautilus style).""" + + state_id: int = Field(..., description="The ID of this navigation state.") + response_type: Literal["nautilus.navigate"] = "nautilus.navigate" + parent_state_id: int | None = Field(None, description="Parent state ID, if this is a child step.") + + objective_symbols: list[str] = Field(..., description="The symbols of the objectives.") + objective_long_names: list[str] = Field(..., description="Long/descriptive names of the objectives.") + units: list[str] | None = Field(None, description="The units of the objectives, empty if unitless.") + is_maximized: list[bool] = Field(..., description="Whether each objective is to be maximized.") + ideal: list[float] = Field(..., description="The ideal values of the objectives.") + nadir: list[float] = Field(..., description="The nadir values of the objectives.") + + lower_bounds: dict[str, list[float]] = Field(..., description="Lower bounds of the reachable region per objective.") + upper_bounds: dict[str, list[float]] = Field(..., description="Upper bounds of the reachable region per objective.") + preferences: dict[str, list[float]] = Field(..., description="Preferences used in each step per objective.") + bounds: dict[str, list[float]] = Field(..., description="Bounds used in each step per objective.") + + total_steps: int = Field(..., description="The total number of steps in the current navigation path.") + reachable_solution: dict[str, float] = Field(..., description="The solution reached at the end of navigation.")