API reference

The public surface is the ten names importable from orlab, plus the exception types in orlab.errors, the jar utilities in orlab.jars, the parallel-running machinery in orlab.parallel, and the dispersion listeners in orlab.listeners. Everything else is internal.

OpenRocketInstance

Use with the 'with' construct: entering starts the JVM and OpenRocket on first use. The JVM cannot be restarted in a process (JPype), so it stays up after the block and ends with the interpreter — a later with OpenRocketInstance(...) on the same jar reuses it (sequential blocks and notebook re-runs work); a different jar raises OrlabError. OpenRocket's log level is process-global: the most recently entered instance's log_level wins.

Source code in src/orlab/core/openrocket_instance.py

class OpenRocketInstance:
    """Use with the 'with' construct: entering starts the JVM and OpenRocket
    on first use. The JVM cannot be restarted in a process (JPype), so it
    stays up after the block and ends with the interpreter — a later
    ``with OpenRocketInstance(...)`` on the same jar reuses it (sequential
    blocks and notebook re-runs work); a different jar raises OrlabError.
    OpenRocket's log level is process-global: the most recently entered
    instance's log_level wins.
    """

    # Deprecated: pass jvm_path to __init__ instead. Honored (with a warning)
    # for one release.
    MANUAL_JVM_PATH = None

    def __init__(
        self,
        jar_path: str | None = None,
        log_level: OrLogLevel | str = OrLogLevel.ERROR,
        *,
        jvm_path: "str | os.PathLike[str] | None" = None,
        jvm_args: Sequence[str] = (),
    ):
        """jar_path is the full path of the OpenRocket .jar file to use;
        defaults to $ORLAB_JAR, then $CLASSPATH, then the newest supported
        OpenRocket-*.jar in the current directory, then the newest jar in
        the orlab.fetch_jar cache (verified, never downloading).
        log_level can be either OFF, ERROR, WARN, INFO, DEBUG, TRACE and ALL.
        jvm_path selects the JVM library explicitly (default: JPype's
        auto-detection via JAVA_HOME). jvm_args are appended to the JVM
        launch arguments, e.g. ("-Xmx4g",) for large monte-carlo runs; both
        only take effect for the instance that actually starts the JVM.
        """
        self.openrocket: Any = None  # JPackage core root once started
        self.started = False

        jar_path = jar_path or _default_jar_path()
        if not os.path.exists(jar_path):
            raise FileNotFoundError(
                f"Jar file {os.path.abspath(jar_path)} does not exist "
                "(pass jar_path, set ORLAB_JAR, or fetch one: "
                "`python -m orlab fetch`)"
            )
        self.jar_path = jar_path
        try:
            self.or_version = read_or_version(jar_path)
            # UnsupportedOpenRocketVersion (too-old jar) passes through untouched
            self.profile, exact = get_profile(self.or_version)
        except (zipfile.BadZipFile, KeyError, ValueError) as e:
            # covers a corrupt zip, missing build.properties/build.version,
            # and an unparseable version string
            raise NotAnOpenRocketJar(
                f"{os.path.abspath(jar_path)} is not an OpenRocket jar ({e})"
            ) from e
        if not exact:
            logger.warning(
                "No profile for OpenRocket %s; falling back to the nearest older "
                "profile (%s). Newer constants may be missing from orlab's enums.",
                self.or_version,
                self.profile.version_string,
            )

        if isinstance(jvm_args, str):
            raise TypeError("jvm_args must be a sequence of strings, not a string")
        # jpype's startJVM rejects PathLike (it reads a stray path as a VM
        # argument) — coerce here so find_installed's Path works directly
        self.jvm_path = os.fspath(jvm_path) if jvm_path is not None else None
        self.jvm_args = tuple(jvm_args)
        self._started_jvm = False

        if isinstance(log_level, str):
            self.or_log_level = OrLogLevel[log_level]
        else:
            self.or_log_level = log_level

    def __enter__(self):
        global _active_core_root, _active_jar_path

        if jpype.isJVMStarted():
            requested = os.path.abspath(self.jar_path)
            if _active_jar_path is None:
                raise OrlabError(
                    "This process's JVM is running but OpenRocket startup never "
                    "completed (an earlier attempt failed, or the JVM was started "
                    "outside orlab). JPype cannot restart a JVM — retry in a new "
                    "process."
                )
            if requested != _active_jar_path:
                raise OrlabError(
                    f"A JVM is already running with {_active_jar_path}, and JPype "
                    "cannot restart a JVM — one process can use only one OpenRocket "
                    f"jar. Use a new process for {requested}."
                )
            # Same jar: reuse the running OpenRocket.
            if not self._started_jvm and (self.jvm_path is not None or self.jvm_args):
                logger.warning(
                    "This instance did not start the JVM; its jvm_path/jvm_args "
                    "have no effect (the JVM is per-process and already running)"
                )
            self.openrocket = _active_core_root
            self.openrocket_swing = _jpackage(self.profile.swing_root)
            self._set_or_log_level()
            self.started = True
            return self

        jvm_path = self.jvm_path
        if jvm_path is None and self.MANUAL_JVM_PATH is not None:
            warnings.warn(
                "MANUAL_JVM_PATH is deprecated; pass jvm_path to OpenRocketInstance",
                DeprecationWarning,
                stacklevel=2,
            )
            jvm_path = os.fspath(self.MANUAL_JVM_PATH)
        jvm_path = jvm_path or jpype.getDefaultJVMPath()

        logger.info(
            f"Starting JVM from {jvm_path} CLASSPATH={self.jar_path} (OpenRocket {self.or_version})"
        )

        # --add-opens: OpenRocket <= 15.03 bundles a Guice that reflects into
        # java.lang, which the module system blocks on modern JVMs. Harmless
        # on newer OpenRocket versions.
        jvm_args = [
            "-ea",
            "--add-opens=java.base/java.lang=ALL-UNNAMED",
            f"-Djava.class.path={self.jar_path}",
        ]
        if self.profile.startup == "core":
            jvm_args.append("-Djava.awt.headless=true")
        jvm_args.extend(self.jvm_args)
        jpype.startJVM(jvm_path, *jvm_args)

        try:
            self._start_openrocket()
        except Exception as e:
            for window in jpype.java.awt.Window.getWindows():
                window.dispose()
            raise OrlabError(
                "OpenRocket startup failed after the JVM launched; the JVM cannot "
                "be restarted, so retry in a new process once the cause is fixed."
            ) from e

        _active_core_root = self.openrocket
        _active_jar_path = os.path.abspath(self.jar_path)
        self._started_jvm = True
        self._warn_on_profile_drift()
        self.started = True

        return self

    def _start_openrocket(self):
        """Bootstraps OpenRocket inside the (fresh) JVM. The module globals are
        set by __enter__ only after this succeeds."""
        # ----- Java imports -----
        # Package roots come from the version profile (OpenRocket 24.12 renamed
        # net.sf.openrocket to info.openrocket.core + info.openrocket.swing).
        self.openrocket = _jpackage(self.profile.core_root)
        self.openrocket_swing = _jpackage(self.profile.swing_root)
        # -----

        if self.profile.startup == "core":
            # Official headless bootstrap (24.12+). PluginModule must be passed
            # explicitly: the Java no-arg initialize() adds it internally, but
            # JPype dispatches a zero-arg call to the varargs overload with an
            # empty module array, and startup then fails on unbound plugins.
            self.openrocket.startup.OpenRocketCore.initialize(self.openrocket.plugin.PluginModule())
        else:
            # Minimally viable translation of openrocket.startup.SwingStartup
            guice = jpype.JPackage("com").google.inject.Guice
            gui_module = self.openrocket_swing.startup.GuiModule()
            plugin_module = self.openrocket.plugin.PluginModule()

            injector = guice.createInjector(gui_module, plugin_module)

            app = self.openrocket.startup.Application
            app.setInjector(injector)

            gui_module.startLoader()

            # Ensure that loaders are done loading before continuing
            # Without this there seems to be a race condition bug that leads to the whole thing freezing
            preset_loader = _get_private_field(gui_module, "presetLoader")
            preset_loader.blockUntilLoaded()
            motor_loader = _get_private_field(gui_module, "motorLoader")
            motor_loader.blockUntilLoaded()

        self._set_or_log_level()

    def _set_or_log_level(self):
        LoggerFactory = jpype.JPackage("org").slf4j.LoggerFactory
        Logger = jpype.JPackage("ch").qos.logback.classic.Logger
        or_logger = LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME)
        or_logger.setLevel(self._translate_log_level())

    def __exit__(self, ex, value, tb):
        # Dispose any open windows (usually just a loading screen); the JVM
        # itself stays up — JPype cannot restart one, so shutting it down here
        # would break every later OpenRocketInstance in this process. It ends
        # with the interpreter. Existing Helpers and listeners stay usable.
        if jpype.isJVMStarted():
            for window in jpype.java.awt.Window.getWindows():
                window.dispose()

        self.started = False
        logger.info("OpenRocketInstance closed (JVM stays up for reuse)")

    def _warn_on_profile_drift(self):
        """Compares the live jar's constants against the profile (drift alarm).
        Pure diagnostics: must never abort startup, whatever the jar looks like.
        """
        try:
            live_types, live_events = reflect_live_constants(self.openrocket)
        except Exception as e:
            logger.warning("Profile drift check failed on OpenRocket %s: %s", self.or_version, e)
            return
        for kind, live, known in (
            ("FlightDataType", live_types, self.profile.flight_data_types),
            ("FlightEvent", live_events, self.profile.flight_events),
        ):
            extra = live - known
            if extra:
                logger.warning(
                    "OpenRocket %s exposes %s constants not in the %s profile: %s "
                    "(regenerate with tools/generate_profile.py)",
                    self.or_version,
                    kind,
                    self.profile.version_string,
                    ", ".join(sorted(extra)),
                )
            gone = known - live
            if gone:
                logger.warning(
                    "Profile %s lists %s constants the loaded jar lacks: %s",
                    self.profile.version_string,
                    kind,
                    ", ".join(sorted(gone)),
                )

    def _translate_log_level(self):
        # ----- Java imports -----
        Level = jpype.JPackage("ch").qos.logback.classic.Level
        # -----

        return getattr(Level, self.or_log_level.name)

__init__(jar_path=None, log_level=OrLogLevel.ERROR, *, jvm_path=None, jvm_args=())

jar_path is the full path of the OpenRocket .jar file to use; defaults to $ORLAB_JAR, then $CLASSPATH, then the newest supported OpenRocket-*.jar in the current directory, then the newest jar in the orlab.fetch_jar cache (verified, never downloading). log_level can be either OFF, ERROR, WARN, INFO, DEBUG, TRACE and ALL. jvm_path selects the JVM library explicitly (default: JPype's auto-detection via JAVA_HOME). jvm_args are appended to the JVM launch arguments, e.g. ("-Xmx4g",) for large monte-carlo runs; both only take effect for the instance that actually starts the JVM.

Source code in src/orlab/core/openrocket_instance.py

def __init__(
    self,
    jar_path: str | None = None,
    log_level: OrLogLevel | str = OrLogLevel.ERROR,
    *,
    jvm_path: "str | os.PathLike[str] | None" = None,
    jvm_args: Sequence[str] = (),
):
    """jar_path is the full path of the OpenRocket .jar file to use;
    defaults to $ORLAB_JAR, then $CLASSPATH, then the newest supported
    OpenRocket-*.jar in the current directory, then the newest jar in
    the orlab.fetch_jar cache (verified, never downloading).
    log_level can be either OFF, ERROR, WARN, INFO, DEBUG, TRACE and ALL.
    jvm_path selects the JVM library explicitly (default: JPype's
    auto-detection via JAVA_HOME). jvm_args are appended to the JVM
    launch arguments, e.g. ("-Xmx4g",) for large monte-carlo runs; both
    only take effect for the instance that actually starts the JVM.
    """
    self.openrocket: Any = None  # JPackage core root once started
    self.started = False

    jar_path = jar_path or _default_jar_path()
    if not os.path.exists(jar_path):
        raise FileNotFoundError(
            f"Jar file {os.path.abspath(jar_path)} does not exist "
            "(pass jar_path, set ORLAB_JAR, or fetch one: "
            "`python -m orlab fetch`)"
        )
    self.jar_path = jar_path
    try:
        self.or_version = read_or_version(jar_path)
        # UnsupportedOpenRocketVersion (too-old jar) passes through untouched
        self.profile, exact = get_profile(self.or_version)
    except (zipfile.BadZipFile, KeyError, ValueError) as e:
        # covers a corrupt zip, missing build.properties/build.version,
        # and an unparseable version string
        raise NotAnOpenRocketJar(
            f"{os.path.abspath(jar_path)} is not an OpenRocket jar ({e})"
        ) from e
    if not exact:
        logger.warning(
            "No profile for OpenRocket %s; falling back to the nearest older "
            "profile (%s). Newer constants may be missing from orlab's enums.",
            self.or_version,
            self.profile.version_string,
        )

    if isinstance(jvm_args, str):
        raise TypeError("jvm_args must be a sequence of strings, not a string")
    # jpype's startJVM rejects PathLike (it reads a stray path as a VM
    # argument) — coerce here so find_installed's Path works directly
    self.jvm_path = os.fspath(jvm_path) if jvm_path is not None else None
    self.jvm_args = tuple(jvm_args)
    self._started_jvm = False

    if isinstance(log_level, str):
        self.or_log_level = OrLogLevel[log_level]
    else:
        self.or_log_level = log_level

Helper

This class contains a variety of useful helper functions and wrapper for using openrocket via jpype. These are intended to take care of some of the more cumbersome aspects of calling methods, or provide more 'pythonic' data structures for general use.

Source code in src/orlab/core/helper.py

class Helper:
    """This class contains a variety of useful helper functions and wrapper for using
    openrocket via jpype. These are intended to take care of some of the more
    cumbersome aspects of calling methods, or provide more 'pythonic' data structures
    for general use.
    """

    def __init__(self, open_rocket_instance: OpenRocketInstance):
        if not open_rocket_instance.started:
            raise OrlabError(
                "OpenRocketInstance not started — enter it first "
                "('with OpenRocketInstance(...) as instance:')"
            )

        self._instance = open_rocket_instance
        self.openrocket = open_rocket_instance.openrocket

    def load_doc(self, or_filename):
        """Loads a .ork file and returns the corresponding openrocket document"""

        or_java_file = jpype.java.io.File(or_filename)
        loader = self.openrocket.file.GeneralRocketLoader(or_java_file)
        doc = loader.load()
        return doc

    def save_doc(self, or_filename, doc):
        """Saves an openrocket document to a .ork file"""

        or_java_file = jpype.java.io.File(or_filename)
        saver = self.openrocket.file.GeneralRocketSaver()
        saver.save(or_java_file, doc)

    def run_simulation(
        self,
        sim,
        listeners: list[AbstractSimulationListener] | None = None,
        *,
        randomize_seed: bool = True,
    ):
        """This is a wrapper to the Simulation.simulate() for running a simulation
        The optional listeners parameter is a sequence of objects which extend orl.AbstractSimulationListener.

        By default the simulation's random seed is randomized before each run
        (identical repeated runs would otherwise produce identical numbers —
        the behavior naive monte-carlo loops rely on). Pass
        randomize_seed=False to respect the seed already set on the options.
        Note: on OpenRocket 24.12 the wind model draws additional per-process
        entropy, so a fixed seed reproduces results within one process but not
        across processes when wind is enabled.
        """

        if listeners is None:
            # this method takes in a vararg of SimulationListeners, which is just a fancy way of passing in an array, so
            # we have to pass in an array of length 0 ..
            listener_array = jpype.JArray(
                self.openrocket.simulation.listeners.AbstractSimulationListener, 1
            )(0)
        else:
            listener_array = [
                jpype.JProxy(
                    (
                        self.openrocket.simulation.listeners.SimulationListener,
                        self.openrocket.simulation.listeners.SimulationEventListener,
                        self.openrocket.simulation.listeners.SimulationComputationListener,
                        jpype.java.lang.Cloneable,
                    ),
                    inst=c,
                )
                for c in listeners
            ]

        if randomize_seed:
            sim.getOptions().randomizeSeed()
        sim.simulate(listener_array)

    def translate_flight_data_type(self, flight_data_type: FlightDataType | str):
        """Translates a FlightDataType (or constant name) to the Java constant.
        Raises UnsupportedFlightDataType when the loaded OpenRocket version
        does not expose it (constants differ across versions).
        """
        if isinstance(flight_data_type, FlightDataType):
            name = flight_data_type.name
        elif isinstance(flight_data_type, str):
            name = flight_data_type
        else:
            raise TypeError("Invalid type for flight_data_type")

        java_type = getattr(self.openrocket.simulation.FlightDataType, name, None)
        if java_type is None:
            loaded = self._instance.or_version
            available = versions_with(name)
            detail = f"available in: {', '.join(available)}" if available else "unknown constant"
            raise UnsupportedFlightDataType(
                f"{name} is not available in OpenRocket {loaded} ({detail})"
            )
        return java_type

    def get_timeseries(
        self, simulation, variables: Iterable[FlightDataType | str], branch_number=0
    ) -> dict[FlightDataType | str, np.ndarray]:
        """
        Gets a dictionary of timeseries data (as numpy arrays) from a simulation given specific variable names.

        :param simulation: An openrocket simulation object.
        :param variables: A sequence of FlightDataType or strings representing the desired variables
        :param branch_number: Stage branch to read (0 = sustainer).
        :return: Dict keyed by the requested variables; values are numpy arrays.
        """

        branch = simulation.getSimulatedData().getBranch(branch_number)
        output: dict[FlightDataType | str, np.ndarray] = {}
        for v in variables:
            output[v] = np.array(branch.get(self.translate_flight_data_type(v)))

        return output

    def get_final_values(
        self, simulation, variables: Iterable[FlightDataType | str], branch_number=0
    ) -> dict[FlightDataType | str, float]:
        """
        Gets a the final value in the time series from a simulation given variable names.

        :param simulation: An openrocket simulation object.
        :param variables: A sequence of FlightDataType or strings representing the desired variables
        :param branch_number: Stage branch to read (0 = sustainer).
        :return: Dict keyed by the requested variables; values are the final samples.
        """

        branch = simulation.getSimulatedData().getBranch(branch_number)
        output: dict[FlightDataType | str, float] = {}
        for v in variables:
            output[v] = branch.get(self.translate_flight_data_type(v))[-1]

        return output

    def translate_flight_event(self, flight_event) -> FlightEvent:
        """Translates a Java FlightEvent.Type constant to the FlightEvent enum.
        Raises ValueError for event types this orlab version does not know
        (newer OpenRocket releases add types; get_events skips them instead).
        """
        name = str(flight_event.name())
        try:
            return FlightEvent[name]
        except KeyError:
            raise ValueError(
                f"Unknown flight event type {name!r} from the loaded OpenRocket version"
            ) from None

    def get_events(self, simulation, branch_number: int = 0) -> dict[FlightEvent, list[float]]:
        """Returns a dictionary of all the flight events in a given simulation.
        Key is FlightEvent and value is a list of all the times at which the event occurs.
        Event types not known to this orlab version are skipped with a warning.

        :param branch_number: Stage branch to read (0 = sustainer).
        """
        branch = simulation.getSimulatedData().getBranch(branch_number)

        output: dict[FlightEvent, list[float]] = {}
        unknown: set[str] = set()
        for name, times in self._events_by_name(branch).items():
            try:
                event = FlightEvent[name]
            except KeyError:
                if name not in unknown:
                    unknown.add(name)
                    logger.warning("Skipping unknown flight event type %s", name)
                continue
            output[event] = times

        return output

    @staticmethod
    def _events_by_name(branch) -> dict[str, list[float]]:
        """One traversal of a branch's events, keyed by stable string names —
        shared by get_events (enum-gated view) and get_summary (which must
        see every event regardless of orlab's enum vintage)."""
        output: dict[str, list[float]] = {}
        for ev in branch.getEvents():
            output.setdefault(str(ev.getType().name()), []).append(float(ev.getTime()))
        return output

    def get_summary(self, simulation, branch_number: int = 0) -> FlightSummary:
        """Returns the scalar flight report for one branch as a
        :class:`~orlab.FlightSummary` — plain-Python values only, safe to
        pickle into JVM-less processes.

        Branch 0 uses OpenRocket's own FlightData summary getters verbatim;
        other branches derive the same numbers from their own time series.
        Missing/inapplicable values are NaN (see the field docs). Degenerate
        flights degrade to NaN; an unsimulated simulation raises OrlabError.

        :param branch_number: Stage branch to summarize (0 = sustainer).
        """
        branch_number = int(branch_number)  # builtin-types contract incl. numpy ints
        flight_data = simulation.getSimulatedData()
        if flight_data is None or int(flight_data.getBranchCount()) == 0:
            raise OrlabError("Simulation has no flight data; call run_simulation first")
        branch_count = int(flight_data.getBranchCount())
        if not 0 <= branch_number < branch_count:
            raise IndexError(f"branch {branch_number} out of range (simulation has {branch_count})")
        branch = flight_data.getBranch(branch_number)
        version = self._instance.or_version

        times = self._branch_series(branch, "TYPE_TIME")
        altitude = self._branch_series(branch, "TYPE_ALTITUDE")
        stability = self._branch_series(branch, "TYPE_STABILITY")
        velocity_z = self._branch_series(branch, "TYPE_VELOCITY_Z")
        velocity_total = self._branch_series(branch, "TYPE_VELOCITY_TOTAL")
        events = self._events_by_name(branch)

        def at_event(name: str, series: np.ndarray, which: int = 0) -> float:
            if name not in events:
                return math.nan
            return _value_at(times, series, events[name][which])

        # stability window: launch rod departure -> apogee. Boosters have no
        # LAUNCHROD (window starts at their first sample, off-rod values NaN);
        # a flight without an APOGEE event windows to its highest sample.
        t_rod = events["LAUNCHROD"][0] if "LAUNCHROD" in events else None
        if "APOGEE" in events:
            t_apogee = events["APOGEE"][0]
        elif len(altitude) and not np.isnan(altitude).all():
            t_apogee = float(times[int(np.nanargmax(altitude))])
        else:
            t_apogee = None
        window_start = t_rod if t_rod is not None else (float(times[0]) if len(times) else None)
        if window_start is not None and t_apogee is not None:
            min_stab, max_stab = _window_stats(times, stability, window_start, t_apogee)
        else:
            min_stab, max_stab = math.nan, math.nan

        t_deploy = (
            events["RECOVERY_DEVICE_DEPLOYMENT"][-1]
            if "RECOVERY_DEVICE_DEPLOYMENT" in events
            else None
        )
        t_ground = (
            events["GROUND_HIT"][0]
            if "GROUND_HIT" in events
            else (float(times[-1]) if len(times) else None)
        )
        if t_deploy is not None and t_ground is not None:
            descent_rate = _mean_descent_rate(times, velocity_z, t_deploy, t_ground)
        else:
            descent_rate = math.nan

        if branch_number == 0:

            def scalar(getter: str) -> float:
                fn = getattr(flight_data, getter, None)
                if fn is None:
                    _warn_absent_once(f"FlightData.{getter}", version)
                    return math.nan
                return float(fn())

            apogee = scalar("getMaxAltitude")
            time_to_apogee = scalar("getTimeToApogee")
            max_velocity = scalar("getMaxVelocity")
            max_acceleration = scalar("getMaxAcceleration")
            max_mach = scalar("getMaxMachNumber")
            velocity_off_rod = scalar("getLaunchRodVelocity")
            velocity_at_deployment = scalar("getDeploymentVelocity")
            ground_hit_velocity = scalar("getGroundHitVelocity")
            flight_time = scalar("getFlightTime")
            optimum_delay = scalar("getOptimumDelay")
        else:
            acceleration = self._branch_series(branch, "TYPE_ACCELERATION_TOTAL")
            mach = self._branch_series(branch, "TYPE_MACH_NUMBER")

            def series_max(series: np.ndarray) -> float:
                if len(series) == 0 or np.isnan(series).all():
                    return math.nan
                return float(np.nanmax(series))

            apogee = series_max(altitude)
            time_to_apogee = t_apogee if t_apogee is not None else math.nan
            max_velocity = series_max(velocity_total)
            max_acceleration = series_max(acceleration)
            max_mach = series_max(mach)
            velocity_off_rod = at_event("LAUNCHROD", velocity_total)
            velocity_at_deployment = at_event("RECOVERY_DEVICE_DEPLOYMENT", velocity_total, -1)
            ground_hit_velocity = at_event("GROUND_HIT", velocity_total)
            flight_time = float(times[-1]) if len(times) else math.nan
            # policy: reported for the sustainer only, where FlightData's own
            # figure is authoritative
            optimum_delay = math.nan

        pos_x = self._branch_series(branch, "TYPE_POSITION_X")
        pos_y = self._branch_series(branch, "TYPE_POSITION_Y")
        landing_x = float(pos_x[-1]) if len(pos_x) else math.nan
        landing_y = float(pos_y[-1]) if len(pos_y) else math.nan
        landing_distance = math.hypot(landing_x, landing_y)

        if hasattr(branch, "getName"):
            branch_name = str(branch.getName())
        elif hasattr(branch, "getBranchName"):
            branch_name = str(branch.getBranchName())
        else:
            branch_name = ""

        # bounded at apogee: the NaN-skip must not report a post-apogee
        # (tumble-regime) sample as the off-rod margin
        off_rod_stability = math.nan
        if t_rod is not None:
            off_rod_stability = _value_at(
                times, stability, t_rod, t_max=t_apogee if t_apogee is not None else math.inf
            )

        return FlightSummary(
            velocity_off_rod=velocity_off_rod,
            stability_off_rod_cal=off_rod_stability,
            apogee=apogee,
            time_to_apogee=time_to_apogee,
            max_velocity=max_velocity,
            max_acceleration=max_acceleration,
            max_mach=max_mach,
            min_stability_cal=min_stab,
            max_stability_cal=max_stab,
            velocity_at_deployment=velocity_at_deployment,
            descent_rate=descent_rate,
            ground_hit_velocity=ground_hit_velocity,
            landing_x=landing_x,
            landing_y=landing_y,
            landing_distance=landing_distance,
            landing_bearing_deg=_bearing_deg(landing_x, landing_y),
            flight_time=flight_time,
            optimum_delay=optimum_delay,
            branch_number=branch_number,
            branch_name=branch_name,
            branch_count=branch_count,
            warnings=tuple(str(w) for w in flight_data.getWarningSet()),
        )

    def _tabular_columns(self, branch, variables):
        """Resolves (label, java_type) columns for tabular export. Default:
        every profile data type populated on the branch, TYPE_TIME first,
        the rest in name order."""
        if variables is None:
            names = sorted(self._instance.profile.flight_data_types)
            if "TYPE_TIME" in names:
                names.remove("TYPE_TIME")
                names.insert(0, "TYPE_TIME")
            columns = []
            for name in names:
                try:
                    java_type = self.translate_flight_data_type(name)
                except UnsupportedFlightDataType:
                    # nearest-older fallback profile on a future jar that
                    # dropped a constant: skip it, never abort the export
                    _warn_absent_once(name, self._instance.or_version)
                    continue
                if branch.get(java_type) is not None:
                    columns.append((self._column_label(java_type, name), java_type))
            return columns
        columns = []
        for v in variables:
            java_type = self.translate_flight_data_type(v)
            name = v.name if isinstance(v, FlightDataType) else str(v)
            if branch.get(java_type) is None:
                raise ValueError(f"{name} is not populated on this branch — nothing to export")
            columns.append((self._column_label(java_type, name), java_type))
        return columns

    @staticmethod
    def _column_label(java_type, name: str) -> str:
        """ "NAME (unit)" from the jar's own SI unit string; units that are
        only whitespace/zero-width-space (OpenRocket's dimensionless marker)
        get no suffix."""
        unit = str(java_type.getUnitGroup().getSIUnit().getUnit())
        if not unit.replace("\u200b", "").strip():
            return name
        return f"{name} ({unit})"

    def get_dataframe(self, simulation, variables=None, branch_number: int = 0):
        """The branch's timeseries as a pandas DataFrame, one column per
        variable, labeled ``NAME (SI unit)``. Requires the ``orlab[pandas]``
        extra; everything else in orlab works without pandas.

        :param variables: FlightDataTypes or constant names; default: every
            profile data type populated on the branch.
        :param branch_number: Stage branch to read (0 = sustainer).
        """
        try:
            import pandas  # type: ignore[import-untyped]  # optional extra, no stubs
        except ImportError as e:
            raise ImportError(
                "pandas is required for get_dataframe — pip install orlab[pandas]"
            ) from e
        branch = simulation.getSimulatedData().getBranch(branch_number)
        columns = self._tabular_columns(branch, variables)
        return pandas.DataFrame(
            {label: np.asarray(branch.get(java_type), dtype=float) for label, java_type in columns}
        )

    def export_csv(self, simulation, path, variables=None, branch_number: int = 0) -> None:
        """Writes the branch's timeseries as UTF-8 CSV with ``NAME (SI
        unit)`` headers — stdlib only, no pandas needed. NaN samples become
        empty cells (``pandas.read_csv`` reads them back as NaN).

        :param variables: FlightDataTypes or constant names; default: every
            profile data type populated on the branch.
        :param branch_number: Stage branch to read (0 = sustainer).
        """
        import csv

        branch = simulation.getSimulatedData().getBranch(branch_number)
        columns = self._tabular_columns(branch, variables)
        series = [np.asarray(branch.get(java_type), dtype=float) for _, java_type in columns]
        # newline='' and explicit utf-8: text-mode newline translation would
        # write \r\r\n on Windows, and locale encodings choke on m/s²
        with open(path, "w", newline="", encoding="utf-8") as fh:
            writer = csv.writer(fh)
            writer.writerow([label for label, _ in columns])
            try:
                for row in zip(*series, strict=True):
                    writer.writerow(
                        ["" if math.isnan(value) else repr(float(value)) for value in row]
                    )
            except ValueError as e:  # pragma: no cover - can't-happen invariant
                raise ValueError(f"branch series lengths differ: {e}") from e

    def _branch_series(self, branch, type_name: str) -> np.ndarray:
        """A branch's series for a FlightDataType constant name as a float
        array; empty when the loaded version lacks the type (warned once per
        process) or the branch holds no such series."""
        try:
            java_type = self.translate_flight_data_type(type_name)
        except UnsupportedFlightDataType:
            _warn_absent_once(type_name, self._instance.or_version)
            return np.array([])
        values = branch.get(java_type)
        if values is None:
            return np.array([])
        return np.asarray(values, dtype=float)

    # --- motor selection and swapping ---

    def _motor_mount_interface(self):
        return jpype.JClass(f"{self._instance.profile.core_root}.rocketcomponent.MotorMount")

    def _sim_fcid(self, sim):
        """The simulation's OWN flight-configuration id — never the rocket's
        selected/default config, which the sim may not fly at all (verified:
        simple.ork's sim flies A8 while the selected config shows C6, and
        assigning to the selected config is a silent no-op)."""
        if hasattr(sim, "getFlightConfigurationId"):
            return sim.getFlightConfigurationId()
        if hasattr(sim.getOptions(), "getMotorConfigurationID"):  # 15.03
            return sim.getOptions().getMotorConfigurationID()
        raise OrlabError(
            "cannot determine this simulation's flight configuration id on "
            f"OpenRocket {self._instance.or_version} — the motor API drifted "
            "beyond both known forms"
        )

    def _motor_config(self, mount, fcid):
        """The mount's MotorConfiguration at an fcid. Two accessor eras
        (probed): getMotorConfig(fcid) on 22.02+, the
        getMotorConfiguration() map on 15.03. The returned object's own API
        (getMotor/setMotor/get/setEjectionDelay) is identical on all four
        versions."""
        if hasattr(mount, "getMotorConfig"):
            return mount.getMotorConfig(fcid)
        if hasattr(mount, "getMotorConfiguration"):
            return mount.getMotorConfiguration().get(fcid)
        raise OrlabError(
            "no known motor-configuration accessor on this mount "
            f"(OpenRocket {self._instance.or_version})"
        )

    def _resolve_mount(self, sim, mount_name):
        """The motor mount to operate on: by name, else the unique mount
        carrying a motor in the sim's config, else the unique MotorMount.
        Candidate enumeration filters via the MotorMount INTERFACE before
        calling isMotorMount() — most 15.03 components lack the method
        entirely while all 24.12 components have it (probed)."""
        interface = self._motor_mount_interface()
        rocket = sim.getRocket()
        if mount_name is not None:
            if isinstance(mount_name, str):
                component = self.get_component_named(rocket, mount_name)
            else:
                component = mount_name  # a mount object works directly
            if not isinstance(component, interface) or not component.isMotorMount():
                raise ValueError(f"{mount_name} is not an active motor mount")
            return component
        candidates = [c for c in JIterator(rocket) if isinstance(c, interface) and c.isMotorMount()]
        if not candidates:
            raise ValueError("this rocket has no active motor mount")
        if len(candidates) == 1:
            return candidates[0]
        fcid = self._sim_fcid(sim)
        bearing = []
        for candidate in candidates:
            config = self._motor_config(candidate, fcid)
            if config is not None and config.getMotor() is not None:
                bearing.append(candidate)
        if len(bearing) == 1:
            return bearing[0]
        names = sorted(str(c.getName()) for c in candidates)
        raise ValueError(f"ambiguous motor mount — pass mount= (candidates: {', '.join(names)})")

    def get_motor(self, sim, mount=None) -> str | None:
        """The designation of the motor the simulation would fly on the
        given (or auto-resolved) mount, as a plain string; None when the
        sim's configuration has no motor there.
        """
        resolved = self._resolve_mount(sim, mount)
        config = self._motor_config(resolved, self._sim_fcid(sim))
        motor = config.getMotor() if config is not None else None
        return str(motor.getDesignation()) if motor is not None else None

    def find_motor(self, designation: str, manufacturer: str | None = None):
        """A motor from OpenRocket's own database by designation
        (case-insensitive exact). Common hobby designations (A8, B6, C6…)
        exist from several manufacturers — those lookups require
        ``manufacturer=`` (motor choice is safety-relevant; orlab refuses to
        guess); manufacturer matching uses OpenRocket's own alias machinery,
        so "CTI" finds Cesaroni. One manufacturer's designation can span
        several motor sets (different diameters/lengths): sets are ordered
        by diameter, then length, and the first variant of the first set is
        returned — OpenRocket keeps each set's variants deterministically
        sorted, so 'the' motor is stable across runs.

        :raises ValueError: no match (message lists near-matches), a
            designation that exists but not from the given manufacturer, or
            an ambiguous designation without ``manufacturer=``.
        """
        wanted = designation.strip().lower()
        database = self.openrocket.startup.Application.getMotorSetDatabase()
        matches = [
            s for s in database.getMotorSets() if str(s.getDesignation()).strip().lower() == wanted
        ]
        if manufacturer is not None and matches:
            # OpenRocket's Manufacturer.matches handles display/simple names
            # and aliases (CTI, CES, ...) on every supported version
            filtered = [s for s in matches if s.getManufacturer().matches(manufacturer)]
            if not filtered:
                available = sorted({str(s.getManufacturer().getDisplayName()) for s in matches})
                raise ValueError(
                    f"{designation!r} exists, but not from {manufacturer!r} "
                    f"(available: {', '.join(available)})"
                )
            matches = filtered
        if not matches:
            near = sorted(
                {
                    str(s.getDesignation())
                    for s in database.getMotorSets()
                    if wanted in str(s.getDesignation()).lower()
                    or str(s.getDesignation()).lower() in wanted
                }
            )[:10]
            hint = f"; near matches: {', '.join(near)}" if near else ""
            raise ValueError(f"no motor matches designation {designation!r}{hint}")
        makers = sorted({str(s.getManufacturer().getDisplayName()) for s in matches})
        if len(makers) > 1:
            raise ValueError(
                f"{designation!r} exists from several manufacturers "
                f"({', '.join(makers)}) — pass manufacturer="
            )
        # deterministic set choice by intrinsic geometry; the set's own
        # variant list is already deterministically sorted by OpenRocket
        chosen = min(
            matches,
            key=lambda s: (float(s.getDiameter()), float(s.getLength()), str(s.getType())),
        )
        return chosen.getMotors()[0]

    def load_motor(self, motor_file, designation: str | None = None):
        """A motor loaded from a thrust-curve file (.eng/.rse/.zip) via
        OpenRocket's own loader — no database needed, works on every
        supported version and startup path. Files holding several motors
        need ``designation=`` to pick one.

        :raises OrlabError: the file doesn't parse as a motor file.
        """
        path = os.fspath(motor_file)
        if not os.path.exists(path):
            raise FileNotFoundError(f"No such motor file: {path}")
        loader = self.openrocket.file.motor.GeneralMotorLoader()
        stream = jpype.java.io.FileInputStream(path)
        try:
            loaded = list(loader.load(stream, os.path.basename(path)))
        except Exception as e:
            raise OrlabError(f"{path} did not parse as a motor file ({e})") from e
        finally:
            stream.close()
        motors = [m.build() if hasattr(m, "build") else m for m in loaded]
        if not motors:
            raise OrlabError(f"{path} holds no motors")
        if designation is not None:
            wanted = designation.strip().lower()
            selected = [m for m in motors if str(m.getDesignation()).strip().lower() == wanted]
            if not selected:
                names = sorted(str(m.getDesignation()) for m in motors)
                raise ValueError(
                    f"{path} holds no motor designated {designation!r} "
                    f"(it holds: {', '.join(names)})"
                )
            motors = selected
        elif len(motors) > 1:
            names = sorted(str(m.getDesignation()) for m in motors)
            raise ValueError(
                f"{path} holds {len(names)} motors ({', '.join(names)}) — pass designation="
            )
        return motors[0]

    def set_motor(
        self,
        sim,
        motor,
        *,
        mount=None,
        manufacturer: str | None = None,
        designation: str | None = None,
        delay: float | None = None,
    ) -> None:
        """Sets the motor the simulation will actually fly — always keyed on
        the sim's own flight configuration (assigning to the rocket's
        *selected* config is the ecosystem's classic silent failure). The
        motor argument is a designation string (database lookup, see
        :meth:`find_motor`), a ``.eng``/``.rse``/``.zip`` path (file load,
        see :meth:`load_motor`), or a ThrustCurveMotor object. ``delay=``
        sets the ejection delay in seconds; None preserves the existing one.
        The assignment is read back and verified — a mismatch raises instead
        of failing silently.
        """
        if isinstance(motor, (str, os.PathLike)):
            text = os.fspath(motor)
            if text.lower().endswith((".eng", ".rse", ".zip")):
                if manufacturer is not None:
                    raise ValueError("manufacturer= does not apply to a motor file")
                motor = self.load_motor(text, designation=designation)
            else:
                if designation is not None:
                    raise ValueError(
                        "designation= selects a motor from a multi-motor "
                        "FILE; for database lookups the designation IS the "
                        "motor argument"
                    )
                motor = self.find_motor(text, manufacturer=manufacturer)
        target_designation = str(motor.getDesignation())

        resolved = self._resolve_mount(sim, mount)
        fcid = self._sim_fcid(sim)
        config = self._motor_config(resolved, fcid)
        if config is None:
            raise OrlabError(
                f"mount {resolved.getName()} has no motor configuration for "
                "this simulation's flight configuration"
            )
        config.setMotor(motor)
        if delay is not None:
            config.setEjectionDelay(float(delay))

        readback = self.get_motor(sim, mount=mount)
        if readback != target_designation:
            raise OrlabError(
                f"motor assignment did not stick: set {target_designation}, "
                f"read back {readback} (wrong mount or flight configuration?)"
            )

    def get_components_of_type(self, root, component_type) -> list:
        """Every component under root of a given type — a class name string
        ("BodyTube", "TrapezoidFinSet", "MotorMount", …) resolved against
        the loaded OpenRocket, or a JClass. Concrete classes, superclasses,
        and interfaces all match; an empty list is a normal answer.

        :raises ValueError: an unknown class name (named with the loaded
            version — component classes differ across OpenRocket releases).
        """
        if isinstance(component_type, str):
            java_class = getattr(self.openrocket.rocketcomponent, component_type, None)
            if not isinstance(java_class, jpype.JClass):
                raise ValueError(
                    f"{component_type!r} is not a rocket-component class in "
                    f"OpenRocket {self._instance.or_version}"
                )
            component_type = java_class
        elif not isinstance(component_type, jpype.JClass):
            # a Python class would silently match nothing; a component
            # instance would crash isinstance — refuse both clearly
            raise TypeError(
                "component_type must be a class-name string or a JClass, "
                f"got {type(component_type).__name__}"
            )
        return [c for c in JIterator(root) if isinstance(c, component_type)]

    def get_component_named(self, root, name):
        """Finds and returns the first rocket component with the given name.
        Requires a root RocketComponent, usually this will be a RocketComponent.rocket instance.
        Raises a ValueError if no component found.
        """

        for component in JIterator(root):
            if component.getName() == name:
                return component
        raise ValueError(root.toString() + " has no component named " + name)

    def get_all_components(self, root) -> list[jpype.JObject]:
        """Returns a list of all rocket components in the loaded OpenRocket file.

        :param root: The root RocketComponent (usually obtained from the simulation)
        :return: List of all component objects
        """
        components = []
        for component in JIterator(root):
            components.append(component)
        return components

load_doc(or_filename)

Loads a .ork file and returns the corresponding openrocket document

Source code in src/orlab/core/helper.py

def load_doc(self, or_filename):
    """Loads a .ork file and returns the corresponding openrocket document"""

    or_java_file = jpype.java.io.File(or_filename)
    loader = self.openrocket.file.GeneralRocketLoader(or_java_file)
    doc = loader.load()
    return doc

save_doc(or_filename, doc)

Saves an openrocket document to a .ork file

Source code in src/orlab/core/helper.py

def save_doc(self, or_filename, doc):
    """Saves an openrocket document to a .ork file"""

    or_java_file = jpype.java.io.File(or_filename)
    saver = self.openrocket.file.GeneralRocketSaver()
    saver.save(or_java_file, doc)

run_simulation(sim, listeners=None, *, randomize_seed=True)

This is a wrapper to the Simulation.simulate() for running a simulation The optional listeners parameter is a sequence of objects which extend orl.AbstractSimulationListener.

By default the simulation's random seed is randomized before each run (identical repeated runs would otherwise produce identical numbers — the behavior naive monte-carlo loops rely on). Pass randomize_seed=False to respect the seed already set on the options. Note: on OpenRocket 24.12 the wind model draws additional per-process entropy, so a fixed seed reproduces results within one process but not across processes when wind is enabled.

Source code in src/orlab/core/helper.py

def run_simulation(
    self,
    sim,
    listeners: list[AbstractSimulationListener] | None = None,
    *,
    randomize_seed: bool = True,
):
    """This is a wrapper to the Simulation.simulate() for running a simulation
    The optional listeners parameter is a sequence of objects which extend orl.AbstractSimulationListener.

    By default the simulation's random seed is randomized before each run
    (identical repeated runs would otherwise produce identical numbers —
    the behavior naive monte-carlo loops rely on). Pass
    randomize_seed=False to respect the seed already set on the options.
    Note: on OpenRocket 24.12 the wind model draws additional per-process
    entropy, so a fixed seed reproduces results within one process but not
    across processes when wind is enabled.
    """

    if listeners is None:
        # this method takes in a vararg of SimulationListeners, which is just a fancy way of passing in an array, so
        # we have to pass in an array of length 0 ..
        listener_array = jpype.JArray(
            self.openrocket.simulation.listeners.AbstractSimulationListener, 1
        )(0)
    else:
        listener_array = [
            jpype.JProxy(
                (
                    self.openrocket.simulation.listeners.SimulationListener,
                    self.openrocket.simulation.listeners.SimulationEventListener,
                    self.openrocket.simulation.listeners.SimulationComputationListener,
                    jpype.java.lang.Cloneable,
                ),
                inst=c,
            )
            for c in listeners
        ]

    if randomize_seed:
        sim.getOptions().randomizeSeed()
    sim.simulate(listener_array)

translate_flight_data_type(flight_data_type)

Translates a FlightDataType (or constant name) to the Java constant. Raises UnsupportedFlightDataType when the loaded OpenRocket version does not expose it (constants differ across versions).

Source code in src/orlab/core/helper.py

def translate_flight_data_type(self, flight_data_type: FlightDataType | str):
    """Translates a FlightDataType (or constant name) to the Java constant.
    Raises UnsupportedFlightDataType when the loaded OpenRocket version
    does not expose it (constants differ across versions).
    """
    if isinstance(flight_data_type, FlightDataType):
        name = flight_data_type.name
    elif isinstance(flight_data_type, str):
        name = flight_data_type
    else:
        raise TypeError("Invalid type for flight_data_type")

    java_type = getattr(self.openrocket.simulation.FlightDataType, name, None)
    if java_type is None:
        loaded = self._instance.or_version
        available = versions_with(name)
        detail = f"available in: {', '.join(available)}" if available else "unknown constant"
        raise UnsupportedFlightDataType(
            f"{name} is not available in OpenRocket {loaded} ({detail})"
        )
    return java_type

get_timeseries(simulation, variables, branch_number=0)

Gets a dictionary of timeseries data (as numpy arrays) from a simulation given specific variable names.

Parameters:

Name	Type	Description	Default
simulation		An openrocket simulation object.	required
variables	Iterable[FlightDataType | str]	A sequence of FlightDataType or strings representing the desired variables	required
branch_number		Stage branch to read (0 = sustainer).	0

Returns:

Type	Description
dict[FlightDataType | str, ndarray]	Dict keyed by the requested variables; values are numpy arrays.

Source code in src/orlab/core/helper.py

def get_timeseries(
    self, simulation, variables: Iterable[FlightDataType | str], branch_number=0
) -> dict[FlightDataType | str, np.ndarray]:
    """
    Gets a dictionary of timeseries data (as numpy arrays) from a simulation given specific variable names.

    :param simulation: An openrocket simulation object.
    :param variables: A sequence of FlightDataType or strings representing the desired variables
    :param branch_number: Stage branch to read (0 = sustainer).
    :return: Dict keyed by the requested variables; values are numpy arrays.
    """

    branch = simulation.getSimulatedData().getBranch(branch_number)
    output: dict[FlightDataType | str, np.ndarray] = {}
    for v in variables:
        output[v] = np.array(branch.get(self.translate_flight_data_type(v)))

    return output

get_final_values(simulation, variables, branch_number=0)

Gets a the final value in the time series from a simulation given variable names.

Parameters:

Name	Type	Description	Default
simulation		An openrocket simulation object.	required
variables	Iterable[FlightDataType | str]	A sequence of FlightDataType or strings representing the desired variables	required
branch_number		Stage branch to read (0 = sustainer).	0

Returns:

Type	Description
dict[FlightDataType | str, float]	Dict keyed by the requested variables; values are the final samples.

Source code in src/orlab/core/helper.py

def get_final_values(
    self, simulation, variables: Iterable[FlightDataType | str], branch_number=0
) -> dict[FlightDataType | str, float]:
    """
    Gets a the final value in the time series from a simulation given variable names.

    :param simulation: An openrocket simulation object.
    :param variables: A sequence of FlightDataType or strings representing the desired variables
    :param branch_number: Stage branch to read (0 = sustainer).
    :return: Dict keyed by the requested variables; values are the final samples.
    """

    branch = simulation.getSimulatedData().getBranch(branch_number)
    output: dict[FlightDataType | str, float] = {}
    for v in variables:
        output[v] = branch.get(self.translate_flight_data_type(v))[-1]

    return output

translate_flight_event(flight_event)

Translates a Java FlightEvent.Type constant to the FlightEvent enum. Raises ValueError for event types this orlab version does not know (newer OpenRocket releases add types; get_events skips them instead).

Source code in src/orlab/core/helper.py

def translate_flight_event(self, flight_event) -> FlightEvent:
    """Translates a Java FlightEvent.Type constant to the FlightEvent enum.
    Raises ValueError for event types this orlab version does not know
    (newer OpenRocket releases add types; get_events skips them instead).
    """
    name = str(flight_event.name())
    try:
        return FlightEvent[name]
    except KeyError:
        raise ValueError(
            f"Unknown flight event type {name!r} from the loaded OpenRocket version"
        ) from None

get_events(simulation, branch_number=0)

Returns a dictionary of all the flight events in a given simulation. Key is FlightEvent and value is a list of all the times at which the event occurs. Event types not known to this orlab version are skipped with a warning.

Parameters:

Name	Type	Description	Default
branch_number	int	Stage branch to read (0 = sustainer).	0

Source code in src/orlab/core/helper.py

def get_events(self, simulation, branch_number: int = 0) -> dict[FlightEvent, list[float]]:
    """Returns a dictionary of all the flight events in a given simulation.
    Key is FlightEvent and value is a list of all the times at which the event occurs.
    Event types not known to this orlab version are skipped with a warning.

    :param branch_number: Stage branch to read (0 = sustainer).
    """
    branch = simulation.getSimulatedData().getBranch(branch_number)

    output: dict[FlightEvent, list[float]] = {}
    unknown: set[str] = set()
    for name, times in self._events_by_name(branch).items():
        try:
            event = FlightEvent[name]
        except KeyError:
            if name not in unknown:
                unknown.add(name)
                logger.warning("Skipping unknown flight event type %s", name)
            continue
        output[event] = times

    return output

get_summary(simulation, branch_number=0)

Returns the scalar flight report for one branch as a :class:~orlab.FlightSummary — plain-Python values only, safe to pickle into JVM-less processes.

Branch 0 uses OpenRocket's own FlightData summary getters verbatim; other branches derive the same numbers from their own time series. Missing/inapplicable values are NaN (see the field docs). Degenerate flights degrade to NaN; an unsimulated simulation raises OrlabError.

Parameters:

Name	Type	Description	Default
branch_number	int	Stage branch to summarize (0 = sustainer).	0

Source code in src/orlab/core/helper.py

def get_summary(self, simulation, branch_number: int = 0) -> FlightSummary:
    """Returns the scalar flight report for one branch as a
    :class:`~orlab.FlightSummary` — plain-Python values only, safe to
    pickle into JVM-less processes.

    Branch 0 uses OpenRocket's own FlightData summary getters verbatim;
    other branches derive the same numbers from their own time series.
    Missing/inapplicable values are NaN (see the field docs). Degenerate
    flights degrade to NaN; an unsimulated simulation raises OrlabError.

    :param branch_number: Stage branch to summarize (0 = sustainer).
    """
    branch_number = int(branch_number)  # builtin-types contract incl. numpy ints
    flight_data = simulation.getSimulatedData()
    if flight_data is None or int(flight_data.getBranchCount()) == 0:
        raise OrlabError("Simulation has no flight data; call run_simulation first")
    branch_count = int(flight_data.getBranchCount())
    if not 0 <= branch_number < branch_count:
        raise IndexError(f"branch {branch_number} out of range (simulation has {branch_count})")
    branch = flight_data.getBranch(branch_number)
    version = self._instance.or_version

    times = self._branch_series(branch, "TYPE_TIME")
    altitude = self._branch_series(branch, "TYPE_ALTITUDE")
    stability = self._branch_series(branch, "TYPE_STABILITY")
    velocity_z = self._branch_series(branch, "TYPE_VELOCITY_Z")
    velocity_total = self._branch_series(branch, "TYPE_VELOCITY_TOTAL")
    events = self._events_by_name(branch)

    def at_event(name: str, series: np.ndarray, which: int = 0) -> float:
        if name not in events:
            return math.nan
        return _value_at(times, series, events[name][which])

    # stability window: launch rod departure -> apogee. Boosters have no
    # LAUNCHROD (window starts at their first sample, off-rod values NaN);
    # a flight without an APOGEE event windows to its highest sample.
    t_rod = events["LAUNCHROD"][0] if "LAUNCHROD" in events else None
    if "APOGEE" in events:
        t_apogee = events["APOGEE"][0]
    elif len(altitude) and not np.isnan(altitude).all():
        t_apogee = float(times[int(np.nanargmax(altitude))])
    else:
        t_apogee = None
    window_start = t_rod if t_rod is not None else (float(times[0]) if len(times) else None)
    if window_start is not None and t_apogee is not None:
        min_stab, max_stab = _window_stats(times, stability, window_start, t_apogee)
    else:
        min_stab, max_stab = math.nan, math.nan

    t_deploy = (
        events["RECOVERY_DEVICE_DEPLOYMENT"][-1]
        if "RECOVERY_DEVICE_DEPLOYMENT" in events
        else None
    )
    t_ground = (
        events["GROUND_HIT"][0]
        if "GROUND_HIT" in events
        else (float(times[-1]) if len(times) else None)
    )
    if t_deploy is not None and t_ground is not None:
        descent_rate = _mean_descent_rate(times, velocity_z, t_deploy, t_ground)
    else:
        descent_rate = math.nan

    if branch_number == 0:

        def scalar(getter: str) -> float:
            fn = getattr(flight_data, getter, None)
            if fn is None:
                _warn_absent_once(f"FlightData.{getter}", version)
                return math.nan
            return float(fn())

        apogee = scalar("getMaxAltitude")
        time_to_apogee = scalar("getTimeToApogee")
        max_velocity = scalar("getMaxVelocity")
        max_acceleration = scalar("getMaxAcceleration")
        max_mach = scalar("getMaxMachNumber")
        velocity_off_rod = scalar("getLaunchRodVelocity")
        velocity_at_deployment = scalar("getDeploymentVelocity")
        ground_hit_velocity = scalar("getGroundHitVelocity")
        flight_time = scalar("getFlightTime")
        optimum_delay = scalar("getOptimumDelay")
    else:
        acceleration = self._branch_series(branch, "TYPE_ACCELERATION_TOTAL")
        mach = self._branch_series(branch, "TYPE_MACH_NUMBER")

        def series_max(series: np.ndarray) -> float:
            if len(series) == 0 or np.isnan(series).all():
                return math.nan
            return float(np.nanmax(series))

        apogee = series_max(altitude)
        time_to_apogee = t_apogee if t_apogee is not None else math.nan
        max_velocity = series_max(velocity_total)
        max_acceleration = series_max(acceleration)
        max_mach = series_max(mach)
        velocity_off_rod = at_event("LAUNCHROD", velocity_total)
        velocity_at_deployment = at_event("RECOVERY_DEVICE_DEPLOYMENT", velocity_total, -1)
        ground_hit_velocity = at_event("GROUND_HIT", velocity_total)
        flight_time = float(times[-1]) if len(times) else math.nan
        # policy: reported for the sustainer only, where FlightData's own
        # figure is authoritative
        optimum_delay = math.nan

    pos_x = self._branch_series(branch, "TYPE_POSITION_X")
    pos_y = self._branch_series(branch, "TYPE_POSITION_Y")
    landing_x = float(pos_x[-1]) if len(pos_x) else math.nan
    landing_y = float(pos_y[-1]) if len(pos_y) else math.nan
    landing_distance = math.hypot(landing_x, landing_y)

    if hasattr(branch, "getName"):
        branch_name = str(branch.getName())
    elif hasattr(branch, "getBranchName"):
        branch_name = str(branch.getBranchName())
    else:
        branch_name = ""

    # bounded at apogee: the NaN-skip must not report a post-apogee
    # (tumble-regime) sample as the off-rod margin
    off_rod_stability = math.nan
    if t_rod is not None:
        off_rod_stability = _value_at(
            times, stability, t_rod, t_max=t_apogee if t_apogee is not None else math.inf
        )

    return FlightSummary(
        velocity_off_rod=velocity_off_rod,
        stability_off_rod_cal=off_rod_stability,
        apogee=apogee,
        time_to_apogee=time_to_apogee,
        max_velocity=max_velocity,
        max_acceleration=max_acceleration,
        max_mach=max_mach,
        min_stability_cal=min_stab,
        max_stability_cal=max_stab,
        velocity_at_deployment=velocity_at_deployment,
        descent_rate=descent_rate,
        ground_hit_velocity=ground_hit_velocity,
        landing_x=landing_x,
        landing_y=landing_y,
        landing_distance=landing_distance,
        landing_bearing_deg=_bearing_deg(landing_x, landing_y),
        flight_time=flight_time,
        optimum_delay=optimum_delay,
        branch_number=branch_number,
        branch_name=branch_name,
        branch_count=branch_count,
        warnings=tuple(str(w) for w in flight_data.getWarningSet()),
    )

get_dataframe(simulation, variables=None, branch_number=0)

The branch's timeseries as a pandas DataFrame, one column per variable, labeled NAME (SI unit). Requires the orlab[pandas] extra; everything else in orlab works without pandas.

Parameters:

Name	Type	Description	Default
variables		FlightDataTypes or constant names; default: every profile data type populated on the branch.	None
branch_number	int	Stage branch to read (0 = sustainer).	0

Source code in src/orlab/core/helper.py

def get_dataframe(self, simulation, variables=None, branch_number: int = 0):
    """The branch's timeseries as a pandas DataFrame, one column per
    variable, labeled ``NAME (SI unit)``. Requires the ``orlab[pandas]``
    extra; everything else in orlab works without pandas.

    :param variables: FlightDataTypes or constant names; default: every
        profile data type populated on the branch.
    :param branch_number: Stage branch to read (0 = sustainer).
    """
    try:
        import pandas  # type: ignore[import-untyped]  # optional extra, no stubs
    except ImportError as e:
        raise ImportError(
            "pandas is required for get_dataframe — pip install orlab[pandas]"
        ) from e
    branch = simulation.getSimulatedData().getBranch(branch_number)
    columns = self._tabular_columns(branch, variables)
    return pandas.DataFrame(
        {label: np.asarray(branch.get(java_type), dtype=float) for label, java_type in columns}
    )

export_csv(simulation, path, variables=None, branch_number=0)

Writes the branch's timeseries as UTF-8 CSV with NAME (SI unit) headers — stdlib only, no pandas needed. NaN samples become empty cells (pandas.read_csv reads them back as NaN).

Parameters:

Name	Type	Description	Default
variables		FlightDataTypes or constant names; default: every profile data type populated on the branch.	None
branch_number	int	Stage branch to read (0 = sustainer).	0

Source code in src/orlab/core/helper.py

def export_csv(self, simulation, path, variables=None, branch_number: int = 0) -> None:
    """Writes the branch's timeseries as UTF-8 CSV with ``NAME (SI
    unit)`` headers — stdlib only, no pandas needed. NaN samples become
    empty cells (``pandas.read_csv`` reads them back as NaN).

    :param variables: FlightDataTypes or constant names; default: every
        profile data type populated on the branch.
    :param branch_number: Stage branch to read (0 = sustainer).
    """
    import csv

    branch = simulation.getSimulatedData().getBranch(branch_number)
    columns = self._tabular_columns(branch, variables)
    series = [np.asarray(branch.get(java_type), dtype=float) for _, java_type in columns]
    # newline='' and explicit utf-8: text-mode newline translation would
    # write \r\r\n on Windows, and locale encodings choke on m/s²
    with open(path, "w", newline="", encoding="utf-8") as fh:
        writer = csv.writer(fh)
        writer.writerow([label for label, _ in columns])
        try:
            for row in zip(*series, strict=True):
                writer.writerow(
                    ["" if math.isnan(value) else repr(float(value)) for value in row]
                )
        except ValueError as e:  # pragma: no cover - can't-happen invariant
            raise ValueError(f"branch series lengths differ: {e}") from e

get_motor(sim, mount=None)

The designation of the motor the simulation would fly on the given (or auto-resolved) mount, as a plain string; None when the sim's configuration has no motor there.

Source code in src/orlab/core/helper.py

def get_motor(self, sim, mount=None) -> str | None:
    """The designation of the motor the simulation would fly on the
    given (or auto-resolved) mount, as a plain string; None when the
    sim's configuration has no motor there.
    """
    resolved = self._resolve_mount(sim, mount)
    config = self._motor_config(resolved, self._sim_fcid(sim))
    motor = config.getMotor() if config is not None else None
    return str(motor.getDesignation()) if motor is not None else None

find_motor(designation, manufacturer=None)

A motor from OpenRocket's own database by designation (case-insensitive exact). Common hobby designations (A8, B6, C6…) exist from several manufacturers — those lookups require manufacturer= (motor choice is safety-relevant; orlab refuses to guess); manufacturer matching uses OpenRocket's own alias machinery, so "CTI" finds Cesaroni. One manufacturer's designation can span several motor sets (different diameters/lengths): sets are ordered by diameter, then length, and the first variant of the first set is returned — OpenRocket keeps each set's variants deterministically sorted, so 'the' motor is stable across runs.

Raises:

Type	Description
ValueError	no match (message lists near-matches), a designation that exists but not from the given manufacturer, or an ambiguous designation without manufacturer=.

Source code in src/orlab/core/helper.py

def find_motor(self, designation: str, manufacturer: str | None = None):
    """A motor from OpenRocket's own database by designation
    (case-insensitive exact). Common hobby designations (A8, B6, C6…)
    exist from several manufacturers — those lookups require
    ``manufacturer=`` (motor choice is safety-relevant; orlab refuses to
    guess); manufacturer matching uses OpenRocket's own alias machinery,
    so "CTI" finds Cesaroni. One manufacturer's designation can span
    several motor sets (different diameters/lengths): sets are ordered
    by diameter, then length, and the first variant of the first set is
    returned — OpenRocket keeps each set's variants deterministically
    sorted, so 'the' motor is stable across runs.

    :raises ValueError: no match (message lists near-matches), a
        designation that exists but not from the given manufacturer, or
        an ambiguous designation without ``manufacturer=``.
    """
    wanted = designation.strip().lower()
    database = self.openrocket.startup.Application.getMotorSetDatabase()
    matches = [
        s for s in database.getMotorSets() if str(s.getDesignation()).strip().lower() == wanted
    ]
    if manufacturer is not None and matches:
        # OpenRocket's Manufacturer.matches handles display/simple names
        # and aliases (CTI, CES, ...) on every supported version
        filtered = [s for s in matches if s.getManufacturer().matches(manufacturer)]
        if not filtered:
            available = sorted({str(s.getManufacturer().getDisplayName()) for s in matches})
            raise ValueError(
                f"{designation!r} exists, but not from {manufacturer!r} "
                f"(available: {', '.join(available)})"
            )
        matches = filtered
    if not matches:
        near = sorted(
            {
                str(s.getDesignation())
                for s in database.getMotorSets()
                if wanted in str(s.getDesignation()).lower()
                or str(s.getDesignation()).lower() in wanted
            }
        )[:10]
        hint = f"; near matches: {', '.join(near)}" if near else ""
        raise ValueError(f"no motor matches designation {designation!r}{hint}")
    makers = sorted({str(s.getManufacturer().getDisplayName()) for s in matches})
    if len(makers) > 1:
        raise ValueError(
            f"{designation!r} exists from several manufacturers "
            f"({', '.join(makers)}) — pass manufacturer="
        )
    # deterministic set choice by intrinsic geometry; the set's own
    # variant list is already deterministically sorted by OpenRocket
    chosen = min(
        matches,
        key=lambda s: (float(s.getDiameter()), float(s.getLength()), str(s.getType())),
    )
    return chosen.getMotors()[0]

load_motor(motor_file, designation=None)

A motor loaded from a thrust-curve file (.eng/.rse/.zip) via OpenRocket's own loader — no database needed, works on every supported version and startup path. Files holding several motors need designation= to pick one.

Raises:

Type	Description
OrlabError	the file doesn't parse as a motor file.

Source code in src/orlab/core/helper.py

def load_motor(self, motor_file, designation: str | None = None):
    """A motor loaded from a thrust-curve file (.eng/.rse/.zip) via
    OpenRocket's own loader — no database needed, works on every
    supported version and startup path. Files holding several motors
    need ``designation=`` to pick one.

    :raises OrlabError: the file doesn't parse as a motor file.
    """
    path = os.fspath(motor_file)
    if not os.path.exists(path):
        raise FileNotFoundError(f"No such motor file: {path}")
    loader = self.openrocket.file.motor.GeneralMotorLoader()
    stream = jpype.java.io.FileInputStream(path)
    try:
        loaded = list(loader.load(stream, os.path.basename(path)))
    except Exception as e:
        raise OrlabError(f"{path} did not parse as a motor file ({e})") from e
    finally:
        stream.close()
    motors = [m.build() if hasattr(m, "build") else m for m in loaded]
    if not motors:
        raise OrlabError(f"{path} holds no motors")
    if designation is not None:
        wanted = designation.strip().lower()
        selected = [m for m in motors if str(m.getDesignation()).strip().lower() == wanted]
        if not selected:
            names = sorted(str(m.getDesignation()) for m in motors)
            raise ValueError(
                f"{path} holds no motor designated {designation!r} "
                f"(it holds: {', '.join(names)})"
            )
        motors = selected
    elif len(motors) > 1:
        names = sorted(str(m.getDesignation()) for m in motors)
        raise ValueError(
            f"{path} holds {len(names)} motors ({', '.join(names)}) — pass designation="
        )
    return motors[0]

set_motor(sim, motor, *, mount=None, manufacturer=None, designation=None, delay=None)

Sets the motor the simulation will actually fly — always keyed on the sim's own flight configuration (assigning to the rocket's selected config is the ecosystem's classic silent failure). The motor argument is a designation string (database lookup, see :meth:find_motor), a .eng/.rse/.zip path (file load, see :meth:load_motor), or a ThrustCurveMotor object. delay= sets the ejection delay in seconds; None preserves the existing one. The assignment is read back and verified — a mismatch raises instead of failing silently.

Source code in src/orlab/core/helper.py

def set_motor(
    self,
    sim,
    motor,
    *,
    mount=None,
    manufacturer: str | None = None,
    designation: str | None = None,
    delay: float | None = None,
) -> None:
    """Sets the motor the simulation will actually fly — always keyed on
    the sim's own flight configuration (assigning to the rocket's
    *selected* config is the ecosystem's classic silent failure). The
    motor argument is a designation string (database lookup, see
    :meth:`find_motor`), a ``.eng``/``.rse``/``.zip`` path (file load,
    see :meth:`load_motor`), or a ThrustCurveMotor object. ``delay=``
    sets the ejection delay in seconds; None preserves the existing one.
    The assignment is read back and verified — a mismatch raises instead
    of failing silently.
    """
    if isinstance(motor, (str, os.PathLike)):
        text = os.fspath(motor)
        if text.lower().endswith((".eng", ".rse", ".zip")):
            if manufacturer is not None:
                raise ValueError("manufacturer= does not apply to a motor file")
            motor = self.load_motor(text, designation=designation)
        else:
            if designation is not None:
                raise ValueError(
                    "designation= selects a motor from a multi-motor "
                    "FILE; for database lookups the designation IS the "
                    "motor argument"
                )
            motor = self.find_motor(text, manufacturer=manufacturer)
    target_designation = str(motor.getDesignation())

    resolved = self._resolve_mount(sim, mount)
    fcid = self._sim_fcid(sim)
    config = self._motor_config(resolved, fcid)
    if config is None:
        raise OrlabError(
            f"mount {resolved.getName()} has no motor configuration for "
            "this simulation's flight configuration"
        )
    config.setMotor(motor)
    if delay is not None:
        config.setEjectionDelay(float(delay))

    readback = self.get_motor(sim, mount=mount)
    if readback != target_designation:
        raise OrlabError(
            f"motor assignment did not stick: set {target_designation}, "
            f"read back {readback} (wrong mount or flight configuration?)"
        )

get_components_of_type(root, component_type)

Every component under root of a given type — a class name string ("BodyTube", "TrapezoidFinSet", "MotorMount", …) resolved against the loaded OpenRocket, or a JClass. Concrete classes, superclasses, and interfaces all match; an empty list is a normal answer.

Raises:

Type	Description
ValueError	an unknown class name (named with the loaded version — component classes differ across OpenRocket releases).

Source code in src/orlab/core/helper.py

def get_components_of_type(self, root, component_type) -> list:
    """Every component under root of a given type — a class name string
    ("BodyTube", "TrapezoidFinSet", "MotorMount", …) resolved against
    the loaded OpenRocket, or a JClass. Concrete classes, superclasses,
    and interfaces all match; an empty list is a normal answer.

    :raises ValueError: an unknown class name (named with the loaded
        version — component classes differ across OpenRocket releases).
    """
    if isinstance(component_type, str):
        java_class = getattr(self.openrocket.rocketcomponent, component_type, None)
        if not isinstance(java_class, jpype.JClass):
            raise ValueError(
                f"{component_type!r} is not a rocket-component class in "
                f"OpenRocket {self._instance.or_version}"
            )
        component_type = java_class
    elif not isinstance(component_type, jpype.JClass):
        # a Python class would silently match nothing; a component
        # instance would crash isinstance — refuse both clearly
        raise TypeError(
            "component_type must be a class-name string or a JClass, "
            f"got {type(component_type).__name__}"
        )
    return [c for c in JIterator(root) if isinstance(c, component_type)]

get_component_named(root, name)

Finds and returns the first rocket component with the given name. Requires a root RocketComponent, usually this will be a RocketComponent.rocket instance. Raises a ValueError if no component found.

Source code in src/orlab/core/helper.py

def get_component_named(self, root, name):
    """Finds and returns the first rocket component with the given name.
    Requires a root RocketComponent, usually this will be a RocketComponent.rocket instance.
    Raises a ValueError if no component found.
    """

    for component in JIterator(root):
        if component.getName() == name:
            return component
    raise ValueError(root.toString() + " has no component named " + name)

get_all_components(root)

Returns a list of all rocket components in the loaded OpenRocket file.

Parameters:

Name	Type	Description	Default
root		The root RocketComponent (usually obtained from the simulation)	required

Returns:

Type	Description
list[JObject]	List of all component objects

Source code in src/orlab/core/helper.py

def get_all_components(self, root) -> list[jpype.JObject]:
    """Returns a list of all rocket components in the loaded OpenRocket file.

    :param root: The root RocketComponent (usually obtained from the simulation)
    :return: List of all component objects
    """
    components = []
    for component in JIterator(root):
        components.append(component)
    return components

AbstractSimulationListener

Python implementation of OpenRocket's AbstractSimulationListener. Subclass it, override the hooks you need, and pass instances to Helper.run_simulation(sim, listeners=[...]).

OpenRocket clones listeners before the run, so collect results through shared mutable state (e.g. a list the caller keeps a reference to), not by assigning to self and reading it back afterwards. Exceptions raised inside a hook propagate out of run_simulation intact.

Hook groups: SimulationListener (startSimulation, postStep, …), SimulationEventListener (handleFlightEvent, …), and SimulationComputationListener (preWindModel, postAerodynamicCalculation, …). Boolean-returning hooks continue the simulation/event on True; pre-computation hooks return an override value or None/NaN to leave OpenRocket's computation untouched.

Source code in src/orlab/core/simulation_listener.py

class AbstractSimulationListener:
    """Python implementation of OpenRocket's AbstractSimulationListener.
    Subclass it, override the hooks you need, and pass instances to
    ``Helper.run_simulation(sim, listeners=[...])``.

    OpenRocket clones listeners before the run, so collect results through
    shared mutable state (e.g. a list the caller keeps a reference to), not
    by assigning to ``self`` and reading it back afterwards. Exceptions
    raised inside a hook propagate out of ``run_simulation`` intact.

    Hook groups: SimulationListener (``startSimulation``, ``postStep``, …),
    SimulationEventListener (``handleFlightEvent``, …), and
    SimulationComputationListener (``preWindModel``,
    ``postAerodynamicCalculation``, …). Boolean-returning hooks continue the
    simulation/event on True; pre-computation hooks return an override value
    or None/NaN to leave OpenRocket's computation untouched.
    """

    def __str__(self):
        return "'" + "Python simulation listener proxy : " + str(self.__class__.__name__) + "'"

    def toString(self):
        return str(self)

    # SimulationListener
    def startSimulation(self, status) -> None:
        pass

    def endSimulation(self, status, simulation_exception) -> None:
        pass

    def preStep(self, status) -> bool:
        return True

    def postStep(self, status) -> None:
        pass

    def isSystemListener(self) -> bool:
        return False

    # SimulationEventListener
    def addFlightEvent(self, status, flight_event) -> bool:
        return True

    def handleFlightEvent(self, status, flight_event) -> bool:
        return True

    def motorIgnition(self, status, motor_id, motor_mount, motor_instance) -> bool:
        return True

    def recoveryDeviceDeployment(self, status, recovery_device) -> bool:
        return True

    # SimulationComputationListener
    def preAccelerationCalculation(self, status):
        return None

    def preAerodynamicCalculation(self, status):
        return None

    def preAtmosphericModel(self, status):
        return None

    def preFlightConditions(self, status):
        return None

    def preGravityModel(self, status):
        return float("nan")

    def preMassCalculation(self, status):
        return None

    def preSimpleThrustCalculation(self, status):
        return float("nan")

    def preWindModel(self, status):
        return None

    def postAccelerationCalculation(self, status, acceleration_data):
        return None

    def postAerodynamicCalculation(self, status, aerodynamic_forces):
        return None

    def postAtmosphericModel(self, status, atmospheric_conditions):
        return None

    def postFlightConditions(self, status, flight_conditions):
        return None

    def postGravityModel(self, status, gravity):
        return float("nan")

    def postMassCalculation(self, status, mass_data):
        return None

    def postSimpleThrustCalculation(self, status, thrust):
        return float("nan")

    def postWindModel(self, status, wind):
        return None

    def clone(self):
        openrocket = active_core_root()
        return jpype.JProxy(
            (
                openrocket.simulation.listeners.SimulationListener,
                openrocket.simulation.listeners.SimulationEventListener,
                openrocket.simulation.listeners.SimulationComputationListener,
                jpype.java.lang.Cloneable,
            ),
            inst=copy(self),
        )

JIterator

Wraps a Java iterable as a Python iterator.

Built for walking a rocket's component tree: for component in JIterator(rocket): print(component.getName())

Parameters:

Name	Type	Description	Default
jit		any Java object exposing iterator(boolean) — in practice a RocketComponent (the boolean requests iteration over the full subtree, the component itself included).	required

Source code in src/orlab/core/jiterator.py

class JIterator:
    """Wraps a Java iterable as a Python iterator.

    Built for walking a rocket's component tree:
    ``for component in JIterator(rocket): print(component.getName())``

    :param jit: any Java object exposing ``iterator(boolean)`` — in practice a
        ``RocketComponent`` (the boolean requests iteration over the full
        subtree, the component itself included).
    """

    def __init__(self, jit):
        self.jit = jit.iterator(True)

    def __iter__(self):
        return self

    def __next__(self):
        if not self.jit.hasNext():
            raise StopIteration()
        else:
            return next(self.jit)

FlightSummary

Scalar summary of one simulated flight branch.

All values are builtin floats/ints/strs (never Java or numpy types), in SI units per :attr:UNITS. Missing or inapplicable values are math.nan, never None — a booster branch has no launch rod departure, pre-23.09 OpenRocket computes no optimum delay, a flight without a recovery event has no descent rate. The field set only grows; pickles are same-orlab-version transport, not an archive format.

Source code in src/orlab/core/summary.py

@dataclass(frozen=True)
class FlightSummary:
    """Scalar summary of one simulated flight branch.

    All values are builtin floats/ints/strs (never Java or numpy types), in
    SI units per :attr:`UNITS`. Missing or inapplicable values are
    ``math.nan``, never ``None`` — a booster branch has no launch rod
    departure, pre-23.09 OpenRocket computes no optimum delay, a flight
    without a recovery event has no descent rate. The field set only grows;
    pickles are same-orlab-version transport, not an archive format.
    """

    velocity_off_rod: float
    """Speed when leaving the launch rod (m/s); NaN on branches without a
    LAUNCHROD event (boosters)."""
    stability_off_rod_cal: float
    """Stability margin (calibers) at launch rod departure; NaN without a
    LAUNCHROD event."""
    apogee: float
    """Highest altitude above the launch site (m)."""
    time_to_apogee: float
    """Time from liftoff to apogee (s)."""
    max_velocity: float
    """Largest total velocity (m/s)."""
    max_acceleration: float
    """Largest total acceleration (m/s²)."""
    max_mach: float
    """Largest Mach number."""
    min_stability_cal: float
    """Smallest stability margin (calibers) between launch rod departure and
    apogee. The window matters: OpenRocket 24.12 computes stability through
    post-apogee tumble, where unwindowed minima are meaningless."""
    max_stability_cal: float
    """Largest stability margin (calibers) in the same window."""
    velocity_at_deployment: float
    """Total velocity at the last recovery device deployment (m/s); NaN
    without a deployment."""
    descent_rate: float
    """Sample mean of vertical descent speed (m/s, positive down) over the
    samples from the last recovery device deployment to ground hit; NaN
    without a deployment."""
    ground_hit_velocity: float
    """Speed at ground contact (m/s)."""
    landing_x: float
    """Landing position east of the launch site (m, flat-earth)."""
    landing_y: float
    """Landing position north of the launch site (m, flat-earth)."""
    landing_distance: float
    """Straight-line distance from launch site to landing (m)."""
    landing_bearing_deg: float
    """Compass bearing from launch site to landing (degrees, 0 = north,
    90 = east). Numerically defined but physically meaningless when
    landing_distance ≈ 0."""
    flight_time: float
    """Total simulated flight time (s)."""
    optimum_delay: float
    """OpenRocket's optimum ejection delay (s); NaN before OpenRocket
    23.09, which does not compute it."""
    branch_number: int
    """Which stage branch this summary describes (0 = sustainer)."""
    branch_name: str
    """OpenRocket's name for the branch (e.g. "Sustainer")."""
    branch_count: int
    """How many branches the simulation produced."""
    warnings: tuple[str, ...]
    """Simulation warnings, verbatim from OpenRocket (opaque strings)."""

    UNITS: ClassVar[dict[str, str]] = {
        "velocity_off_rod": "m/s",
        "stability_off_rod_cal": "cal",
        "apogee": "m",
        "time_to_apogee": "s",
        "max_velocity": "m/s",
        "max_acceleration": "m/s²",
        "max_mach": "",
        "min_stability_cal": "cal",
        "max_stability_cal": "cal",
        "velocity_at_deployment": "m/s",
        "descent_rate": "m/s",
        "ground_hit_velocity": "m/s",
        "landing_x": "m",
        "landing_y": "m",
        "landing_distance": "m",
        "landing_bearing_deg": "°",
        "flight_time": "s",
        "optimum_delay": "s",
    }

    _SECTIONS: ClassVar[list[tuple[str, list[str]]]] = [
        ("Rail departure", ["velocity_off_rod", "stability_off_rod_cal"]),
        (
            "Ascent",
            [
                "apogee",
                "time_to_apogee",
                "max_velocity",
                "max_acceleration",
                "max_mach",
                "min_stability_cal",
                "max_stability_cal",
                "optimum_delay",
            ],
        ),
        (
            "Recovery & descent",
            ["velocity_at_deployment", "descent_rate", "ground_hit_velocity"],
        ),
        (
            "Landing",
            [
                "landing_x",
                "landing_y",
                "landing_distance",
                "landing_bearing_deg",
                "flight_time",
            ],
        ),
    ]

    def to_dict(self) -> dict:
        """The summary as a flat dict — one dispersion-study row.
        ``pandas.DataFrame([s.to_dict() for s in summaries])`` is the table.
        """
        return {f.name: getattr(self, f.name) for f in fields(self)}

    def __str__(self) -> str:
        lines = [
            f"Flight summary — branch {self.branch_number} of "
            f"{self.branch_count} ({self.branch_name})"
        ]
        for section, names in self._SECTIONS:
            lines.append(f"  {section}:")
            for name in names:
                value = getattr(self, name)
                unit = self.UNITS.get(name, "")
                shown = "n/a" if isinstance(value, float) and math.isnan(value) else f"{value:.6g}"
                lines.append(f"    {name.replace('_', ' ')}: {shown}{f' {unit}' if unit else ''}")
        lines.append("  Warnings:")
        if self.warnings:
            lines.extend(f"    - {w}" for w in self.warnings)
        else:
            lines.append("    none")
        return "\n".join(lines)

velocity_off_rod instance-attribute

Speed when leaving the launch rod (m/s); NaN on branches without a LAUNCHROD event (boosters).

stability_off_rod_cal instance-attribute

Stability margin (calibers) at launch rod departure; NaN without a LAUNCHROD event.

apogee instance-attribute

Highest altitude above the launch site (m).

time_to_apogee instance-attribute

Time from liftoff to apogee (s).

max_velocity instance-attribute

Largest total velocity (m/s).

max_acceleration instance-attribute

Largest total acceleration (m/s²).

max_mach instance-attribute

Largest Mach number.

min_stability_cal instance-attribute

Smallest stability margin (calibers) between launch rod departure and apogee. The window matters: OpenRocket 24.12 computes stability through post-apogee tumble, where unwindowed minima are meaningless.

max_stability_cal instance-attribute

Largest stability margin (calibers) in the same window.

velocity_at_deployment instance-attribute

Total velocity at the last recovery device deployment (m/s); NaN without a deployment.

descent_rate instance-attribute

Sample mean of vertical descent speed (m/s, positive down) over the samples from the last recovery device deployment to ground hit; NaN without a deployment.

ground_hit_velocity instance-attribute

Speed at ground contact (m/s).

landing_x instance-attribute

Landing position east of the launch site (m, flat-earth).

landing_y instance-attribute

Landing position north of the launch site (m, flat-earth).

landing_distance instance-attribute

Straight-line distance from launch site to landing (m).

landing_bearing_deg instance-attribute

Compass bearing from launch site to landing (degrees, 0 = north, 90 = east). Numerically defined but physically meaningless when landing_distance ≈ 0.

flight_time instance-attribute

Total simulated flight time (s).

optimum_delay instance-attribute

OpenRocket's optimum ejection delay (s); NaN before OpenRocket 23.09, which does not compute it.

branch_number instance-attribute

Which stage branch this summary describes (0 = sustainer).

branch_name instance-attribute

OpenRocket's name for the branch (e.g. "Sustainer").

branch_count instance-attribute

How many branches the simulation produced.

warnings instance-attribute

Simulation warnings, verbatim from OpenRocket (opaque strings).

to_dict()

The summary as a flat dict — one dispersion-study row. pandas.DataFrame([s.to_dict() for s in summaries]) is the table.

Source code in src/orlab/core/summary.py

def to_dict(self) -> dict:
    """The summary as a flat dict — one dispersion-study row.
    ``pandas.DataFrame([s.to_dict() for s in summaries])`` is the table.
    """
    return {f.name: getattr(self, f.name) for f in fields(self)}

Jar management

Downloads (or returns the cached copy of) the OpenRocket jar for version, verified by sha256. version=None means the default, orlab._pins.DEFAULT_VERSION.

Versions orlab pins are verified against the shipped pin. Any other version requires sha256= — compute it yourself from a copy you trust; there is no way to skip verification.

Raises:

Type	Description
JarVerificationError	no digest to verify against, or a downloaded jar whose digest does not match (a corrupt cached entry is evicted and re-downloaded once first).
ValueError	a malformed version string or digest, or sha256= contradicting a shipped pin.

Source code in src/orlab/jars.py

def fetch_jar(version: str | None = None, *, sha256: str | None = None) -> Path:
    """Downloads (or returns the cached copy of) the OpenRocket jar for
    ``version``, verified by sha256. ``version=None`` means the default,
    ``orlab._pins.DEFAULT_VERSION``.

    Versions orlab pins are verified against the shipped pin. Any other
    version requires ``sha256=`` — compute it yourself from a copy you
    trust; there is no way to skip verification.

    :raises JarVerificationError: no digest to verify against, or a
        downloaded jar whose digest does not match (a corrupt cached entry
        is evicted and re-downloaded once first).
    :raises ValueError: a malformed version string or digest, or ``sha256=``
        contradicting a shipped pin.
    """
    if version is None:
        version = DEFAULT_VERSION
    if not _VERSION_RE.fullmatch(version):
        raise ValueError(f"not an OpenRocket version string: {version!r}")
    if sha256 is not None:
        sha256 = sha256.lower()
        if not re.fullmatch(r"[0-9a-f]{64}", sha256):
            raise ValueError(f"not a sha256 hex digest: {sha256!r}")
    pin = PINNED_SHA256.get(version)
    if pin is not None and sha256 is not None and sha256 != pin:
        raise ValueError(
            f"sha256= for OpenRocket {version} contradicts orlab's pin: "
            f"given {sha256}, pinned {pin}"
        )
    expected = pin or sha256
    url = RELEASE_URL.format(v=version)
    cache = jar_cache_dir()
    path = cache / f"OpenRocket-{version}.jar"

    if expected is None:
        cached_digest = _sha256_or_none(path)
        found = f"\nA cached file exists with sha256 {cached_digest}." if cached_digest else ""
        raise JarVerificationError(
            f"orlab {_orlab_version()} has no sha256 pin for OpenRocket {version}, "
            f"so it will not fetch it unverified. Download {url} from a "
            "machine/network you trust, compute its digest (`sha256sum "
            f"OpenRocket-{version}.jar`), and pass it: "
            f"fetch_jar({version!r}, sha256=...). A newer orlab may pin this "
            "version already." + found
        )

    cache.mkdir(parents=True, exist_ok=True)
    digest = _sha256_or_none(path)
    if digest == expected:
        return path
    if digest is not None:
        logger.warning("cached %s failed sha256 verification; re-downloading", path)
        _evict(path)
    _fetch_verified(url, path, expected)
    return path

The directory fetch_jar caches jars in: $ORLAB_JAR_CACHE, else $XDG_CACHE_HOME/orlab-jars, else ~/.cache/orlab-jars.

Source code in src/orlab/jars.py

def jar_cache_dir() -> Path:
    """The directory ``fetch_jar`` caches jars in: ``$ORLAB_JAR_CACHE``,
    else ``$XDG_CACHE_HOME/orlab-jars``, else ``~/.cache/orlab-jars``.
    """
    env = os.environ.get("ORLAB_JAR_CACHE")
    if env:
        return Path(env)
    xdg = os.environ.get("XDG_CACHE_HOME")
    base = Path(xdg) if xdg else Path.home() / ".cache"
    return base / "orlab-jars"

Enums

orlab.FlightDataType

Bases: Enum

Flight-data series selectable in Helper.get_timeseries / get_final_values — the union across all supported OpenRocket versions. Requesting one the loaded version lacks raises UnsupportedFlightDataType naming the versions that have it.

Source code in src/orlab/_enums.py

class FlightDataType(Enum):
    """Flight-data series selectable in Helper.get_timeseries /
    get_final_values — the union across all supported OpenRocket versions.
    Requesting one the loaded version lacks raises
    UnsupportedFlightDataType naming the versions that have it."""

    TYPE_ACCELERATION_TOTAL = auto()
    TYPE_ACCELERATION_XY = auto()
    TYPE_ACCELERATION_Z = auto()
    TYPE_AIR_DENSITY = auto()
    TYPE_AIR_PRESSURE = auto()
    TYPE_AIR_TEMPERATURE = auto()
    TYPE_ALTITUDE = auto()
    TYPE_ALTITUDE_ABOVE_SEA = auto()
    TYPE_AOA = auto()
    TYPE_AXIAL_DRAG_COEFF = auto()
    TYPE_BASE_DRAG_COEFF = auto()
    TYPE_CG_LOCATION = auto()
    TYPE_COMPUTATION_TIME = auto()
    TYPE_CORIOLIS_ACCELERATION = auto()
    TYPE_CP_LOCATION = auto()
    TYPE_DRAG_COEFF = auto()
    TYPE_DRAG_FORCE = auto()
    TYPE_FRICTION_DRAG_COEFF = auto()
    TYPE_GRAVITY = auto()
    TYPE_LATITUDE = auto()
    TYPE_LONGITUDE = auto()
    TYPE_LONGITUDINAL_INERTIA = auto()
    TYPE_MACH_NUMBER = auto()
    TYPE_MASS = auto()
    TYPE_MOTOR_MASS = auto()
    TYPE_NORMAL_FORCE_COEFF = auto()
    TYPE_ORIENTATION_PHI = auto()
    TYPE_ORIENTATION_THETA = auto()
    TYPE_PITCH_DAMPING_MOMENT_COEFF = auto()
    TYPE_PITCH_MOMENT_COEFF = auto()
    TYPE_PITCH_RATE = auto()
    TYPE_POSITION_DIRECTION = auto()
    TYPE_POSITION_X = auto()
    TYPE_POSITION_XY = auto()
    TYPE_POSITION_Y = auto()
    TYPE_PRESSURE_DRAG_COEFF = auto()
    TYPE_PROPELLANT_MASS = auto()
    TYPE_REFERENCE_AREA = auto()
    TYPE_REFERENCE_LENGTH = auto()
    TYPE_REYNOLDS_NUMBER = auto()
    TYPE_ROLL_DAMPING_COEFF = auto()
    TYPE_ROLL_FORCING_COEFF = auto()
    TYPE_ROLL_MOMENT_COEFF = auto()
    TYPE_ROLL_RATE = auto()
    TYPE_ROTATIONAL_INERTIA = auto()
    TYPE_SIDE_FORCE_COEFF = auto()
    TYPE_SPEED_OF_SOUND = auto()
    TYPE_STABILITY = auto()
    TYPE_THRUST_FORCE = auto()
    TYPE_THRUST_WEIGHT_RATIO = auto()
    TYPE_TIME = auto()
    TYPE_TIME_STEP = auto()
    TYPE_VELOCITY_TOTAL = auto()
    TYPE_VELOCITY_XY = auto()
    TYPE_VELOCITY_Z = auto()
    TYPE_WIND_DIRECTION = auto()
    TYPE_WIND_VELOCITY = auto()
    TYPE_YAW_DAMPING_MOMENT_COEFF = auto()
    TYPE_YAW_MOMENT_COEFF = auto()
    TYPE_YAW_RATE = auto()

orlab.FlightEvent

Bases: Enum

Flight events returned by Helper.get_events — the union across all supported OpenRocket versions (SIM_WARN/SIM_ABORT are 24.12+). Unknown event types from newer jars are skipped with a warning.

Source code in src/orlab/_enums.py

class FlightEvent(Enum):
    """Flight events returned by Helper.get_events — the union across
    all supported OpenRocket versions (SIM_WARN/SIM_ABORT are 24.12+).
    Unknown event types from newer jars are skipped with a warning."""

    ALTITUDE = auto()
    APOGEE = auto()
    BURNOUT = auto()
    EJECTION_CHARGE = auto()
    EXCEPTION = auto()
    GROUND_HIT = auto()
    IGNITION = auto()
    LAUNCH = auto()
    LAUNCHROD = auto()
    LIFTOFF = auto()
    RECOVERY_DEVICE_DEPLOYMENT = auto()
    SIMULATION_END = auto()
    SIM_ABORT = auto()
    SIM_WARN = auto()
    STAGE_SEPARATION = auto()
    TUMBLE = auto()

orlab.OrLogLevel

Bases: Enum

OpenRocket's logback log levels, for OpenRocketInstance(log_level=...).

Source code in src/orlab/_enums.py

class OrLogLevel(Enum):
    """OpenRocket's logback log levels, for OpenRocketInstance(log_level=...)."""

    OFF = auto()
    ERROR = auto()
    WARN = auto()
    INFO = auto()
    DEBUG = auto()
    TRACE = auto()
    ALL = auto()

FlightDataType and FlightEvent are generated as the union of constants across all supported OpenRocket versions; availability on the loaded version is enforced at translation time with UnsupportedFlightDataType.

Parallel running and declarative options

Parallel simulation running and the declarative options contract.

SimulationPool runs dispersion studies across worker processes — one JVM plus one loaded .ork per worker (JPype allows one JVM per process, so parallelism is multi-process by construction). Tasks are plain mappings of DECLARATIVE_KEYS (portable, validated, notebook-safe) or an importable worker_fn for anything the whitelist doesn't cover. Results come back as plain-Python records; per-task failures are data, and every abort path preserves the partial results collected so far.

DECLARATIVE_KEYS is the curated set of SimulationOptions knobs that are verified to apply-and-read-back identically on every supported OpenRocket version (15.03, 22.02, 23.09, 24.12 — probed per release, and re-checked by the profile contract manifest and an all-version integration case).

One interaction to know: OpenRocket launches into the wind by default, and while launch_into_wind is true the rod-direction getter reports the into-wind direction — launch_rod_direction only takes effect together with launch_into_wind: False (verified on all four versions; the pool refuses a task set that varies the direction without pinning the flag).

DeclarativeKey

Bases: NamedTuple

How one declarative key maps onto SimulationOptions.

Source code in src/orlab/parallel.py

class DeclarativeKey(NamedTuple):
    """How one declarative key maps onto SimulationOptions."""

    setter: str
    getter: str
    kind: type  # value type: float or bool
    unit: str  # SI unit for float keys ("" for bool/dimensionless)

SimResult dataclass

One successful simulation in a study.

Source code in src/orlab/parallel.py

@dataclass(frozen=True)
class SimResult:
    """One successful simulation in a study."""

    index: int
    """Submission order (results are returned sorted by this)."""
    task: dict
    """The task mapping as submitted."""
    seed: int
    """The seed the simulation actually used, read back after the run."""
    seed_reassigned: bool
    """True when the readback differs from the pool-assigned seed — e.g. a
    worker_fn that called run_simulation with randomize_seed=True. The
    recorded seed is still the authoritative one for replay."""
    payload: Any
    """The default extractor's summary fields, or worker_fn's return value.
    Always plain-Python data (pickle-checked in the worker)."""

index instance-attribute

Submission order (results are returned sorted by this).

task instance-attribute

The task mapping as submitted.

seed instance-attribute

The seed the simulation actually used, read back after the run.

seed_reassigned instance-attribute

True when the readback differs from the pool-assigned seed — e.g. a worker_fn that called run_simulation with randomize_seed=True. The recorded seed is still the authoritative one for replay.

payload instance-attribute

The default extractor's summary fields, or worker_fn's return value. Always plain-Python data (pickle-checked in the worker).

SimError dataclass

One failed simulation in a study — failure as data, not an abort.

Source code in src/orlab/parallel.py

@dataclass(frozen=True)
class SimError:
    """One failed simulation in a study — failure as data, not an abort."""

    index: int
    task: dict
    seed: int
    error_type: str
    """Exception type name; Java exceptions keep their class name."""
    message: str
    traceback: str
    """Python traceback, plus the Java stack when one exists."""

error_type instance-attribute

Exception type name; Java exceptions keep their class name.

traceback instance-attribute

Python traceback, plus the Java stack when one exists.

StudyResult dataclass

Everything a study produced, in submission order.

Source code in src/orlab/parallel.py

@dataclass(frozen=True)
class StudyResult:
    """Everything a study produced, in submission order."""

    results: tuple[SimResult, ...]
    errors: tuple[SimError, ...]

    def to_records(self) -> list[dict]:
        """Flat dicts (task keys + payload + seed/index) — a dispersion
        table is ``pandas.DataFrame(study.to_records())``. Column precedence
        on name collisions: the reserved ``index``/``seed``/
        ``seed_reassigned`` columns override task keys, and payload keys
        override both (the declarative path is collision-free by
        construction)."""
        records = []
        for r in self.results:
            record = dict(r.task)
            # reserved columns win over task keys; payload keys win last
            record.update(index=r.index, seed=r.seed, seed_reassigned=r.seed_reassigned)
            if isinstance(r.payload, dict):
                record.update(r.payload)
            else:
                record["payload"] = r.payload
            records.append(record)
        return records

    def raise_if_errors(self) -> "StudyResult":
        """Returns self, or raises OrlabError describing the first failure
        when any task errored."""
        if self.errors:
            first = self.errors[0]
            raise OrlabError(
                f"{len(self.errors)} of {len(self.results) + len(self.errors)} "
                f"simulations failed; first: task {first.index} "
                f"({first.error_type}: {first.message})"
            )
        return self

to_records()

Flat dicts (task keys + payload + seed/index) — a dispersion table is pandas.DataFrame(study.to_records()). Column precedence on name collisions: the reserved index/seed/ seed_reassigned columns override task keys, and payload keys override both (the declarative path is collision-free by construction).

Source code in src/orlab/parallel.py

def to_records(self) -> list[dict]:
    """Flat dicts (task keys + payload + seed/index) — a dispersion
    table is ``pandas.DataFrame(study.to_records())``. Column precedence
    on name collisions: the reserved ``index``/``seed``/
    ``seed_reassigned`` columns override task keys, and payload keys
    override both (the declarative path is collision-free by
    construction)."""
    records = []
    for r in self.results:
        record = dict(r.task)
        # reserved columns win over task keys; payload keys win last
        record.update(index=r.index, seed=r.seed, seed_reassigned=r.seed_reassigned)
        if isinstance(r.payload, dict):
            record.update(r.payload)
        else:
            record["payload"] = r.payload
        records.append(record)
    return records

raise_if_errors()

Returns self, or raises OrlabError describing the first failure when any task errored.

Source code in src/orlab/parallel.py

def raise_if_errors(self) -> "StudyResult":
    """Returns self, or raises OrlabError describing the first failure
    when any task errored."""
    if self.errors:
        first = self.errors[0]
        raise OrlabError(
            f"{len(self.errors)} of {len(self.results) + len(self.errors)} "
            f"simulations failed; first: task {first.index} "
            f"({first.error_type}: {first.message})"
        )
    return self

SimulationPool

A spawn-based worker pool for dispersion studies: each worker boots one JVM, loads the .ork once, and runs many simulations against it.

Constructing the pool validates everything cheap (jar, document, simulation index); workers boot lazily on the first :meth:run. Use from a script's if __name__ == "__main__": block — a requirement of multiprocessing's spawn start method, which is the only start method that is safe with a JVM and uniform across platforms.

Source code in src/orlab/parallel.py

class SimulationPool:
    """A spawn-based worker pool for dispersion studies: each worker boots
    one JVM, loads the ``.ork`` once, and runs many simulations against it.

    Constructing the pool validates everything cheap (jar, document,
    simulation index); workers boot lazily on the first :meth:`run`. Use
    from a script's ``if __name__ == "__main__":`` block — a requirement of
    multiprocessing's spawn start method, which is the only start method
    that is safe with a JVM and uniform across platforms.
    """

    def __init__(
        self,
        ork_file,
        jar_path=None,
        *,
        simulation_index: int = 0,
        max_workers: int | None = None,
        log_level: OrLogLevel | str = OrLogLevel.ERROR,
        jvm_path=None,
        jvm_args=(),
        max_tasks_per_child: int | None = None,
        worker_stdout: str = "discard",
        _test_init=None,
    ):
        """ork_file/jar_path resolve exactly like OpenRocketInstance (the
        default jar resolution chain applies) but are pinned to absolute
        paths here — workers never re-run environment-dependent resolution.
        max_workers defaults to min(4, cpu_count), clamped to the first
        run's task count; each worker holds a JVM (hundreds of MB — size
        jvm_args=("-Xmx512m",) accordingly). max_tasks_per_child recycles
        workers (Python 3.11+). worker_stdout='discard' silences worker
        stdout including the JVM's native writes on POSIX (best-effort on
        Windows); stderr is always inherited.
        """
        from .core.openrocket_instance import OpenRocketInstance, _default_jar_path

        if worker_stdout not in ("discard", "inherit"):
            raise ValueError("worker_stdout must be 'discard' or 'inherit'")
        if max_tasks_per_child is not None and sys.version_info < (3, 11):
            raise OrlabError("max_tasks_per_child needs Python 3.11+")
        ork = os.path.abspath(os.fspath(ork_file))
        if not os.path.exists(ork):
            raise FileNotFoundError(f"No such .ork file: {ork}")
        jar = (
            os.path.abspath(os.fspath(jar_path))
            if jar_path is not None
            else os.path.abspath(_default_jar_path())
        )
        # validates the jar and fires the fallback-profile warning once,
        # in the parent, instead of once per worker
        OpenRocketInstance(jar, log_level=log_level, jvm_path=jvm_path, jvm_args=tuple(jvm_args))
        count = _count_simulations(ork)
        if count is not None and not 0 <= simulation_index < max(count, 1):
            raise IndexError(
                f"simulation_index {simulation_index} out of range: {ork} defines {count}"
            )
        self._cfg = {
            "ork_file": ork,
            "jar_path": jar,
            "simulation_index": simulation_index,
            "log_level": log_level,
            "jvm_path": os.fspath(jvm_path) if jvm_path is not None else None,
            "jvm_args": tuple(jvm_args),
            "worker_stdout": worker_stdout,
            "_test_init": _test_init,
        }
        self._max_workers = max_workers
        self._max_tasks_per_child = max_tasks_per_child
        self._executor: ProcessPoolExecutor | None = None
        self._broken = False

    # -- validation helpers (pure; unit-tested directly) --

    @staticmethod
    def _validate_declarative(tasks: list[dict]) -> None:
        keysets = {frozenset(t) - {"seed"} for t in tasks}
        if len(keysets) > 1:
            union = frozenset().union(*keysets)
            common = frozenset(union)
            for keys in keysets:
                common &= keys
            raise ValueError(
                "declarative tasks must share one key set; "
                f"{sorted(union - common)} appear in some tasks but not all"
            )
        keys = keysets.pop() if keysets else frozenset()
        unknown = keys - set(DECLARATIVE_KEYS)
        if unknown:
            raise ValueError(
                f"unknown declarative keys {sorted(unknown)}; the verified "
                f"set is {sorted(DECLARATIVE_KEYS)} (use worker_fn= for "
                "anything else)"
            )
        if "launch_rod_direction" in keys and "launch_into_wind" not in keys:
            raise ValueError(
                "launch_rod_direction has no effect while OpenRocket's "
                "launch-into-wind default is active — include "
                "launch_into_wind: False in the tasks"
            )
        for task in tasks:
            if "seed" in task:
                seed = task["seed"]
                if (
                    not isinstance(seed, int)
                    or isinstance(seed, bool)
                    or not _SEED_MIN <= seed <= _SEED_MAX
                ):
                    raise OverflowError(
                        f"task seed {seed!r} is not a signed 32-bit int "
                        "(OpenRocket seeds are Java ints)"
                    )
            for key, value in task.items():
                if key == "seed":
                    continue
                spec = DECLARATIVE_KEYS[key]
                if spec.kind is float and isinstance(value, bool):
                    raise TypeError(f"{key} expects a number, got a bool")
                if not isinstance(value, (int, float) if spec.kind is float else bool):
                    raise TypeError(
                        f"{key} expects {spec.kind.__name__}, got {type(value).__name__}"
                    )

    @staticmethod
    def _validate_worker_fn(worker_fn) -> None:
        try:
            pickle.dumps(worker_fn)
        except Exception as e:
            raise ValueError(
                f"worker_fn is not picklable ({e}); define it at module "
                "level in an importable module — lambdas, closures, and "
                "locally defined functions cannot cross the spawn boundary"
            ) from e
        module = getattr(worker_fn, "__module__", "")
        if module in ("__main__", "__mp_main__"):
            main = sys.modules.get("__main__")
            main_file = getattr(main, "__file__", None)
            if not (main_file and os.path.exists(main_file)):
                raise ValueError(
                    "worker_fn is defined in an interactive __main__ "
                    "(notebook/REPL) — spawn workers cannot re-import it. "
                    "Save it to a .py module and import it (in a notebook: "
                    "%%writefile workers.py, then from workers import ...)"
                )

    @staticmethod
    def _derive_seeds(tasks: list[dict], seed) -> list[int]:
        """Unique 31-bit seeds per task (Java's own randomizeSeed draws a
        full 32-bit space where 10k-run studies collide by birthday
        statistics); task['seed'] overrides."""
        # note: several tasks pinning the SAME seed are honored verbatim —
        # uniqueness is guaranteed only for derived seeds
        rng = random.Random(seed) if seed is not None else random.SystemRandom()
        used: set[int] = {t["seed"] for t in tasks if "seed" in t}
        seeds = []
        for task in tasks:
            if "seed" in task:
                seeds.append(task["seed"])
                continue
            while True:
                candidate = rng.getrandbits(31)
                if candidate not in used:
                    used.add(candidate)
                    seeds.append(candidate)
                    break
        return seeds

    def _ensure_executor(self, task_count: int) -> ProcessPoolExecutor:
        if self._broken:
            raise OrlabError(
                "this SimulationPool is dead after a worker crash or interrupt; create a new one"
            )
        if self._executor is None:
            workers = self._max_workers
            if workers is None:
                # the default is sized to the first run; an explicit value is
                # respected verbatim (idle workers just idle)
                workers = max(1, min(min(4, os.cpu_count() or 1), task_count))
            kwargs: dict = {
                "max_workers": workers,
                "mp_context": multiprocessing.get_context("spawn"),
                "initializer": _worker_init,
                "initargs": (self._cfg,),
            }
            if self._max_tasks_per_child is not None:
                kwargs["max_tasks_per_child"] = self._max_tasks_per_child
            self._executor = ProcessPoolExecutor(**kwargs)
        return self._executor

    def run(
        self,
        tasks,
        *,
        worker_fn=None,
        seed=None,
        progress=None,
        on_error: str = "collect",
    ) -> StudyResult:
        """Runs one task per simulation and collects the results.

        tasks: an iterable of mappings (DECLARATIVE_KEYS names, plus an
        optional reserved 'seed'), or an int meaning that many runs of the
        document as-is. worker_fn(helper, sim, task): an importable,
        module-level callable taking over the whole task body — set options
        yourself (everything you vary), run the simulation, return
        plain-Python data. seed: study seed for reproducible seed
        derivation; default draws from SystemRandom. progress(done, total):
        called with (0, total) up front — JVM boot takes seconds and bars
        should render — then once per completion; wrap
        ``pool.run(...)``'s completions with tqdm by passing
        ``progress=lambda done, total: bar.update(...)`` or iterate
        to_records afterwards. on_error: 'collect' (failures are SimError
        data) or 'abort' (first failure cancels the study).

        :raises StudyAborted: interrupt / worker-crash / worker-init /
            task-error (with ``.partial`` holding everything collected).
        """
        if on_error not in ("collect", "abort"):
            raise ValueError("on_error must be 'collect' or 'abort'")
        if isinstance(tasks, bool):
            raise TypeError("tasks must be an iterable of mappings or an int")
        if isinstance(tasks, int):
            if tasks < 0:
                raise ValueError(f"cannot run {tasks} simulations")
            task_list: list[dict] = [{} for _ in range(tasks)]
        else:
            try:
                task_list = [dict(t) for t in tasks]
            except (TypeError, ValueError) as e:
                raise TypeError(
                    "tasks must be an iterable of mappings or an int "
                    f"(got {type(tasks).__name__}); did you pass a single "
                    "mapping instead of a list?"
                ) from e
        if worker_fn is None:
            self._validate_declarative(task_list)
        else:
            for task in task_list:
                if "seed" in task:
                    self._validate_declarative([{"seed": task["seed"]}])
            self._validate_worker_fn(worker_fn)
            try:
                pickle.dumps(task_list)
            except Exception as e:
                raise ValueError(
                    f"tasks are not picklable ({e}); worker_fn tasks must "
                    "hold plain-Python data only"
                ) from e
        if not task_list:
            return StudyResult((), ())
        seeds = self._derive_seeds(task_list, seed)

        results: list[SimResult] = []
        errors: list[SimError] = []
        total = len(task_list)

        def partial() -> StudyResult:
            return StudyResult(
                tuple(sorted(results, key=lambda r: r.index)),
                tuple(sorted(errors, key=lambda e: e.index)),
            )

        def kill_pool() -> None:
            self._broken = True
            if self._executor is not None:
                self._executor.shutdown(wait=False, cancel_futures=True)
                self._executor = None

        futures: list = []
        try:
            executor = self._ensure_executor(total)
            for index, task in enumerate(task_list):
                futures.append(executor.submit(_worker_run, index, task, seeds[index], worker_fn))
            if progress is not None:
                try:
                    progress(0, total)
                except BaseException:
                    for f in futures:
                        f.cancel()
                    raise
            for done, future in enumerate(as_completed(futures), start=1):
                outcome = future.result()
                kind = outcome[0]
                if kind == "init_error":
                    for f in futures:
                        f.cancel()
                    kill_pool()  # init config is identical across workers:
                    # every worker holds an unusable JVM — release them
                    raise StudyAborted(
                        "worker-init",
                        partial(),
                        f"a worker failed to initialize:\n{outcome[2]}",
                    )
                index = outcome[1]
                if kind == "ok":
                    _, _, payload, actual_seed, reassigned = outcome
                    results.append(
                        SimResult(index, task_list[index], actual_seed, reassigned, payload)
                    )
                else:
                    _, _, error_type, message, tb = outcome
                    errors.append(
                        SimError(index, task_list[index], seeds[index], error_type, message, tb)
                    )
                    if on_error == "abort":
                        for f in futures:
                            f.cancel()
                        raise StudyAborted(
                            "task-error",
                            partial(),
                            f"task {index} failed ({error_type}: {message}) and on_error='abort'",
                        )
                if progress is not None:
                    try:
                        progress(done, total)
                    except BaseException:
                        for f in futures:
                            f.cancel()
                        raise
            return partial()
        except KeyboardInterrupt:
            kill_pool()
            raise StudyAborted(
                "interrupt", partial(), "interrupted; partial results preserved"
            ) from None
        except BrokenProcessPool as e:
            kill_pool()
            raise StudyAborted(
                "worker-crash",
                partial(),
                "a worker process died (JVM crash or out-of-memory?) — "
                "consider jvm_args=('-Xmx512m',) per worker, fewer workers, "
                "or max_tasks_per_child",
            ) from e

    def shutdown(self) -> None:
        """Stops the workers (their JVMs end with their processes). The pool
        cannot be reused afterwards."""
        self._broken = True
        if self._executor is not None:
            self._executor.shutdown(wait=True, cancel_futures=True)
            self._executor = None

__init__(ork_file, jar_path=None, *, simulation_index=0, max_workers=None, log_level=OrLogLevel.ERROR, jvm_path=None, jvm_args=(), max_tasks_per_child=None, worker_stdout='discard', _test_init=None)

ork_file/jar_path resolve exactly like OpenRocketInstance (the default jar resolution chain applies) but are pinned to absolute paths here — workers never re-run environment-dependent resolution. max_workers defaults to min(4, cpu_count), clamped to the first run's task count; each worker holds a JVM (hundreds of MB — size jvm_args=("-Xmx512m",) accordingly). max_tasks_per_child recycles workers (Python 3.11+). worker_stdout='discard' silences worker stdout including the JVM's native writes on POSIX (best-effort on Windows); stderr is always inherited.

Source code in src/orlab/parallel.py

def __init__(
    self,
    ork_file,
    jar_path=None,
    *,
    simulation_index: int = 0,
    max_workers: int | None = None,
    log_level: OrLogLevel | str = OrLogLevel.ERROR,
    jvm_path=None,
    jvm_args=(),
    max_tasks_per_child: int | None = None,
    worker_stdout: str = "discard",
    _test_init=None,
):
    """ork_file/jar_path resolve exactly like OpenRocketInstance (the
    default jar resolution chain applies) but are pinned to absolute
    paths here — workers never re-run environment-dependent resolution.
    max_workers defaults to min(4, cpu_count), clamped to the first
    run's task count; each worker holds a JVM (hundreds of MB — size
    jvm_args=("-Xmx512m",) accordingly). max_tasks_per_child recycles
    workers (Python 3.11+). worker_stdout='discard' silences worker
    stdout including the JVM's native writes on POSIX (best-effort on
    Windows); stderr is always inherited.
    """
    from .core.openrocket_instance import OpenRocketInstance, _default_jar_path

    if worker_stdout not in ("discard", "inherit"):
        raise ValueError("worker_stdout must be 'discard' or 'inherit'")
    if max_tasks_per_child is not None and sys.version_info < (3, 11):
        raise OrlabError("max_tasks_per_child needs Python 3.11+")
    ork = os.path.abspath(os.fspath(ork_file))
    if not os.path.exists(ork):
        raise FileNotFoundError(f"No such .ork file: {ork}")
    jar = (
        os.path.abspath(os.fspath(jar_path))
        if jar_path is not None
        else os.path.abspath(_default_jar_path())
    )
    # validates the jar and fires the fallback-profile warning once,
    # in the parent, instead of once per worker
    OpenRocketInstance(jar, log_level=log_level, jvm_path=jvm_path, jvm_args=tuple(jvm_args))
    count = _count_simulations(ork)
    if count is not None and not 0 <= simulation_index < max(count, 1):
        raise IndexError(
            f"simulation_index {simulation_index} out of range: {ork} defines {count}"
        )
    self._cfg = {
        "ork_file": ork,
        "jar_path": jar,
        "simulation_index": simulation_index,
        "log_level": log_level,
        "jvm_path": os.fspath(jvm_path) if jvm_path is not None else None,
        "jvm_args": tuple(jvm_args),
        "worker_stdout": worker_stdout,
        "_test_init": _test_init,
    }
    self._max_workers = max_workers
    self._max_tasks_per_child = max_tasks_per_child
    self._executor: ProcessPoolExecutor | None = None
    self._broken = False

run(tasks, *, worker_fn=None, seed=None, progress=None, on_error='collect')

Runs one task per simulation and collects the results.

tasks: an iterable of mappings (DECLARATIVE_KEYS names, plus an optional reserved 'seed'), or an int meaning that many runs of the document as-is. worker_fn(helper, sim, task): an importable, module-level callable taking over the whole task body — set options yourself (everything you vary), run the simulation, return plain-Python data. seed: study seed for reproducible seed derivation; default draws from SystemRandom. progress(done, total): called with (0, total) up front — JVM boot takes seconds and bars should render — then once per completion; wrap pool.run(...)'s completions with tqdm by passing progress=lambda done, total: bar.update(...) or iterate to_records afterwards. on_error: 'collect' (failures are SimError data) or 'abort' (first failure cancels the study).

Raises:

Type	Description
StudyAborted	interrupt / worker-crash / worker-init / task-error (with .partial holding everything collected).

Source code in src/orlab/parallel.py

def run(
    self,
    tasks,
    *,
    worker_fn=None,
    seed=None,
    progress=None,
    on_error: str = "collect",
) -> StudyResult:
    """Runs one task per simulation and collects the results.

    tasks: an iterable of mappings (DECLARATIVE_KEYS names, plus an
    optional reserved 'seed'), or an int meaning that many runs of the
    document as-is. worker_fn(helper, sim, task): an importable,
    module-level callable taking over the whole task body — set options
    yourself (everything you vary), run the simulation, return
    plain-Python data. seed: study seed for reproducible seed
    derivation; default draws from SystemRandom. progress(done, total):
    called with (0, total) up front — JVM boot takes seconds and bars
    should render — then once per completion; wrap
    ``pool.run(...)``'s completions with tqdm by passing
    ``progress=lambda done, total: bar.update(...)`` or iterate
    to_records afterwards. on_error: 'collect' (failures are SimError
    data) or 'abort' (first failure cancels the study).

    :raises StudyAborted: interrupt / worker-crash / worker-init /
        task-error (with ``.partial`` holding everything collected).
    """
    if on_error not in ("collect", "abort"):
        raise ValueError("on_error must be 'collect' or 'abort'")
    if isinstance(tasks, bool):
        raise TypeError("tasks must be an iterable of mappings or an int")
    if isinstance(tasks, int):
        if tasks < 0:
            raise ValueError(f"cannot run {tasks} simulations")
        task_list: list[dict] = [{} for _ in range(tasks)]
    else:
        try:
            task_list = [dict(t) for t in tasks]
        except (TypeError, ValueError) as e:
            raise TypeError(
                "tasks must be an iterable of mappings or an int "
                f"(got {type(tasks).__name__}); did you pass a single "
                "mapping instead of a list?"
            ) from e
    if worker_fn is None:
        self._validate_declarative(task_list)
    else:
        for task in task_list:
            if "seed" in task:
                self._validate_declarative([{"seed": task["seed"]}])
        self._validate_worker_fn(worker_fn)
        try:
            pickle.dumps(task_list)
        except Exception as e:
            raise ValueError(
                f"tasks are not picklable ({e}); worker_fn tasks must "
                "hold plain-Python data only"
            ) from e
    if not task_list:
        return StudyResult((), ())
    seeds = self._derive_seeds(task_list, seed)

    results: list[SimResult] = []
    errors: list[SimError] = []
    total = len(task_list)

    def partial() -> StudyResult:
        return StudyResult(
            tuple(sorted(results, key=lambda r: r.index)),
            tuple(sorted(errors, key=lambda e: e.index)),
        )

    def kill_pool() -> None:
        self._broken = True
        if self._executor is not None:
            self._executor.shutdown(wait=False, cancel_futures=True)
            self._executor = None

    futures: list = []
    try:
        executor = self._ensure_executor(total)
        for index, task in enumerate(task_list):
            futures.append(executor.submit(_worker_run, index, task, seeds[index], worker_fn))
        if progress is not None:
            try:
                progress(0, total)
            except BaseException:
                for f in futures:
                    f.cancel()
                raise
        for done, future in enumerate(as_completed(futures), start=1):
            outcome = future.result()
            kind = outcome[0]
            if kind == "init_error":
                for f in futures:
                    f.cancel()
                kill_pool()  # init config is identical across workers:
                # every worker holds an unusable JVM — release them
                raise StudyAborted(
                    "worker-init",
                    partial(),
                    f"a worker failed to initialize:\n{outcome[2]}",
                )
            index = outcome[1]
            if kind == "ok":
                _, _, payload, actual_seed, reassigned = outcome
                results.append(
                    SimResult(index, task_list[index], actual_seed, reassigned, payload)
                )
            else:
                _, _, error_type, message, tb = outcome
                errors.append(
                    SimError(index, task_list[index], seeds[index], error_type, message, tb)
                )
                if on_error == "abort":
                    for f in futures:
                        f.cancel()
                    raise StudyAborted(
                        "task-error",
                        partial(),
                        f"task {index} failed ({error_type}: {message}) and on_error='abort'",
                    )
            if progress is not None:
                try:
                    progress(done, total)
                except BaseException:
                    for f in futures:
                        f.cancel()
                    raise
        return partial()
    except KeyboardInterrupt:
        kill_pool()
        raise StudyAborted(
            "interrupt", partial(), "interrupted; partial results preserved"
        ) from None
    except BrokenProcessPool as e:
        kill_pool()
        raise StudyAborted(
            "worker-crash",
            partial(),
            "a worker process died (JVM crash or out-of-memory?) — "
            "consider jvm_args=('-Xmx512m',) per worker, fewer workers, "
            "or max_tasks_per_child",
        ) from e

shutdown()

Stops the workers (their JVMs end with their processes). The pool cannot be reused afterwards.

Source code in src/orlab/parallel.py

def shutdown(self) -> None:
    """Stops the workers (their JVMs end with their processes). The pool
    cannot be reused afterwards."""
    self._broken = True
    if self._executor is not None:
        self._executor.shutdown(wait=True, cancel_futures=True)
        self._executor = None

Dispersion listeners

Canned dispersion listeners.

Both listeners hold only plain-Python/numpy state, so they pickle across :class:~orlab.SimulationPool's process boundary and survive OpenRocket's listener cloning. The clone semantics matter for extenders: OpenRocket runs a copy of your listener, so instance attributes mutated inside hooks are invisible to the caller afterwards — these two are read-only by design; subclasses must not accumulate results on self (share a list instead, as the listeners guide shows).

WindProfile

Bases: AbstractSimulationListener

Deterministic altitude-dependent wind, replacing OpenRocket's wind model including its turbulence — the cross-version mechanism for layered-wind studies (the only native alternative is 24.12-only).

altitudes_m must be strictly increasing; speeds_ms are wind speeds at those altitudes; directions_rad (scalar or per-point) is the direction the wind blows from, matching setWindDirection (0 = from north, π/2 = from east — verified: both produce identical drift). Between points, the wind vector is interpolated component-wise (no 359°→1° wraparound artifacts; under strong direction shear the interpolated magnitude dips below the endpoint speeds — inherent to vector interpolation). Outside the range, the end values hold. A sharp layer needs an epsilon step: altitudes_m=[0, 20, 20.001, 30].

Source code in src/orlab/listeners.py

class WindProfile(AbstractSimulationListener):
    """Deterministic altitude-dependent wind, replacing OpenRocket's wind
    model **including its turbulence** — the cross-version mechanism for
    layered-wind studies (the only native alternative is 24.12-only).

    ``altitudes_m`` must be strictly increasing; ``speeds_ms`` are wind
    speeds at those altitudes; ``directions_rad`` (scalar or per-point) is
    the direction the wind blows *from*, matching
    ``setWindDirection`` (0 = from north, π/2 = from east — verified: both
    produce identical drift). Between points, the wind *vector* is
    interpolated component-wise (no 359°→1° wraparound artifacts; under
    strong direction shear the interpolated magnitude dips below the
    endpoint speeds — inherent to vector interpolation). Outside the range,
    the end values hold. A sharp layer needs an epsilon step:
    ``altitudes_m=[0, 20, 20.001, 30]``.
    """

    def __init__(self, altitudes_m, speeds_ms, directions_rad=0.0):
        altitudes = np.asarray(altitudes_m, dtype=float)
        speeds = np.asarray(speeds_ms, dtype=float)
        directions = np.asarray(directions_rad, dtype=float)
        if directions.ndim == 0:
            directions = np.full(len(altitudes), float(directions))
        if len(altitudes) != len(speeds) or len(altitudes) != len(directions):
            raise ValueError("altitudes_m, speeds_ms, directions_rad lengths differ")
        if len(altitudes) == 0:
            raise ValueError("at least one profile point is required")
        if not (
            np.all(np.isfinite(altitudes))
            and np.all(np.isfinite(speeds))
            and np.all(np.isfinite(directions))
        ):
            raise ValueError("profile values must be finite")
        if np.any(np.diff(altitudes) <= 0):
            raise ValueError("altitudes_m must be strictly increasing")
        if np.any(speeds < 0):
            raise ValueError("wind speeds cannot be negative")
        self.altitudes = altitudes
        # OpenRocket's from-vector convention (see _wind_at)
        self.u = speeds * np.sin(directions)
        self.v = speeds * np.cos(directions)

    def _wind_at(self, altitude_m: float) -> tuple[float, float]:
        """The interpolated (u, v) Coordinate components at an altitude —
        pure numpy, unit-testable without a JVM.

        Convention note for extenders: OpenRocket's wind Coordinate points
        toward where the wind blows *from* (the stepper ADDS it to rocket
        velocity to form airspeed) — u, v here are from-vector components,
        the sign-inverse of physical airflow components. Wind from the east
        is (+speed, 0), and the rocket drifts west. This reproduces
        OpenRocket's own model exactly; weather-data u/v (eastward/northward
        FLOW) must be negated before use in a custom hook."""
        u = float(np.interp(altitude_m, self.altitudes, self.u))
        v = float(np.interp(altitude_m, self.altitudes, self.v))
        return u, v

    def postWindModel(self, status, wind):
        altitude = float(status.getRocketPosition().z)
        u, v = self._wind_at(altitude)
        return active_core_root().util.Coordinate(u, v, 0.0)

ThrustFactor

Bases: AbstractSimulationListener

Multiplies every thrust sample by a constant factor — the classic motor-variation dispersion knob (thrust curves vary batch to batch).

Source code in src/orlab/listeners.py

class ThrustFactor(AbstractSimulationListener):
    """Multiplies every thrust sample by a constant factor — the classic
    motor-variation dispersion knob (thrust curves vary batch to batch).
    """

    def __init__(self, factor: float):
        factor = float(factor)
        if not math.isfinite(factor) or factor <= 0:
            raise ValueError(f"thrust factor must be positive and finite, got {factor}")
        self.factor = factor

    def postSimpleThrustCalculation(self, status, thrust):
        return float(thrust) * self.factor

Errors

orlab exception types.

OrlabError

Bases: Exception

Base class for all orlab errors.

Source code in src/orlab/errors.py

class OrlabError(Exception):
    """Base class for all orlab errors."""

JarVerificationError

Bases: OrlabError

A jar's sha256 could not be verified (no pin, or a digest mismatch).

Source code in src/orlab/errors.py

class JarVerificationError(OrlabError):
    """A jar's sha256 could not be verified (no pin, or a digest mismatch)."""

NotAnOpenRocketJar

Bases: OrlabError

The jar path does not point to a readable OpenRocket jar.

Source code in src/orlab/errors.py

class NotAnOpenRocketJar(OrlabError):
    """The jar path does not point to a readable OpenRocket jar."""

UnsupportedFlightDataType

Bases: OrlabError

A FlightDataType constant does not exist in the loaded OpenRocket version.

Source code in src/orlab/errors.py

class UnsupportedFlightDataType(OrlabError):
    """A FlightDataType constant does not exist in the loaded OpenRocket version."""

UnsupportedOpenRocketVersion

Bases: OrlabError

The OpenRocket jar's version is older than any known profile.

Source code in src/orlab/errors.py

class UnsupportedOpenRocketVersion(OrlabError):
    """The OpenRocket jar's version is older than any known profile."""

StudyAborted

Bases: OrlabError

A SimulationPool study ended early. reason is one of interrupt, worker-crash, worker-init, task-error; partial holds every result collected before the abort.

Source code in src/orlab/errors.py

class StudyAborted(OrlabError):
    """A SimulationPool study ended early. ``reason`` is one of
    ``interrupt``, ``worker-crash``, ``worker-init``, ``task-error``;
    ``partial`` holds every result collected before the abort."""

    def __init__(self, reason: str, partial: "StudyResult", message: str):
        super().__init__(f"study aborted ({reason}): {message}")
        self.reason = reason
        self.partial = partial
        self._message = message

    def __reduce__(self):
        return (StudyAborted, (self.reason, self.partial, self._message))