API Reference

@dataclass(slots=True)
class Mission:
    """Top-level user-facing mission container."""

    phases: list[Phase] = field(default_factory=list)
    spacecraft: dict[str, Spacecraft] = field(default_factory=dict)
    name: str = "Mission"
    initial_epoch: str | datetime | float | None = None

    objectives: list[Objective] = field(default_factory=lambda: [minimize_total_delta_v()])

    plan: RunPlan = field(default_factory=RunPlan.default)
    retry: RetryPolicy = field(default_factory=RetryPolicy.default)
    solve_config: SolveConfig = field(default_factory=SolveConfig)
    solver_options: SolverOptions = field(default_factory=SolverOptions)

    mesh_nsegs_transfer: int = 60
    mesh_nsegs_precoast: int = 30
    lambert_grid_size: int = 60
    nrevs_to_try: tuple[int, ...] = (0, 1)
    precoast_grid_size: int = 10
    limit_precoast_to_one_period: bool = True
    w_time: float = 0.0

    def __post_init__(self) -> None:
        self.phases = list(self.phases)

    def add_phase(self, phase: Phase) -> None:
        """Append a phase to the mission.

        Args:
            phase: Phase to append.
        """
        self.phases.append(phase)

    def validate(self) -> None:
        """Validate mission structure before solving.

        Raises:
            ValueError: If the mission has no phases or a phase is invalid.
        """
        if not self.phases:
            raise ValueError("Mission has no phases.")
        for phase in self.phases:
            phase.validate()

    def solve(
        self,
        *,
        plan: RunPlan | None = None,
        retry: RetryPolicy | None = None,
        solve_config: SolveConfig | None = None,
        solver_options: SolverOptions | None = None,
    ) -> Solution:
        """Solve the mission using the configured runner.

        Args:
            plan: Optional continuation or staging plan override.
            retry: Optional retry-policy override.
            solve_config: Optional solve-configuration override.
            solver_options: Optional backend solver-options override.

        Returns:
            The mission solution wrapper.
        """
        mission_runner = MissionRunner(
            solve_options=solver_options or self.solver_options,
            solve_config=solve_config or self.solve_config,
            plan=plan or self.plan,
            retry=retry or self.retry,
        )
        return mission_runner.solve(self)

class MissionBuildError(ValueError):
    """Raised when a mission cannot be mapped to a supported solver."""

@dataclass(slots=True)
class MissionRunner:
    """Coordinate mission validation, backend selection, and retry behavior."""

    solve_options: SolverOptions
    solve_config: SolveConfig = field(default_factory=SolveConfig)
    plan: RunPlan = field(default_factory=RunPlan.default)
    retry: RetryPolicy = field(default_factory=RetryPolicy.default)

    def solve(self, mission: Mission) -> Solution:
        """Solve a mission through validation, staging, backend dispatch, and retry.

        Parameters
        ----------
        mission
            Mission to validate, compile, and solve.

        Returns
        -------
        Solution
            Solver result plus an attempt log describing each stage and retry.
        """
        self._validate_mission(mission)
        backend = _select_backend(mission)
        attempt_logs: list[AttemptLog] = []
        last_error: str | None = None
        last_result: RendezvousResult | None = None

        for stage_index, stage in enumerate(_runner_stages(self.plan)):
            stage_label = _stage_label(stage)
            stage_problem = _build_stage_problem(mission, backend, stage)

            for attempt_index in range(1, self._max_attempts() + 1):
                result = None
                try:
                    result = _solve_stage_problem(stage_problem, self.solve_options)
                except Exception as exc:  # noqa: BLE001
                    last_result = None
                    last_error = str(exc)
                    attempt_logs.append(
                        AttemptLog(
                            stage=stage_label,
                            attempt=attempt_index,
                            status="fail",
                            message=last_error,
                        )
                    )
                    if not self._should_retry(attempt_index):
                        break
                    stage_problem = _retry_stage_problem(
                        stage_problem,
                        attempt_index,
                        last_error,
                    )
                    continue

                last_result = result
                if result.converged:
                    attempt_logs.append(
                        AttemptLog(stage=stage_label, attempt=attempt_index, status="ok")
                    )
                else:
                    last_error = "Solver did not converge"
                    attempt_logs.append(
                        AttemptLog(
                            stage=stage_label,
                            attempt=attempt_index,
                            status="fail",
                            message=last_error,
                        )
                    )
                solution = Solution(ok=bool(result.converged), result=result, attempts=attempt_logs)
                solution.info.update(
                    {
                        "stage": stage_label,
                        "stage_index": stage_index,
                        "mission_name": mission.name,
                        "initial_epoch": mission.initial_epoch,
                    }
                )
                if result.converged or not self.solve_config.raise_on_fail:
                    return solution
                if not self._should_retry(attempt_index):
                    break
                stage_problem = _retry_stage_problem(stage_problem, attempt_index, last_error)

        solution = Solution(
            ok=False,
            result=last_result,
            attempts=attempt_logs,
            last_error=last_error,
        )
        if self.solve_config.raise_on_fail:
            raise RuntimeError(solution.summary())
        return solution

    def _validate_mission(self, mission: Mission) -> None:
        """Validate the mission object before backend selection.

        Parameters
        ----------
        mission
            Candidate mission object.

        Raises
        ------
        TypeError
            If ``mission`` is not an Octavian ``Mission``.
        ValueError
            If mission-level validation fails.
        """
        from .mission import Mission

        if not isinstance(mission, Mission):
            raise TypeError("MissionRunner.solve expects a Mission.")
        mission.validate()

    def _max_attempts(self) -> int:
        """Return the configured maximum attempt count.

        Returns
        -------
        int
            At least one attempt.
        """
        configured_attempts = max(1, int(self.solve_config.max_attempts))
        if not self.retry.enabled:
            return configured_attempts
        retry_limited_attempts = max(1, 1 + int(self.retry.max_retries))
        return min(configured_attempts, retry_limited_attempts)

    def _should_retry(self, attempt_index: int) -> bool:
        """Return whether another attempt should be made.

        Parameters
        ----------
        attempt_index
            One-based attempt index that just finished.

        Returns
        -------
        bool
            Whether retry policy and solve config allow another attempt.
        """
        return bool(self.retry.enabled) and attempt_index < self._max_attempts()

@dataclass(slots=True)
class AttemptLog:
    """One solver attempt recorded by `MissionRunner`.

    Attributes:
        stage: Runner stage label, or ``"default"`` when no staged plan is used.
        attempt: One-based attempt number within the stage.
        status: Short status string such as ``"ok"`` or ``"fail"``.
        message: Optional failure or diagnostic message.
    """

    stage: str
    attempt: int
    status: str
    message: str = ""

@dataclass(slots=True)
class Solution:
    """Stable user-facing wrapper around a backend solve result.

    `Solution` keeps backend result objects from leaking directly into mission
    scripts. Successful solves expose the backend result through ``result`` and
    convenience accessors such as ``traj``. Failed solves still return structured
    attempt logs when ``SolveConfig.raise_on_fail`` is false.
    """

    ok: bool
    result: RendezvousResult | None = None
    attempts: list[AttemptLog] = field(default_factory=list)
    info: dict[str, Any] = field(default_factory=dict)
    last_error: str | None = None

    def summary(self) -> str:
        """Return a human-readable solve summary."""
        if self.result is not None:
            return self.result.summary()
        lines = ["Octavian solution: FAILED"]
        if self.last_error:
            lines.append(f"  last_error: {self.last_error}")
        if self.attempts:
            lines.append("  attempts:")
            for a in self.attempts:
                lines.append(
                    f"    - stage={a.stage} attempt={a.attempt}: {a.status} {a.message}".rstrip()
                )
        return "\n".join(lines)

    @property
    def traj(self) -> np.ndarray:
        """Return the solved trajectory array, or an empty array on failure."""
        if self.result is None:
            return np.empty((0, 0), dtype=float)
        return np.asarray(self.result.traj, dtype=float)

    @property
    def frame(self) -> CoordinateFrame | None:
        """Return the reference frame declared by the solver result."""
        if self.result is None:
            return None
        value = self.result.info.get("frame")
        if isinstance(value, CoordinateFrame):
            return value
        if isinstance(value, dict):
            return CoordinateFrame.from_dict(value)
        return None

    @property
    def scaling(self) -> SolverScaling | None:
        """Return characteristic units used by the solver, when available."""
        if self.result is None:
            return None
        value = self.result.info.get("scaling")
        if isinstance(value, SolverScaling):
            return value
        if isinstance(value, dict):
            return SolverScaling(**value)
        return None

    @property
    def chief_trajectory_eci(self) -> np.ndarray:
        """Return the propagated chief absolute history for relative solves."""
        if self.result is None:
            return np.empty((0, 7), dtype=float)
        value = self.result.info.get("chief_trajectory_eci")
        if value is None:
            return np.empty((0, 7), dtype=float)
        return np.asarray(value, dtype=float)

    @property
    def deputy_trajectory_eci(self) -> np.ndarray:
        """Return the propagated deputy absolute history for relative solves."""
        if self.result is None:
            return np.empty((0, 7), dtype=float)
        value = self.result.info.get("deputy_trajectory_eci")
        if value is None:
            return np.empty((0, 7), dtype=float)
        return np.asarray(value, dtype=float)

    @property
    def native_relative_trajectory(self) -> np.ndarray:
        """Return the solver's native relative state and time history.

        For a D'Amico phase, for example, columns are
        ``[δa, δλ, δex, δey, δix, δiy, t]``. The ordinary :attr:`traj`
        accessor always remains the reconstructed ``[RIC state, t]`` view.
        """
        if self.result is None:
            return np.empty((0, 7), dtype=float)
        value = self.result.info.get("native_relative_trajectory")
        if value is None:
            return np.empty((0, 7), dtype=float)
        return np.asarray(value, dtype=float)

    @property
    def native_relative_phase_trajectories(self) -> tuple[np.ndarray, ...]:
        """Return one control-free native state/time array per relative phase.

        A stitched native trajectory is unavailable when phase layouts change,
        for example when an initial coast does not carry mass but a later
        finite-burn chain does. This accessor always preserves each phase's
        declared native representation.
        """
        if self.result is None:
            return ()
        value = self.result.info.get("native_relative_phase_trajectories")
        if value is None:
            return ()
        return tuple(np.asarray(trajectory, dtype=float) for trajectory in value)

    @property
    def relative_propagation_mode(self) -> str | None:
        """Return the selected relative formulation, or ``None`` if inertial."""
        if self.result is None:
            return None
        value = self.result.info.get("relative_propagation_mode")
        return None if value is None else str(value)

    @property
    def cr3bp_system(self) -> CR3BPSystem | None:
        """Return the solved CR3BP system, or ``None`` for other dynamics."""
        if self.result is None:
            return None
        value = self.result.info.get("cr3bp_system")
        if not isinstance(value, dict):
            return None
        return CR3BPSystem(
            primary=str(value["primary"]),
            secondary=str(value["secondary"]),
            separation_m=float(value["separation_m"]),
            name=str(value["name"]),
        )

    @property
    def cr3bp_dimensional(self) -> bool | None:
        """Return the solved CR3BP unit mode, or ``None`` for other dynamics."""
        if self.result is None:
            return None
        value = self.result.info.get("cr3bp_system")
        if not isinstance(value, dict):
            return None
        return bool(value.get("dimensional", True))

    @property
    def phase_control_trajectories(self) -> tuple[np.ndarray, ...]:
        """Return ``[time, controls...]`` arrays for every compiled phase.

        Vector-throttle components and scalar throttle are dimensionless.
        Euler-rate columns are converted from normalized optimizer variables
        to physical radians per second.
        """
        if self.result is None:
            return ()
        value = self.result.info.get("phase_control_trajectories")
        if value is None:
            return ()
        return tuple(np.asarray(trajectory, dtype=float) for trajectory in value)

    @property
    def attitude_phase_trajectories(self) -> tuple[np.ndarray, ...]:
        """Return Euler attitude histories for kinematic-attitude phases.

        Each array contains ``[yaw, pitch, roll, time, yaw_rate, pitch_rate,
        roll_rate]`` in radians, seconds, and radians per second.
        """
        if self.result is None:
            return ()
        value = self.result.info.get("attitude_phase_trajectories")
        if value is None:
            return ()
        return tuple(np.asarray(trajectory, dtype=float) for trajectory in value)

    def to_ephemeris(
        self,
        *,
        trajectory: str = "auto",
        epoch: str | datetime | float | int | None = None,
        frame_name: str | None = None,
        center_name: str | None = None,
        object_name: str | None = None,
        object_id: int = -100_000,
        center_id: int | None = None,
    ) -> Ephemeris:
        """Return a validated ephemeris selected from this solution.

        Args:
            trajectory: ``"auto"`` (the deputy for relative solutions,
                otherwise the solved trajectory), ``"solved"``, ``"chief"``,
                or ``"deputy"``. Chief and deputy histories are absolute ECI
                trajectories reconstructed by exact relative formulations.
            epoch: Epoch corresponding to trajectory time zero. When omitted,
                use the originating mission's ``initial_epoch``.
            frame_name: Frame label for the state values. This does not rotate
                coordinates. Absolute ECI histories default to ``"J2000"``.
            center_name: Center-of-motion label, normally inferred from the
                solution metadata.
            object_name: Exported object label.
            object_id: NAIF object ID used by BSP output.
            center_id: NAIF center ID used by BSP output, inferred for Earth,
                Moon, and Sun when omitted.

        Returns:
            An :class:`octavian.exports.Ephemeris` ready to write.

        Raises:
            RuntimeError: If the solution has no result.
            ValueError: If the selected trajectory or epoch is unavailable.

        Note:
            File export labels the selected data but does not perform a frame
            rotation. Select ``"chief"`` or ``"deputy"`` when exporting a
            relative solution to an inertial OEM or BSP.
        """
        if self.result is None:
            raise RuntimeError("No result to export")

        selected = str(trajectory).strip().lower()
        if selected == "auto":
            selected = (
                "deputy"
                if self.frame is not None
                and self.frame.kind == "relative"
                and self.deputy_trajectory_eci.size
                else "solved"
            )
        if selected not in {"solved", "chief", "deputy"}:
            raise ValueError("trajectory must be one of: auto, solved, chief, deputy")
        if selected == "solved" and self.cr3bp_dimensional is False:
            raise ValueError(
                "Canonical CR3BP trajectories cannot be exported as SI ephemerides. "
                "Dimensionalize the state and time history first."
            )

        if selected == "chief":
            rows = self.chief_trajectory_eci
        elif selected == "deputy":
            rows = self.deputy_trajectory_eci
        else:
            rows = self.traj
        if rows.ndim != 2 or rows.shape[1] != 7 or len(rows) < 2:
            raise ValueError(f"The solution does not contain an exportable {selected} trajectory")

        export_epoch = epoch
        if export_epoch is None:
            export_epoch = self.info.get("initial_epoch")
        if export_epoch is None:
            export_epoch = self.result.info.get("initial_epoch")
        if export_epoch is None:
            raise ValueError("Ephemeris export requires epoch= or Mission(initial_epoch=...).")

        absolute_history = selected in {"chief", "deputy"}
        solution_frame = self.frame
        inferred_frame = "J2000"
        if not absolute_history and solution_frame is not None:
            orientation = solution_frame.orientation.strip()
            inferred_frame = (
                "J2000"
                if orientation.upper() in {"ECI", "ICRF", "J2000", "EME2000"}
                else orientation
            )
        inferred_center = str(
            self.result.info.get("central_body")
            or (solution_frame.origin if solution_frame is not None else "earth")
        )
        if not absolute_history and solution_frame is not None:
            inferred_center = solution_frame.origin

        resolved_center = str(center_name or inferred_center).strip().upper()
        resolved_object = object_name or (
            "CHIEF" if selected == "chief" else "DEPUTY" if selected == "deputy" else "SPACECRAFT"
        )
        resolved_center_id = (
            int(center_id) if center_id is not None else _common_naif_center_id(resolved_center)
        )
        return Ephemeris.from_trajectory(
            rows,
            epoch=export_epoch,
            frame_name=frame_name or inferred_frame,
            center_name=resolved_center,
            object_name=resolved_object,
            object_id=object_id,
            center_id=resolved_center_id,
        )

    def export_ephemeris(
        self,
        path: str | Path,
        *,
        trajectory: str = "auto",
        epoch: str | datetime | float | int | None = None,
        frame_name: str | None = None,
        center_name: str | None = None,
        object_name: str | None = None,
        object_id: int = -100_000,
        center_id: int | None = None,
        format: EphemerisFormat | None = None,
        overwrite: bool = False,
        interpolation_degree: int = 7,
    ) -> Path:
        """Export a solved, chief, or deputy trajectory to an ephemeris file.

        The output extension selects STK ``.e``, CCSDS ``.oem``, SPICE
        ``.bsp``/``.spk``, or SI-unit ``.csv``. See :meth:`to_ephemeris` for
        trajectory and frame semantics.

        Returns:
            The resolved path written by the selected exporter.
        """
        ephemeris = self.to_ephemeris(
            trajectory=trajectory,
            epoch=epoch,
            frame_name=frame_name,
            center_name=center_name,
            object_name=object_name,
            object_id=object_id,
            center_id=center_id,
        )
        return ephemeris.write(
            path,
            format=format,
            overwrite=overwrite,
            interpolation_degree=interpolation_degree,
        )

    def viz(self):
        """Namespace-style access to visualization helpers."""
        from .viz import matplotlib as _matplotlib
        from .viz import plotly as _plotly

        self_outer = self

        class _Viz:
            def save_html(self, out_html: str, *, title: str = "trajectory") -> None:
                """Save a frame-aware trajectory plot.

                Relative results use RIC axes and a chief marker, rotating
                CR3BP results show both primaries and Lagrange points, and
                inertial results retain the Earth-centered trajectory view.
                """
                if self_outer.result is None:
                    raise RuntimeError("No result to visualize")
                frame = self_outer.frame
                phase_segments = self_outer.result.info.get("phase_segments")
                if frame is not None and frame.kind == "relative":
                    _plotly.save_relative_trajectory_html(
                        self_outer.result.traj,
                        out_html,
                        maneuvers=self_outer.result.maneuvers,
                        phase_segments=phase_segments,
                        title=title,
                    )
                    return
                if frame is not None and frame.kind == "rotating":
                    system = self_outer.cr3bp_system
                    if system is None:
                        raise RuntimeError(
                            "Rotating-frame visualization requires CR3BP system metadata"
                        )
                    _plotly.save_cr3bp_trajectory_html(
                        self_outer.result.traj,
                        out_html,
                        system=system,
                        dimensional=self_outer.cr3bp_dimensional is not False,
                        title=title,
                    )
                    return
                _plotly.save_trajectory_html(
                    self_outer.result.traj,
                    out_html,
                    maneuvers=self_outer.result.maneuvers,
                    phase_segments=phase_segments,
                    title=title,
                )

            def figure(
                self,
                *,
                title: str = "trajectory",
                projection: str = "3d",
            ):
                """Return a frame-aware Matplotlib trajectory figure.

                Use ``projection="xy"``, ``"xz"``, or ``"yz"`` for a 2D
                view; the default ``"3d"`` retains all position axes.
                """
                if self_outer.result is None:
                    raise RuntimeError("No result to visualize")
                frame = self_outer.frame
                phase_segments = self_outer.result.info.get("phase_segments")
                if frame is not None and frame.kind == "relative":
                    return _matplotlib.relative_trajectory_figure(
                        self_outer.result.traj,
                        maneuvers=self_outer.result.maneuvers,
                        phase_segments=phase_segments,
                        title=title,
                        projection=projection,
                    )
                if frame is not None and frame.kind == "rotating":
                    system = self_outer.cr3bp_system
                    if system is None:
                        raise RuntimeError(
                            "Rotating-frame visualization requires CR3BP system metadata"
                        )
                    return _matplotlib.cr3bp_trajectory_figure(
                        self_outer.result.traj,
                        system=system,
                        dimensional=self_outer.cr3bp_dimensional is not False,
                        title=title,
                        projection=projection,
                    )
                return _matplotlib.trajectory_figure(
                    self_outer.result.traj,
                    maneuvers=self_outer.result.maneuvers,
                    phase_segments=phase_segments,
                    title=title,
                    projection=projection,
                )

            def save_image(
                self,
                out_image: str | Path,
                *,
                title: str = "trajectory",
                projection: str = "3d",
                dpi: int = 160,
            ) -> None:
                """Save a frame-aware Matplotlib trajectory as PNG or JPEG."""
                _matplotlib.save_figure_image(
                    self.figure(title=title, projection=projection),
                    out_image,
                    dpi=dpi,
                )

            def show(
                self,
                *,
                title: str = "trajectory",
                projection: str = "3d",
            ):
                """Open a frame-aware trajectory using the Matplotlib GUI backend."""
                return _matplotlib.show_figure(self.figure(title=title, projection=projection))

            def save_relative_html(
                self,
                out_html: str,
                *,
                title: str = "relative trajectory",
                chief_radius_m: float = 0.0,
            ) -> None:
                """Save the result explicitly as a chief-centered RIC plot."""
                if self_outer.result is None:
                    raise RuntimeError("No result to visualize")
                _plotly.save_relative_trajectory_html(
                    self_outer.result.traj,
                    out_html,
                    maneuvers=self_outer.result.maneuvers,
                    phase_segments=self_outer.result.info.get("phase_segments"),
                    title=title,
                    chief_radius_m=chief_radius_m,
                )

            def save_diagnostics_html(
                self,
                out_html: str,
                *,
                title: str = "trajectory diagnostics",
            ) -> None:
                """Save frame-aware state and geometry values over time."""
                if self_outer.result is None:
                    raise RuntimeError("No result to visualize")
                frame = self_outer.frame
                _plotly.save_trajectory_diagnostics_html(
                    self_outer.result.traj,
                    out_html,
                    frame_kind=(frame.kind if frame is not None else "inertial"),
                    mu_m3ps2=self_outer.result.info.get("mu_m3ps2"),
                    solar_directions_ric=self_outer.result.info.get("solar_directions_ric"),
                    cr3bp_system=self_outer.cr3bp_system,
                    cr3bp_dimensional=self_outer.cr3bp_dimensional is not False,
                    title=title,
                )

            def diagnostics_figure(
                self,
                *,
                title: str = "trajectory diagnostics",
            ):
                """Return frame-aware Matplotlib state and geometry panels."""
                if self_outer.result is None:
                    raise RuntimeError("No result to visualize")
                frame = self_outer.frame
                return _matplotlib.trajectory_diagnostics_figure(
                    self_outer.result.traj,
                    frame_kind=(frame.kind if frame is not None else "inertial"),
                    mu_m3ps2=self_outer.result.info.get("mu_m3ps2"),
                    solar_directions_ric=self_outer.result.info.get("solar_directions_ric"),
                    cr3bp_system=self_outer.cr3bp_system,
                    cr3bp_dimensional=self_outer.cr3bp_dimensional is not False,
                    title=title,
                )

            def save_diagnostics_image(
                self,
                out_image: str | Path,
                *,
                title: str = "trajectory diagnostics",
                dpi: int = 160,
            ) -> None:
                """Save frame-aware Matplotlib diagnostics as PNG or JPEG."""
                _matplotlib.save_figure_image(
                    self.diagnostics_figure(title=title),
                    out_image,
                    dpi=dpi,
                )

            def show_diagnostics(
                self,
                *,
                title: str = "trajectory diagnostics",
            ):
                """Open frame-aware diagnostics using the Matplotlib GUI backend."""
                return _matplotlib.show_figure(self.diagnostics_figure(title=title))

        return _Viz()

class MissionConfigError(ValueError):
    """A mission configuration file is malformed or internally inconsistent."""