Coverage for core / src / sensorkit / cli / config.py: 0%
108 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-02 00:03 +0000
1# SPDX-License-Identifier: Apache-2.0
2from __future__ import annotations
4import asyncio
5import json
6import pathlib
8import asyncclick as click
9import yaml
10from pydantic import BaseModel, ValidationError
11from rich.text import Text
13import sensorkit.api as sk
14from sensorkit.backend.base import BackendError, KeyNotFound
15from sensorkit.cli.utils import console, with_kit
16from sensorkit.config.parser import (
17 ConfigError,
18 ConfigSectionUnknown,
19 ConfigVersionUnsupported,
20)
23@click.group("config")
24async def config_group():
25 """Manage unified SensorKit configuration."""
28def config_error_message(e: Exception, file: str | None) -> str | None:
29 """Describe a configuration loading failure.
31 Args:
32 e: The exception raised while loading configuration.
33 file: The configuration file the command was given, if any.
35 Returns:
36 A message to report, or None if the exception is not a loading failure.
37 """
38 match e:
39 case FileNotFoundError():
40 return f"file not found: {file}"
41 case yaml.YAMLError():
42 return f"invalid YAML in {file}: {e}"
43 case ConfigVersionUnsupported():
44 return f"unsupported config version: {e}"
45 case ConfigSectionUnknown():
46 return f"unknown config section: {e}"
47 case ConfigError():
48 return f"config processing failed: {e}"
49 case ValidationError():
50 return f"config validation failed: {e}"
51 case _:
52 return None
55async def get_changed_ekv(
56 kit: sk.SensorKit,
57 ekv: dict[str, list[BaseModel]],
58) -> dict[str, list[BaseModel]]:
59 out = {}
61 for entity_id, models in ekv.items():
62 filtered = []
64 for model in models:
65 try:
66 current = await kit.entity(entity_id).kv_get_model(type(model))
68 if current != model:
69 filtered.append(model)
70 except KeyNotFound:
71 filtered.append(model)
72 except BackendError:
73 # Other backend errors are fatal.
74 raise
75 except Exception:
76 # Custom validators can raise any exception, so we cannot only catch
77 # ValidationError here.
78 filtered.append(model)
80 if filtered:
81 out[entity_id] = filtered
83 return out
86def print_ekv(
87 ekv: dict[str, list[BaseModel]],
88 *,
89 status: dict[int, str] | None = None,
90 show_values: bool = False,
91):
92 for entity_id, models in ekv.items():
93 console.print(f"[red]{entity_id}[/red]")
95 for i, model in enumerate(models, 1):
96 bar = "├" if i < len(models) else "└"
97 stat = status[id(model)] if status is not None else ""
98 console.print(f" {bar}── [green]{type(model).__name__:16s}[/green] {stat}")
100 if show_values:
101 bar = "│" if i < len(models) else " "
103 with console.capture() as capture:
104 console.print_json(model.model_dump_json())
106 json = "\n".join(f" {bar} {line}" for line in capture.get().splitlines())
107 console.print(Text.from_ansi(json))
110@config_group.command("load")
111@click.argument("file")
112@click.option("-n", "--dry-run", is_flag=True, help="Do not write to configuration backend")
113@click.option("-f", "--force", is_flag=True, help="Write configuration even if unchanged")
114@click.option("-v", "verbose", count=True, help="Increase output verbosity")
115@with_kit
116async def config_load(kit, *, file: str | None, force: bool, dry_run: bool, verbose: int):
117 """Load a SensorKit unified configuration file."""
118 try:
119 sk.set_config_location(file)
120 config = await sk.load_config()
121 except Exception as e:
122 if (message := config_error_message(e, file)) is None:
123 raise
125 console.print(f"[bold red]ERROR: {message}[/bold red]")
126 return
128 ekv = config.entity_kv
129 status = {id(model): "🆗 Unchanged" for models in ekv.values() for model in models}
131 if not force:
132 ekv = await get_changed_ekv(kit, ekv)
134 if not dry_run:
135 update_coros = {
136 # Write global config keys.
137 # TODO: Write global config when available at the backend.
138 # Write entity config keys.
139 **{
140 kit.entity(entity_id).kv_put_model(model): (entity_id, model)
141 for entity_id, models in ekv.items()
142 for model in models
143 },
144 }
146 results = await asyncio.gather(*update_coros, return_exceptions=True)
147 status.update(
148 {
149 id(model): "✅ Updated"
150 if not isinstance(result, Exception)
151 else f"❌ Error: {type(result).__name__}"
152 for (_, model), result in zip(update_coros.values(), results, strict=True)
153 }
154 )
156 if any(isinstance(result, Exception) for result in results):
157 console.print(f"[bold red]ERROR: config update failed with {sum(1 for result in results if isinstance(result, Exception))} exceptions[/bold red]")
158 return
160 if verbose >= 1:
161 print_ekv(config.entity_kv, status=status, show_values=verbose >= 2)
162 elif updated := sum(len(models) for models in ekv.values()):
163 console.print(f"[green]{updated}[/green] updated keys for [red]{len(ekv)}[/red] entities")
166@config_group.command("schema")
167@click.option("-c", "--config", "file", help="Config file naming the modules to describe")
168@click.option("-o", "--output", help="Write the schema to FILE instead of standard output")
169async def config_schema(*, file: str | None, output: str | None):
170 """Generate a JSON Schema for the unified configuration file.
172 Sections are contributed by modules, so the schema describes the ones imported for this
173 run. Point at a config file (or set SENSORKIT_CONFIG) to cover a whole site, including
174 the sections its own modules register.
175 """
176 try:
177 sk.set_config_location(file)
178 await sk.import_modules(fail_policy="warn")
179 except Exception as e:
180 if (message := config_error_message(e, file)) is None:
181 raise
183 console.print(f"[bold red]ERROR: {message}[/bold red]")
184 return
186 document = json.dumps(sk.config_json_schema(), indent=2)
188 if output:
189 pathlib.Path(output).write_text(f"{document}\n")
190 console.print(f"[green]{output}[/green] written")
191 else:
192 click.echo(document)