koios-component-builder 1.1.1

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.

Pin Layout

By default pins are laid out on the canvas in the order they appear in your class body. For control-systems blocks where grouping matters — setpoints together, tuning constants together, status outputs apart — you can declare an explicit arrangement in Meta and insert gaps between pins. The arrangement ships in the .kcl manifest so every instance of the component starts with the same visual default. Users can further tweak the layout per-instance from the Koios UI.

from koios_component_builder import (
    Component,
    ComponentCategory,
    ComponentIcon,
    Gap,
    Input,
    NumberConfig,
    Output,
)


class PIDController(Component):
    """Discrete PID controller with anti-windup."""

    class Meta:
        icon = ComponentIcon.GAUGE
        category = ComponentCategory.CONTROL
        canvas_width = 8

        # Group setpoint + process variable, give the tuning constants their
        # own visual block, separate the enable/reset inputs at the bottom,
        # and keep the diagnostic outputs distinct from the control output.
        inputs_layout = [
            "setpoint",
            "process_variable",
            Gap(),
            "kp",
            "ki",
            "kd",
            Gap(size=2),
            "enable",
            "reset",
        ]
        outputs_layout = [
            "control_output",
            Gap(),
            "saturated",
            "integral",
        ]

    setpoint: Input[float] = Input(default=0.0, description="Target value")
    process_variable: Input[float] = Input(default=0.0, description="Measured value")
    kp: Input[float] = Input(default=1.0, description="Proportional gain")
    ki: Input[float] = Input(default=0.0, description="Integral gain")
    kd: Input[float] = Input(default=0.0, description="Derivative gain")
    enable: Input[bool] = Input(default=True, description="Enable control")
    reset: Input[bool] = Input(default=False, description="Reset integrator")

    output_min: NumberConfig = NumberConfig(default=-100.0, description="Output lower bound")
    output_max: NumberConfig = NumberConfig(default=100.0, description="Output upper bound")

    control_output: Output[float] = Output(default=0.0, description="Manipulated variable")
    saturated: Output[bool] = Output(default=False, description="Output is at a limit")
    integral: Output[float] = Output(default=0.0, description="Current integrator state")

    def execute(self) -> None:
        ...  # PID math

Gap(size=1) — Insert a vertical spacer between pins. size is in grid units (1 row = the height of one pin); defaults to 1.

Validation — Strings in inputs_layout / outputs_layout are checked against your declared Input / Output / HistoryInput fields at class-creation time. A typo raises ComponentDefinitionError on import, not at runtime.

Partial layouts — Pins declared on the class but absent from the layout list are appended in declaration order, so you can add new pins to a component without touching the layout.

Empty layouts — If neither list is declared, pins use class-body order with no gaps — the same behavior components had before this feature.

Note on Input(order=...) — The old per-field order= parameter is deprecated for Input, Output, and HistoryInput. Use Meta.inputs_layout / Meta.outputs_layout instead — it gives finer control (gaps, explicit grouping) and the contract is clearer to read at the top of the class. The order= parameter on Config family fields is unaffected.

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

Security Audit

Every .kcl export automatically runs a static security analysis on your source code. The audit classifies patterns into three tiers:

Tier	Meaning	Effect
✅ ALLOW	Safe — no friction	Not reported
⚠️ REVIEW	Flagged for attention, but build continues	Logged as warning, yellow badge in Koios UI
❌ DENY	Blocked	Export fails unless --allow-unsafe is used

What gets flagged

DENY — patterns that have no legitimate use in a component:

• Dangerous imports: os, subprocess, socket, threading, pickle, ctypes, sys, etc.
• Unsafe builtins: eval(), exec(), compile(), __import__()
• Sandbox escape patterns: __subclasses__, __builtins__, __code__, __globals__

REVIEW — patterns that are often legitimate but worth being aware of:

• Filesystem imports: pathlib, io, csv (common for loading AI model files in setup())
• open() calls (useful for reading model weights or config files)
• Unknown third-party packages not in the platform allow-list

Audit command

Run the audit without building a .kcl:

# Audit and print results
koios-component-builder audit my_library/

# Output as JSON (for CI integration)
koios-component-builder audit my_library/ --json

# Audit with a custom policy
koios-component-builder audit my_library/ --policy security_policy.yaml

Exit code is 1 if any DENY findings are present, 0 otherwise — suitable for CI pipelines.

Custom policies

Override the defaults by providing a YAML policy file:

# security_policy.yaml
policy_version: "1.0"

imports:
  allow:
    - my_internal_sdk      # Trust your own internal package
    - requests             # Allow HTTP if your components genuinely need it
  deny:
    - pandas               # Tighten if you want to forbid heavy deps

builtins:
  review:
    - open                 # Already REVIEW by default; shown here for reference

attributes:
  deny:
    - __dict__             # Add to deny if you want stricter introspection rules

Policy overrides work bidirectionally — you can move items to a less restrictive tier (e.g. DENY → ALLOW) or a more restrictive one (e.g. ALLOW → DENY). Items are removed from conflicting tiers automatically.

koios-component-builder export my_library/ --policy security_policy.yaml

Overriding for power users

If your library has a legitimate reason for a flagged pattern (e.g. loading an ONNX model from disk in setup()), the recommended approach is to use pathlib / io which are REVIEW-tier rather than DENY. For edge cases that genuinely need a denied import:

koios-component-builder export my_library/ --allow-unsafe

The .kcl is still created, but the security_audit in manifest.json records passed: false — the Koios UI will display a warning badge when the library is uploaded.

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)
  --allow-unsafe           Create .kcl even with DENY findings
  --policy PATH            Custom security policy YAML file

# Audit a library without building
koios-component-builder audit <source_path> [OPTIONS]

Options:
  --policy PATH            Custom security policy YAML file
  --json                   Output results as JSON

# 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.