Writing plugins¶

First, scaffold out a new Python library:

uv init --lib minisky-example

from minisky import Plugin, PluginContext, PluginSpec


class Example:
    pass


def build(context: PluginContext) -> PluginSpec:
    context.mount(Example())  # (2)!
    return context.finish()


plugin = Plugin(build=build)  # (1)!

1. Plugin describes how the package creates its plugin. Its build function is called each time a MiniSky runtime loads the plugin.
2. context.mount(...) attaches an object to the runtime. Note that here, we create the Example() object inside the build function so multiple minisky runtimes can have their own independent state.

Register the plugin¶

Now that the plugin is defined, we need to advertise its location to minisky. Define a new entry point in your pyproject.toml:

[project.entry-points."minisky.plugins"]
example = "minisky_example:plugin"

The left-hand side, example, is the plugin ID, used in the PLUGIN LOAD <plugin_id> stack command and the user configuration TOML.

The right-hand side should point to the Python object we defined above.

Configuration¶

minisky's configuration TOML supports defining plugin-specific configuration.

Suppose you want to accept the following in the user TOML:

[plugins.example]
message = "hello from my plugin :)"

Define the shape with a Pydantic BaseModel:

from minisky import Plugin, PluginContext, PluginSpec
from pydantic import BaseModel, ConfigDict


class ExampleConfig(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)

    message: str


class Example:
    def __init__(self, config: ExampleConfig) -> None:
        self.config = config


def build(context: PluginContext[ExampleConfig]) -> PluginSpec:
    context.mount(Example(context.config))
    return context.finish()


plugin = Plugin(build=build, config_class=ExampleConfig)  # (1)!

1. When you specify config_class=ExampleConfig, minisky will first validate the plugin's configuration using Pydantic. The validated configuration can then be accessed in the build function via context.config.

Commands¶

To add a new stack command, use the @command decorator. As a simple example:

from minisky import Ok, Result, command


class Example:
    def __init__(self, config: ExampleConfig) -> None:
        self.config = config

    @command
    def hello(self) -> Result[str, str]:
        """Print the configured message."""
        return Ok(self.config.message)

> PLUGIN LOAD EXAMPLE
Successfully loaded plugin EXAMPLE
> HELLO
hello from my plugin

More details (the decorator, parser, argument types, validation, generated documentation) are covered in the command developer guide.

Simulation hooks¶

To run code automatically as the simulation advances, hook plugin methods into the simulation loop with minisky.hook.

For example, to run a method once during every normal update:

from minisky import hook


class Example:
    @hook  # (1)!
    def update(self) -> None:
        self.nupdates += 1

1. With no argument, @hook uses the method name as the hook name. A method named update() attaches to minisky's update phase.

Or periodically:

class Example:
    @hook("update", interval=5.0)  # (1)!
    def sample(self, dt: float) -> None:
        self.elapsed += dt

1. Here, interval=5.0 runs sample() every five simulation seconds. A dt parameter receives the simulated time elapsed since the previous call.

Per-aircraft state¶

minisky stores aircraft state with the structure-of-arrays model. The \(i\)th row of each numpy array refers to an aircraft.

To manipulate these arrays, inherit from Entity and use settrafarrays() to register arrays or lists as per-aircraft state:

import numpy as np

from minisky import Entity


class Example(Entity):
    def __init__(self) -> None:
        super().__init__()
        with self.settrafarrays():  # (1)!
            self.npassengers = np.array([], dtype=int)

    def create(self, n: int = 1) -> None:
        super().create(n)  # (2)!
        self.npassengers[-n:] = 0

Here, Entity.create() overrides the base class behaviour. The arrays can grow and shrink as aircraft are added, removed or reset.

Remember to mount the entity in the Plugin.build function. The self.traffic object will then be available after attachment.

class Example(Entity):
    """Track passenger count for every aircraft in the owning runtime."""

    def __init__(self, random: Random) -> None:
        super().__init__()
        self.random = random
        self.updates = 0
        with self.settrafarrays():
            self.npassengers = np.array([])

    def create(self, n: int = 1) -> None:
        super().create(n)
        self.npassengers[-n:] = [self.random.randint(50, 250) for _ in range(n)]

    @hook(interval=5.0)
    def update(self) -> None:
        """Count periodic execution."""
        self.updates += 1

    @command(name="PASSENGERS")
    def passenger_count(self, index: AcId) -> Result[str, str]:
        """Show the number of passengers on an aircraft."""
        callsign = self.traffic.callsign[index]
        return Ok(f"Aircraft {callsign} has {int(self.npassengers[index])} passengers")

    @command(name="PASSENGERS")
    def set_passenger_count(
        self, index: AcId, count: Annotated[int, Ge(0), Le(500)]
    ) -> Result[str, str]:
        """Set the number of passengers on an aircraft."""
        callsign = self.traffic.callsign[index]
        self.npassengers[index] = count
        return Ok(f"Set {callsign} passengers to {count}")

External resources¶

In some cases, you may want to configure resources (e.g. httpx.AsyncClient or background threads) on plugin startup, and shut it down cleanly. To do so, use an async context manager, for example:

@asynccontextmanager
async def lifespan(runtime: PluginRuntime):
    bridge.start(runtime)  # (1)!
    try:
        yield  # (2)!
    finally:
        bridge.stop()  # (3)!


return context.finish(lifespan=lifespan)  # (4)!

1. This starts the external resource as the plugin starts up.
2. Execution is suspended here (plugin continues to load).
3. Run any cleanup here during plugin shutdown.
4. context.finish() attaches the lifespan to this plugin.

Here, code before yield runs during plugin startup, and cleans up after yield runs during plugin shutdown.

def build(context: PluginContext[TangramConfig]) -> PluginSpec:
    bridge = context.mount(
        TangramBridge(
            redis_url=context.config.redis_url,
            channel=context.config.channel,
            max_hz=context.config.max_hz,
        )
    )

    @asynccontextmanager
    async def lifespan(runtime: PluginRuntime) -> AsyncGenerator[None]:
        runtime.subscribe_console(bridge.capture_console)
        match bridge.start(runtime):
            case Ok(message):
                runtime.echo(message)
            case Err(message):
                runtime.echo(message)
                raise RuntimeError(message)
        try:
            yield
        finally:
            bridge.stop()

    return context.finish(lifespan=lifespan)


plugin = Plugin(build=build, config_class=TangramConfig)

Replacing a simulator component¶

Most plugins add behavior alongside minisky. However, in some cases you may want to replace an entire core implementation. For example, to override the internal autopilot implementation:

@replacement  # (1)!
class CustomAutoPilot(Autopilot):
    def update(self) -> None:
        super().update()
        self.new_variable += 1


def build(context: PluginContext) -> PluginSpec:
    return context.finish(replacements=(CustomAutoPilot,))  # (2)!

1. @replacement declares CustomAutoPilot as an alternative implementation of its supported base component.
2. Pass the class to replacements= to advertise it to the core.

Users can then select an implementation with the SELECTIMPL stack command:

SELECTIMPL AUTOPILOT CUSTOMAUTOPILOT