From 0728c21520535d132447aab3c392d860b9ccf7e6 Mon Sep 17 00:00:00 2001 From: Peter Bednarik Date: Mon, 23 Feb 2026 15:40:39 +0200 Subject: [PATCH] Nautilus - models and schemas for /navigate and /initialize. --- desdeo/api/models/nautilus.py | 44 ++++++++++++++++++++++++ desdeo/api/schemas/__init__.py | 6 ++++ desdeo/api/schemas/nautilus.py | 61 ++++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+) create mode 100644 desdeo/api/models/nautilus.py create mode 100644 desdeo/api/schemas/__init__.py create mode 100644 desdeo/api/schemas/nautilus.py diff --git a/desdeo/api/models/nautilus.py b/desdeo/api/models/nautilus.py new file mode 100644 index 000000000..027697050 --- /dev/null +++ b/desdeo/api/models/nautilus.py @@ -0,0 +1,44 @@ +import datetime + +from sqlalchemy import JSON, Column, DateTime, ForeignKey, Integer, String + +from desdeo.api.db import Base + + +class NautilusStateDB(Base): + """Database model storing the state of a NAUTILUS interactive navigation session. + + Each record represents a single interaction step within a NAUTILUS session, + such as initialization, navigation, or finalization. The table supports + branching session trees by maintaining parent-child relationships between states. + + Attributes: + id (int): Primary key identifying this state entry. + session_id (int): Identifier of the interactive NAUTILUS session. + parent_state_id (int | None): Foreign key referencing the parent state. + Null for root (initialization) nodes. + request (JSON): Serialized Nautilus request model used to generate this state. + response (JSON): Serialized Nautilus response model produced by the algorithm. + node_type (str): Type of interaction that generated the state. + Expected values include: + - "initialize" + - "navigate" + - "final" + depth (int): Depth of this node within the session tree (root = 0). + created_at (datetime): Timestamp when this state was created. + """ + + __tablename__ = "nautilus_states" + + id = Column(Integer, primary_key=True) + session_id = Column(Integer, nullable=False) + parent_state_id = Column(Integer, ForeignKey("nautilus_states.id"), nullable=True) + + request = Column(JSON, nullable=False) + response = Column(JSON, nullable=False) + + node_type = Column(String, nullable=False) # "initialize", "navigate", "final" + + depth = Column(Integer, nullable=False) + + created_at = Column(DateTime, default=datetime.utcnow) 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..90299857d --- /dev/null +++ b/desdeo/api/schemas/nautilus.py @@ -0,0 +1,61 @@ +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" + node_type: Literal["initialize"] = "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" + node_type: Literal["navigate", "final"] = "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 = Field(..., description="The solution reached at the end of navigation.") \ No newline at end of file