What is a Custom Orochi Module?

A custom Orochi module does not need to be written completely from scratch. In fact, one of the primary design strengths of the Orochi architecture is wrapping existing, battle-tested scientific Python packages into standardized Orochi heads.

For example:

• ``orochi.fit``: Wraps LMFIT and Optuna, turning external numerical optimization and Bayesian sampling libraries into seamless Orochi components that directly inspect, bound, and refine parameters across multiple physical modules (such as orochi.pdfsim and orochi.phase).

• ``orochi.pixia``: Bridges 2D image processing libraries and detector integration algorithms into the Orochi pipeline.

By subclassing OrochiBase, existing libraries instantly acquire:

1. Unified Parameter Control: Parameters, search boundaries, and active refinement flags (fit, with backward-compatible alias optimize) are stored in standardized dictionaries accessible via flat string paths (e.g., @phase_0/a/val).

2. Dependency-Aware Caching: Expensive external calls or simulations are automatically skipped when inputs have not changed.

3. Pipeline & UI Interoperability: Automatic integration into multi-step execution graphs and automated Gradio web UI generation via param_boxes.

Example: Wrapping an External Library (The orochi.fit LMFIT Pattern)

Below is an illustrative example demonstrating how an external library (such as LMFIT) is wrapped into an Orochi module:

from orochi.base.core import OrochiBase
import lmfit

class CustomLmfitAdapter(OrochiBase):
    """Wraps LMFIT to optimize parameters across attached Orochi simulators."""

    def __init__(self):
        super().__init__()
        # 1. Define optimizer hyperparameters
        self["method"] = "leastsq"
        self["max_nfev"] = 1000
        self["ftol"] = 1e-5
        self.simulators = []

    def build_lmfit_params(self) -> lmfit.Parameters:
        """Translate Orochi parameters marked for optimization into lmfit Parameters."""
        params = lmfit.Parameters()
        for sim in self.simulators:
            for key in sim.keys():
                if sim.get((key, "fit"), False):
                    pname = f"{sim.id}_{key}"
                    val = sim[key]
                    pmin = sim.get((key, "min"), -float("inf"))
                    pmax = sim.get((key, "max"), float("inf"))
                    params.add(pname, value=val, min=pmin, max=pmax, vary=True)
        return params

    def objective(self, params: lmfit.Parameters, y_obs):
        """Objective function evaluated at each optimization iteration."""
        # Update Orochi simulators with trial values
        for sim in self.simulators:
            for key in sim.keys():
                pname = f"{sim.id}_{key}"
                if pname in params:
                    sim[key] = params[pname].value
            # Run simulator (benefits from Orochi's internal caching!)
            sim()

        y_calc = sum(sim.y for sim in self.simulators)
        return y_obs - y_calc

    def __call__(self, y_obs, force: bool = False):
        """Primary execution: run the wrapped external optimizer."""
        if force:
            self._state = "INIT"
        lmfit_params = self.build_lmfit_params()
        result = lmfit.minimize(self.objective, lmfit_params, args=(y_obs,), method=self["method"])
        self.result = result
        self._state = "DONE"
        return result

    def run(self, y_obs, force: bool = False):
        """Explicit execution entry point for manual scripts."""
        return self.__call__(y_obs, force=force)

Standalone Custom Head Template (From Scratch)

For specialized physics models or custom data processing created from scratch, subclass OrochiBase and register method dependencies for selective caching:

from orochi.base.core import OrochiBase

class CustomScatteringHead(OrochiBase):
    """Custom Orochi calculation head."""

    def __init__(self):
        super().__init__()
        # 1. Define module-specific parameters
        self["wavelength"] = 0.18033
        self["attenuation"] = 1.0
        self["attenuation", "min"] = 0.0
        self["attenuation", "max"] = 5.0
        self["attenuation", "fit"] = True

        # 2. Register method dependencies for automatic caching
        # When self["wavelength"] changes, 'calc_geometry' will re-execute;
        # otherwise it will be skipped during repeated calls.
        self._method_dependencies["calc_geometry"] = {
            "self": ["wavelength"]
        }

    def calc_geometry(self):
        """Heavy calculation step executed only when dependent parameters change."""
        # Expensive calculation here...
        pass

    def calc_output(self):
        """Fast calculation step."""
        pass

    def __call__(self, force: bool = False):
        """Primary execution entry point."""
        if force:
            self._state = "INIT"
        super().__call__(force=force)
        self.calc_geometry()
        self.calc_output()
        self._state = "DONE"

    def run(self, force: bool = False):
        """Explicit execution entry point for manual scripts."""
        return self.__call__(force=force)

    def explain(self) -> str:
        """Structured Markdown diagnostic report for AI agents and users."""
        return f"### Custom Module Status\\n- State: {self._state}\\n- Wavelength: {self['wavelength']}"