Source code for suboptimumg.sweep.pack_optimizer_sweep

import time
import warnings
from token import OP
from typing import List, Optional

import numpy as np
from click import Option
from scipy.optimize import differential_evolution

from ..compsim import energy_data
from ..compsim.competition_factory import from_data
from ..compsim.models import CompetitionData
from .pack_optimizer_models import *
from .pack_optimizer_results import *




class PackOptimizer:
    """Differential-evolution optimizer for vehicle pack and driving parameters."""

    def __init__(self, comp_data: CompetitionData, config: PackOptimizerConfig):
        self.comp_data = comp_data
        self.config = config

        self.best_score: float = -np.inf
        self.best_valaution: Optional[EvaluationRecord] = None
        self.history: List[EvaluationRecord] = []



    def sweep(self) -> PackOptimizerResults:
        """Run the optimization and return a PackOptimizerResults object.

        Returns
        -------
        PackOptimizerResults
        """

        print(f"Starting optimization with {len(self.config.parameters)} parameters")
        print(f"Max iterations:  {self.config.max_iterations}")
        print(f"Population size: {self.config.population_size}")

        # Convert our SweepParamConfig list to scipy DE bounds format
        bounds = [(p.min, p.max) for p in self.config.parameters]

        de_result = differential_evolution(
            self._objective,
            bounds,
            maxiter=self.config.max_iterations,
            popsize=self.config.population_size,
            seed=self.config.seed,
        )

        if de_result.success:
            print(f"Optimization completed successfully.")
        else:
            print(f"Optimization did not converge: {de_result.message}")

        data = PackOptimizationData(
            optimal_evaluation=self.best_record,
            evaluation_history=self.history,
        )

        return PackOptimizerResults(data, self.config)


    def _objective(self, params: np.ndarray) -> float:
        """
        Simulates a competition with the given car parameter modifications. Then, calculates
        an objective score based on the competition results and capacity constraints

        Parameters
        -----------
        params : np.ndarray
            Array of parameter values corresponding to self.config.parameters order (DE passes a numpy array)
            e.g. if parameters=[mass, ratio], then params[0] is mass,
            params[1] is ratio, etc.

        Returns
        -------
        float
            The penalized score to be minimized by the optimizer (negative total points minus any penalties)
        """
        try:
            comp = from_data(self.comp_data)

            # Apply each optimized parameter
            for param, value in zip(self.config.parameters, params):
                comp.mycar.modify_params(param.name, float(value))

            # Apply fixed mass and capacity when in fixed pack mode
            if isinstance(self.config, FixedPackConfig):
                capacity_adjusted_mass = comp.mycar.accum.capacity_to_mass(
                    self.config.capacity
                )
                comp.mycar.modify_params("mass", capacity_adjusted_mass)
                comp.mycar.modify_params("accum.capacity", self.config.capacity)

            result = comp.run()

            # Penalized score
            energy_consumption_overage = result.endu_consumed_energy - (
                comp.mycar.accum.params.capacity - self.config.capacity_margin
            )
            penalty = 1000 * max(0, energy_consumption_overage)

            score = result.total_points - penalty

            record = EvaluationRecord(
                params=params,
                param_names=[p.name for p in self.config.parameters],
                mass=comp.mycar.params.mass,
                capacity=comp.mycar.accum.params.capacity,
                score=score,
                competition_results=result,
                feasible=energy_consumption_overage <= 0,
            )
            self.history.append(record)

            # Update best score and record if this is the best so far
            if score > self.best_score:
                self.best_score = score
                self.best_record = record

            # We return the negative score because DE minimizes the objective, but we want to maximize points
            return -score

        except Exception as e:
            print(f"Evaluation failed with params {params}: {e}")
            record = EvaluationRecord(
                params=params,
                param_names=[p.name for p in self.config.parameters],
                mass=np.nan,
                capacity=np.nan,
                score=-np.inf,
                competition_results=None,
                feasible=False,
            )
            self.history.append(record)
            return np.inf