Skip to content

Data Model

data_model

Core data model for optimization results.

This module defines the main OptimizationResult class that holds all optimization data and metadata in a structured, extensible format.

Classes

ProblemMetadata

Bases: BaseModel

Metadata about the optimization problem.

OptimizationResult

OptimizationResult(problem_metadata: ProblemMetadata | None = None, design_variables: Any = None, objectives: Any = None, inequality_constraints: Any = None, equality_constraints: Any = None, observables: Any = None, custom_data: dict[str, Any] | None = None)

Main container for optimization results.

This class holds all data from an optimization run including design variables, objectives, constraints, observables, and any custom data.

Initialize optimization result.

Narwhals is used here to accept Pandas, Polars, or generic iterables and normalize them to the internal Polars storage format.

Source code in optiscope/core/data_model.py
def __init__(
    self,
    problem_metadata: ProblemMetadata | None = None,
    design_variables: Any = None,
    objectives: Any = None,
    inequality_constraints: Any = None,
    equality_constraints: Any = None,
    observables: Any = None,
    custom_data: dict[str, Any] | None = None,
) -> None:
    """
    Initialize optimization result.

    Narwhals is used here to accept Pandas, Polars, or generic iterables
    and normalize them to the internal Polars storage format.
    """
    self.problem_metadata = problem_metadata or ProblemMetadata(name="Unnamed Problem")  # type: ignore

    # Data storage - internal storage uses Polars
    # We use a helper method that leverages Narwhals for conversion
    self._design_variables = self._normalize_to_frame(design_variables)
    self._objectives = self._normalize_to_frame(objectives)
    self._inequality_constraints = self._normalize_to_frame(inequality_constraints)
    self._equality_constraints = self._normalize_to_frame(equality_constraints)
    self._observables = self._normalize_to_frame(observables)

    self._custom_data: dict[str, pl.DataFrame] = {}
    if custom_data:
        for k, v in custom_data.items():
            self._custom_data[k] = self._normalize_to_frame(v)

    # Variable metadata storage
    self._variable_metadata: dict[str, AnyVariableMetadata] = {}

    # Set management
    self._set_manager = SetManager()

    # Validate consistency
    self._validate_data()
Functions
to_dict
to_dict() -> dict[str, Any]

Serialize the OptimizationResult to a dictionary.

Source code in optiscope/core/data_model.py
def to_dict(self) -> dict[str, Any]:
    """Serialize the OptimizationResult to a dictionary."""

    def serialize_df(df: pl.DataFrame) -> str | None:
        if df.is_empty():
            return None
        # Narwhals doesn't handle JSON serialization formats, so we stick to
        # Polars -> Pandas -> JSON for compatibility with the existing schema
        return df.to_pandas().to_json(orient="split", date_format="iso")

    return {
        "problem_metadata": self.problem_metadata.model_dump(mode="json"),
        "design_variables": serialize_df(self._design_variables),
        "objectives": serialize_df(self._objectives),
        "inequality_constraints": serialize_df(self._inequality_constraints),
        "equality_constraints": serialize_df(self._equality_constraints),
        "observables": serialize_df(self._observables),
        "custom_data": {name: serialize_df(df) for name, df in self._custom_data.items()},
        "variable_metadata": {
            name: meta.model_dump(mode="json") for name, meta in self._variable_metadata.items()
        },
        "set_manager": self._set_manager.to_dict(),
    }
from_dict classmethod
from_dict(data: dict[str, Any]) -> OptimizationResult

Deserialize an OptimizationResult from a dictionary.

Source code in optiscope/core/data_model.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> OptimizationResult:
    """Deserialize an OptimizationResult from a dictionary."""

    def deserialize_df(json_str: str | None) -> pd.DataFrame:
        if json_str is None:
            return pd.DataFrame()
        df = pd.read_json(StringIO(json_str), orient="split")
        # Ensure date columns are parsed correctly if any
        for col in df.columns:
            if pd.api.types.is_object_dtype(df[col]):
                try:
                    df[col] = pd.to_datetime(df[col], errors="raise")
                except (ValueError, TypeError):
                    pass
        return df

    metadata_cls_map: dict[DataTypeCategory, type[AnyVariableMetadata]] = {
        DataTypeCategory.DESIGN_VARIABLE: DesignVariable,
        DataTypeCategory.OBJECTIVE: Objective,
        DataTypeCategory.CONSTRAINT: Constraint,
        DataTypeCategory.OBSERVABLE: Observable,
    }

    variable_metadata = {}
    for name, meta_data in data.get("variable_metadata", {}).items():
        category = DataTypeCategory(meta_data.get("category"))
        cls_ = metadata_cls_map.get(category)
        if cls_:
            variable_metadata[name] = cls_(**meta_data)

    result = cls(
        problem_metadata=ProblemMetadata(**data["problem_metadata"]),
        design_variables=deserialize_df(data.get("design_variables")),
        objectives=deserialize_df(data.get("objectives")),
        inequality_constraints=deserialize_df(data.get("inequality_constraints")),
        equality_constraints=deserialize_df(data.get("equality_constraints")),
        observables=deserialize_df(data.get("observables")),
        custom_data={
            name: deserialize_df(df_json)
            for name, df_json in data.get("custom_data", {}).items()
        },
    )

    result._variable_metadata = variable_metadata

    if "set_manager" in data:
        result._set_manager.from_dict(data["set_manager"])

    return result
rename
rename(new_name: str) -> None

Rename the optimization problem.

Source code in optiscope/core/data_model.py
def rename(self, new_name: str) -> None:
    """Rename the optimization problem."""
    self.problem_metadata.name = new_name
add_column
add_column(name: str, data: Any, category: DataTypeCategory = OBSERVABLE, metadata: AnyVariableMetadata | None = None) -> None

Add a new column of data to the optimization result. Uses Narwhals to handle input formats (list, np.array, pd.Series, pl.Series) identically.

Source code in optiscope/core/data_model.py
def add_column(
    self,
    name: str,
    data: Any,
    category: DataTypeCategory = DataTypeCategory.OBSERVABLE,
    metadata: AnyVariableMetadata | None = None,
) -> None:
    """
    Add a new column of data to the optimization result.
    Uses Narwhals to handle input formats (list, np.array, pd.Series, pl.Series) identically.
    """
    # 1. Normalize input to Polars Series using Narwhals
    if isinstance(data, (list, np.ndarray)):
        series = pl.Series(name, data)
    else:
        # nw.from_native(..., series_only=True) handles pd.Series, pl.Series, etc.
        series = nw.from_native(data, series_only=True).to_polars().rename(name)

    current_n_points = self.n_points
    if current_n_points > 0 and series.len() != current_n_points:
        raise ValueError(
            f"Data length {series.len()} does not match result length {current_n_points}"
        )

    # 2. Create default metadata if not provided
    if metadata is None:
        if category == DataTypeCategory.DESIGN_VARIABLE:
            metadata = DesignVariable(name=name)
        elif category == DataTypeCategory.OBJECTIVE:
            metadata = Objective(name=name)
        elif category == DataTypeCategory.CONSTRAINT:
            metadata = Constraint(name=name)
        elif category == DataTypeCategory.OBSERVABLE:
            metadata = Observable(name=name)
        elif category == DataTypeCategory.CUSTOM:
            metadata = CustomDataType(name=name, custom_type="generic")
        else:
            raise ValueError(f"Unknown category: {category}")

    # Validate metadata matches category and name
    if metadata.category != category:
        raise ValueError(
            f"Metadata category {metadata.category} does not match provided category {category}"
        )

    if metadata.name != name:
        raise ValueError(f"Metadata name {metadata.name} does not match provided name {name}")

    # 3. Add to appropriate DataFrame
    # Note: We use direct polars assignment here because internal storage is concrete Polars
    if category == DataTypeCategory.DESIGN_VARIABLE:
        self._design_variables = self._design_variables.with_columns(series)
    elif category == DataTypeCategory.OBJECTIVE:
        self._objectives = self._objectives.with_columns(series)
    elif category == DataTypeCategory.CONSTRAINT:
        if isinstance(metadata, Constraint):
            if metadata.constraint_type == ConstraintType.EQUALITY:
                self._equality_constraints = self._equality_constraints.with_columns(series)
            else:
                self._inequality_constraints = self._inequality_constraints.with_columns(series)
        else:
            raise ValueError("Metadata must be instance of Constraint for CONSTRAINT category")
    elif category == DataTypeCategory.OBSERVABLE:
        self._observables = self._observables.with_columns(series)
    elif category == DataTypeCategory.CUSTOM:
        if isinstance(metadata, CustomDataType):
            table_name = metadata.custom_type
            if table_name not in self._custom_data:
                self._custom_data[table_name] = pl.DataFrame()
            self._custom_data[table_name] = self._custom_data[table_name].with_columns(series)
        else:
            raise ValueError("Metadata must be instance of CustomDataType for CUSTOM category")

    self.add_variable_metadata(metadata)
get_set_data
get_set_data(set_name: str, category: DataTypeCategory | None = None) -> DataFrame

Get data for a specific set as Pandas DataFrame.

Source code in optiscope/core/data_model.py
def get_set_data(self, set_name: str, category: DataTypeCategory | None = None) -> pd.DataFrame:
    """
    Get data for a specific set as Pandas DataFrame.
    """
    result_set = self.get_set(set_name)
    indices = result_set.indices

    # We collect frames in native Polars first
    data_frames = []

    if category is None or category == DataTypeCategory.DESIGN_VARIABLE:
        if not self._design_variables.is_empty():
            data_frames.append(self._design_variables[indices])

    if category is None or category == DataTypeCategory.OBJECTIVE:
        if not self._objectives.is_empty():
            data_frames.append(self._objectives[indices])

    if category is None or category == DataTypeCategory.CONSTRAINT:
        if not self._inequality_constraints.is_empty():
            data_frames.append(self._inequality_constraints[indices])
        if not self._equality_constraints.is_empty():
            data_frames.append(self._equality_constraints[indices])

    if category is None or category == DataTypeCategory.OBSERVABLE:
        if not self._observables.is_empty():
            data_frames.append(self._observables[indices])

    if not data_frames:
        return pd.DataFrame()

    # Use Narwhals to concatenate regardless of what backend we might use in future
    # Convert all to Narwhals DataFrame -> Concat -> Convert to Pandas
    nw_frames = [nw.from_native(df) for df in data_frames]
    return nw.concat(nw_frames, how="horizontal").to_pandas()
get_all_data
get_all_data() -> DataFrame

Get all data as a single Pandas DataFrame.

Source code in optiscope/core/data_model.py
def get_all_data(self) -> pd.DataFrame:
    """Get all data as a single Pandas DataFrame."""
    data_frames = []

    # Collect all potential frames
    candidates = [
        self._design_variables,
        self._objectives,
        self._inequality_constraints,
        self._equality_constraints,
        self._observables,
    ]

    # Add custom data values
    candidates.extend(self._custom_data.values())

    for df in candidates:
        # FIX: Use len() which works for both Polars and Pandas.
        # Polars .is_empty() crashes on Pandas; Pandas .empty crashes on Polars (if used as method).
        if len(df) > 0:
            data_frames.append(df)

    if not data_frames:
        return pd.DataFrame()

    # Narwhals horizontal concatenation
    # This part handles mixed Pandas/Polars inputs correctly
    nw_frames = [nw.from_native(df) for df in data_frames]
    return nw.concat(nw_frames, how="horizontal").to_pandas()