koios-component-builder 1.0.0rc2

Simplified component builder for the Koios platform

Koios Component Builder

SDK and CLI for building component libraries for the Koios platform. Components are reusable logic blocks with typed inputs and outputs that get wired together and executed by the Koios Component Engine at configurable scan rates.

Installation

pip install koios-component-builder

Requires Python 3.12+.

Quick Start

1. Write a component

# my_library/math_ops.py
from koios_component_builder import (
    Component,
    ComponentCategory,
    ComponentIcon,
    Input,
    Output,
)


class SimpleAdder(Component):
    """A component that adds two numbers."""

    class Meta:
        icon = ComponentIcon.SUM
        category = ComponentCategory.MATH

    a: Input[float] = Input(default=0.0, description="First number")
    b: Input[float] = Input(default=0.0, description="Second number")
    result: Output[float] = Output(default=0.0, description="Sum of a and b")

    def execute(self) -> None:
        self.result = self.a + self.b

2. Define a library

# my_library/__init__.py
from koios_component_builder import ComponentLibrary
from .math_ops import SimpleAdder, Multiplier


class MyLibrary(ComponentLibrary):
    """My custom component library."""

    name = "my-library"
    major = 1
    minor = 0
    patch = 0
    description = "Custom math components"

    components = [SimpleAdder, Multiplier]


__all__ = ["MyLibrary", "SimpleAdder", "Multiplier"]

3. Export

koios-component-builder export my_library/

This creates a .kcl (Koios Component Library) package in dist/ that can be uploaded to Koios.

Component Lifecycle

When deployed to Koios, each execution cycle:

1. Field Injection — Input and config values are set on the component
2. Setup — setup() runs once before the first execute() (if overridden)
3. Execute — execute() runs with current input values
4. Output Propagation — Output values are sent to wired destinations

Your component implements setup() (optional) and execute() — the engine handles all wiring and data flow.

setup() — One-Time Initialization

Override setup() for expensive work that should only happen once, such as loading models, creating registries, or parsing configuration. All field values are available when setup() runs.

If setup() raises an exception, the component is marked FAILED and retried on the next cycle.

class UnitConverter(Component):
    """Converts Fahrenheit to Celsius using pint."""

    class Meta:
        icon = ComponentIcon.TRANSFORM
        category = ComponentCategory.TRANSFORM

    value: Input[float] = Input(default=0.0, description="Temperature in °F")
    decimals: NumberConfig = NumberConfig(
        default=2, min_value=0, max_value=6, description="Decimal places"
    )
    result: Output[float] = Output(default=0.0, description="Temperature in °C")

    def setup(self) -> None:
        from pint import UnitRegistry
        self._ureg = UnitRegistry()  # ~80ms — only runs once

    def execute(self) -> None:
        temp = self._ureg.Quantity(self.value, self._ureg.degF)
        self.result = round(temp.to(self._ureg.degC).magnitude, int(self.decimals))

For resources shared across all instances of the same component class, use a class-level guard instead:

class MyComponent(Component):
    def execute(self) -> None:
        if not hasattr(MyComponent, "_shared_model"):
            MyComponent._shared_model = load_model()
        self.result = MyComponent._shared_model.predict(self.input)

Field Types

Input / Output

from koios_component_builder import Component, Input, Output

class ExampleComponent(Component):
    # Supported types: float, int, bool, str, list, dict
    temperature: Input[float] = Input(default=0.0, description="Temperature in Celsius")
    count: Input[int] = Input(default=0, description="Item count")
    enabled: Input[bool] = Input(default=True, description="Enable processing")
    mode: Input[str] = Input(default="auto", description="Operating mode")

    alarm: Output[bool] = Output(default=False, description="High temperature alarm")
    status: Output[str] = Output(default="ok", description="Current status")

    def execute(self) -> None:
        if self.enabled and self.temperature > 100:
            self.alarm = True
            self.status = "overtemp"
        else:
            self.alarm = False
            self.status = "ok"

Config Fields

Config fields are set when the component instance is created in the Koios UI and remain constant during execution. They appear as configuration controls on the component node.

from koios_component_builder import (
    Component, Input, Output,
    NumberConfig, StringConfig, ChoiceConfig, BoolConfig,
)

class ConfigurableComponent(Component):
    # Numeric with min/max validation
    threshold: NumberConfig = NumberConfig(
        default=75.0, min_value=0.0, max_value=100.0,
        description="Alert threshold"
    )

    # Dropdown with predefined options
    mode: ChoiceConfig = ChoiceConfig(
        default="average", choices=["average", "median", "max"],
        description="Calculation mode"
    )

    # Text with optional regex validation
    label: StringConfig = StringConfig(
        default="Sensor", description="Display label"
    )

    # Boolean toggle
    verbose: BoolConfig = BoolConfig(
        default=False, description="Enable verbose output"
    )

    value: Input[float] = Input(default=0.0)
    alert: Output[bool] = Output(default=False)

    def execute(self) -> None:
        self.alert = self.value > self.threshold

HistoryInput

HistoryInput fields provide access to historical tag data via InfluxDB. They must be wired to a HISTORY connector on the environment canvas.

from koios_component_builder import Component, HistoryInput, Output

class TrendAnalyzer(Component):
    """Calculates trend from historical data."""

    sensor_history: HistoryInput = HistoryInput(
        description="Historical sensor readings"
    )
    trend: Output[float] = Output(default=0.0, description="Trend slope")

    def execute(self) -> None:
        if self.sensor_history is None:
            return  # Not wired to a history connector

        # Fetch last hour of data, max 200 samples
        df = self.sensor_history.get_history(
            period_seconds=3600,
            num_samples=200,
        )
        # df has columns: timestamp, value
        if not df.empty:
            values = df["value"].tolist()
            self.trend = values[-1] - values[0]

Component Metadata

Customize how components appear in the Koios UI:

class MyComponent(Component):
    """Component description shown in the UI."""

    class Meta:
        icon = ComponentIcon.CHART_LINE       # Tabler icon name
        category = ComponentCategory.ANALYSIS  # UI grouping
        canvas_width = 8                       # Node width (4–15 grid units, default: 6)
        canvas_minimal = False                 # Compact mode (no header/footer)

        # Optional: component-specific version (overrides library version)
        major = 2
        minor = 1
        patch = 0
        prerelease = "beta"

Icons — Any Tabler icon name in kebab-case. Common constants: SUM, CALCULATOR, CHART_LINE, GAUGE, THERMOMETER, FILTER, WAVE_SINE, TOGGLE_LEFT, ALERT_TRIANGLE, TRANSFORM.

Categories — Standard constants: MATH, STATISTICS, LOGIC, ANALYSIS, TRANSFORM, FILTER, CONTROL, MONITORING. Custom strings are also accepted.

Dependencies

Libraries can declare third-party Python package dependencies. The builder resolves them against the Koios platform manifest to determine what's pre-installed in the container vs. what needs bundling.

class MyProtocolLibrary(ComponentLibrary):
    name = "my-protocol-library"
    major = 1
    minor = 0
    dependencies = ["crcmod", "minimalmodbus>=2.0"]
    components = [MyDevice]

Three tiers:

Tier	Description	Example
Platform	Pre-installed in the Koios container	numpy, pandas, scipy
Bundled	Downloaded and included in the .kcl	crcmod, pint
SDK	Always available (koios-component-builder itself)	pydantic, click

# Export without bundling (warns about non-platform deps)
koios-component-builder export my_library/

# Bundle non-platform dependencies into the .kcl
koios-component-builder export my_library/ --include-deps

# Target specific platforms
koios-component-builder export my_library/ --include-deps \
    --platform manylinux2014_x86_64 --platform manylinux2014_aarch64

# List available platform packages
koios-component-builder platform-packages

Package Format

The export command creates a .kcl (Koios Component Library) package — a ZIP archive:

my-library-1.0.0.kcl
├── manifest.json                              # Library metadata
├── my_library-1.0.0-py3-none-any.whl         # Python wheel
└── deps/                                      # Bundled dependency wheels (if any)
    ├── crcmod-1.7-cp312-...-x86_64.whl
    └── crcmod-1.7-cp312-...-aarch64.whl

CLI Reference

# Export a library to a .kcl package
koios-component-builder export <source_path> [OPTIONS]

Options:
  -o, --output PATH        Output directory (default: dist/)
  --include-deps           Bundle non-platform dependencies
  --platform TEXT          Target platform tag (repeatable)

# List pre-installed platform packages
koios-component-builder platform-packages

Local Testing

Test components locally before deploying:

from my_library.math_ops import SimpleAdder

adder = SimpleAdder("test-instance")
adder.a = 5.0
adder.b = 3.0
adder.execute()

print(f"Result: {adder.result}")  # Result: 8.0

License

Copyright Ai-OPs, Inc. All rights reserved.