from __future__ import annotations
from pathlib import Path
from typing import Any, Literal, Optional
from pydantic import Field
from kelvin.config.common import (
AppBaseConfig,
AppTypes,
ConfigBaseModel,
CustomActionsIO,
read_schema_file,
resolve_schema_path,
)
from .manifest import (
AppDefaults,
AppManifest,
CustomActionWay,
DefaultsDefinition,
DynamicIODefinition,
DynamicIoOwnership,
DynamicIoType,
Flags,
IOSchema,
ManifCustomAction,
RuntimeUpdateFlags,
SchemasDefinition,
)
# =============================================================================
# Configuration Models
# =============================================================================
[docs]
class RuntimeUpdateConfig(ConfigBaseModel):
"""Controls which parts of the exporter can be updated at runtime."""
configuration: bool = False
[docs]
class ExporterFlags(ConfigBaseModel):
"""Feature flags for exporter behavior."""
enable_runtime_update: RuntimeUpdateConfig = RuntimeUpdateConfig()
resources_required: Optional[bool] = None
[docs]
class SchemasConfig(ConfigBaseModel):
"""Paths to JSON schema files for configuration validation."""
configuration: Optional[str] = None
io_configuration: dict[str, str] = Field(default_factory=dict) # Maps IO names to schema file paths
[docs]
class ExporterIO(ConfigBaseModel):
"""Defines a dynamic IO channel for the exporter."""
name: str
data_types: list[str] = Field(default_factory=lambda: ["number", "string", "boolean"])
[docs]
class DeploymentDefaults(ConfigBaseModel):
"""Default values applied when deploying the exporter."""
system: dict[str, Any] = Field(default_factory=dict)
configuration: dict[str, Any] = Field(default_factory=dict)
[docs]
class ExporterConfig(AppBaseConfig):
"""Complete configuration for an exporter application."""
type: Literal[AppTypes.exporter] # pyright: ignore[reportIncompatibleVariableOverride]
spec_version: str = "5.0.0"
flags: ExporterFlags = ExporterFlags()
exporter_io: list[ExporterIO] = Field(default_factory=list)
ui_schemas: SchemasConfig = SchemasConfig()
defaults: DeploymentDefaults = DeploymentDefaults()
custom_actions: CustomActionsIO = CustomActionsIO()
api_permissions: Optional[list[str]] = None
[docs]
def to_manifest(self, workdir: Path, read_schemas: bool) -> AppManifest:
return convert_exporter_to_manifest(self, workdir=workdir, read_schemas=read_schemas)
# =============================================================================
# Manifest Conversion
# =============================================================================
def _build_schemas(config: ExporterConfig, workdir: Path, read_schemas: bool) -> SchemasDefinition:
"""Build schema definitions, optionally reading schema files from disk."""
schemas = SchemasDefinition()
if not read_schemas:
return schemas
# Load main configuration schema
if config.ui_schemas.configuration:
schema_path = resolve_schema_path(workdir, config.ui_schemas.configuration)
schemas.configuration = read_schema_file(schema_path)
# Load per-IO configuration schemas
schemas.io_configurations = [
IOSchema(
type_name=io_name,
schema=read_schema_file(resolve_schema_path(workdir, schema_path)) if schema_path else {},
)
for io_name, schema_path in config.ui_schemas.io_configuration.items()
]
return schemas
def _build_custom_actions(config: ExporterConfig) -> list[ManifCustomAction]:
"""Convert input/output custom actions to manifest format."""
input_actions = [
ManifCustomAction(type=action.type, way=CustomActionWay.input_ca) for action in config.custom_actions.inputs
]
output_actions = [
ManifCustomAction(type=action.type, way=CustomActionWay.output_ca) for action in config.custom_actions.outputs
]
return input_actions + output_actions
def _build_defaults(config: ExporterConfig) -> DefaultsDefinition | None:
"""Build defaults definition if any defaults were explicitly set."""
fields_set = config.defaults.model_fields_set
has_configuration = "configuration" in fields_set
has_system = "system" in fields_set
has_api_permissions = config.api_permissions is not None
if not (has_configuration or has_system or has_api_permissions):
return None
defaults = DefaultsDefinition()
if has_configuration:
defaults.app = AppDefaults(configuration=config.defaults.configuration)
if has_system:
defaults.system = config.defaults.system
if has_api_permissions:
defaults.api_permissions = config.api_permissions
return defaults
def _build_flags(config: ExporterConfig) -> Flags:
"""Build manifest flags from exporter config."""
return Flags(
spec_version=config.spec_version,
enable_runtime_update=RuntimeUpdateFlags(
configuration=config.flags.enable_runtime_update.configuration,
),
resources_required=config.flags.resources_required,
)
def _build_dynamic_io(config: ExporterConfig) -> list[DynamicIODefinition]:
"""Convert exporter IO definitions to dynamic IO manifest format."""
return [
DynamicIODefinition(
type_name=io.name,
data_types=io.data_types,
ownership=DynamicIoOwnership.remote,
type=DynamicIoType.data,
)
for io in config.exporter_io
]
[docs]
def convert_exporter_to_manifest(config: ExporterConfig, workdir: Path, read_schemas: bool) -> AppManifest:
"""Convert an ExporterConfig to its AppManifest representation."""
return AppManifest(
name=config.name,
title=config.title,
description=config.description,
type=config.type,
version=config.version,
category=config.category,
flags=_build_flags(config),
dynamic_io=_build_dynamic_io(config),
schemas=_build_schemas(config, workdir, read_schemas),
defaults=_build_defaults(config),
custom_actions=_build_custom_actions(config),
)