domains.py
python
| 1 | """muse domains — domain plugin dashboard and scaffold wizard. |
| 2 | |
| 3 | Output (default — no flags):: |
| 4 | |
| 5 | ╔══════════════════════════════════════════════════════════════╗ |
| 6 | ║ Muse Domain Plugin Dashboard ║ |
| 7 | ╚══════════════════════════════════════════════════════════════╝ |
| 8 | |
| 9 | Registered domains: 2 |
| 10 | ───────────────────────────────────────────────────────────── |
| 11 | |
| 12 | midi ● midi/plugin.py |
| 13 | Capabilities: Typed Deltas · Domain Schema · OT Merge |
| 14 | Schema: version 1.0 · merge_mode: three_way |
| 15 | Elements: note_event (sequence), dimension_axes (set) |
| 16 | Dimensions: notes, pitch_bend, cc_volume, cc_sustain, tempo_map, track_structure (21 total) |
| 17 | |
| 18 | scaffold ○ scaffold/plugin.py |
| 19 | Capabilities: Typed Deltas · Domain Schema · OT Merge · CRDT |
| 20 | Schema: version 1.0 · merge_mode: three_way |
| 21 | Elements: record (sequence), attribute_set (set) |
| 22 | Dimensions: primary, metadata |
| 23 | |
| 24 | ───────────────────────────────────────────────────────────── |
| 25 | To scaffold a new domain: |
| 26 | muse domains --new <name> |
| 27 | ───────────────────────────────────────────────────────────── |
| 28 | |
| 29 | --json flag produces machine-readable output. |
| 30 | |
| 31 | --new <name> scaffolds a new domain plugin directory. |
| 32 | """ |
| 33 | |
| 34 | import json |
| 35 | import logging |
| 36 | import pathlib |
| 37 | import shutil |
| 38 | import sys |
| 39 | from typing import Literal |
| 40 | |
| 41 | import typer |
| 42 | |
| 43 | from muse.domain import CRDTPlugin, MuseDomainPlugin, StructuredMergePlugin |
| 44 | from muse.plugins.registry import _REGISTRY |
| 45 | |
| 46 | logger = logging.getLogger(__name__) |
| 47 | |
| 48 | app = typer.Typer() |
| 49 | |
| 50 | # --------------------------------------------------------------------------- |
| 51 | # Internal types |
| 52 | # --------------------------------------------------------------------------- |
| 53 | |
| 54 | _CapabilityLabel = Literal["Typed Deltas", "Domain Schema", "OT Merge", "CRDT"] |
| 55 | |
| 56 | |
| 57 | def _capabilities(plugin: MuseDomainPlugin) -> list[_CapabilityLabel]: |
| 58 | """Return the capability labels the plugin implements. |
| 59 | |
| 60 | Checks each optional protocol via ``isinstance`` — the same runtime |
| 61 | mechanism the core engine uses during merge dispatch. |
| 62 | |
| 63 | Args: |
| 64 | plugin: A registered ``MuseDomainPlugin`` instance. |
| 65 | |
| 66 | Returns: |
| 67 | Capability labels in ascending order: Typed Deltas → Domain Schema |
| 68 | → OT Merge → CRDT. Every plugin gets at least "Typed Deltas". |
| 69 | """ |
| 70 | caps: list[_CapabilityLabel] = ["Typed Deltas"] |
| 71 | try: |
| 72 | plugin.schema() |
| 73 | caps.append("Domain Schema") |
| 74 | except NotImplementedError: |
| 75 | return caps |
| 76 | if isinstance(plugin, StructuredMergePlugin): |
| 77 | caps.append("OT Merge") |
| 78 | if isinstance(plugin, CRDTPlugin): |
| 79 | caps.append("CRDT") |
| 80 | return caps |
| 81 | |
| 82 | |
| 83 | def _plugin_module_path(name: str) -> str: |
| 84 | """Return the module path for a plugin, relative to ``muse/plugins/``. |
| 85 | |
| 86 | Args: |
| 87 | name: Domain name string (key in the registry). |
| 88 | |
| 89 | Returns: |
| 90 | A display-friendly path string like ``music/plugin.py``. |
| 91 | """ |
| 92 | return f"{name}/plugin.py" |
| 93 | |
| 94 | |
| 95 | def _active_domain(root: pathlib.Path | None) -> str | None: |
| 96 | """Return the domain name of the repository at *root*, or ``None``. |
| 97 | |
| 98 | Args: |
| 99 | root: Repository root or ``None`` when not inside a repo. |
| 100 | |
| 101 | Returns: |
| 102 | Domain name string or ``None``. |
| 103 | """ |
| 104 | if root is None: |
| 105 | return None |
| 106 | repo_json = root / ".muse" / "repo.json" |
| 107 | if not repo_json.exists(): |
| 108 | return None |
| 109 | try: |
| 110 | data = json.loads(repo_json.read_text()) |
| 111 | domain = data.get("domain") |
| 112 | return str(domain) if domain else "midi" |
| 113 | except (OSError, json.JSONDecodeError): |
| 114 | return None |
| 115 | |
| 116 | |
| 117 | def _find_repo_root() -> pathlib.Path | None: |
| 118 | """Walk up from cwd to find a ``.muse/`` directory. |
| 119 | |
| 120 | Returns: |
| 121 | The repository root path, or ``None`` if not inside a repo. |
| 122 | """ |
| 123 | here = pathlib.Path.cwd() |
| 124 | for candidate in [here, *here.parents]: |
| 125 | if (candidate / ".muse").is_dir(): |
| 126 | return candidate |
| 127 | return None |
| 128 | |
| 129 | |
| 130 | # --------------------------------------------------------------------------- |
| 131 | # Scaffold wizard |
| 132 | # --------------------------------------------------------------------------- |
| 133 | |
| 134 | def _scaffold_new_domain(name: str) -> None: |
| 135 | """Create a new plugin directory by copying the scaffold template. |
| 136 | |
| 137 | Copies ``muse/plugins/scaffold/`` to ``muse/plugins/<name>/``, then |
| 138 | renames ``ScaffoldPlugin`` to ``<Name>Plugin`` in the source files. |
| 139 | |
| 140 | Args: |
| 141 | name: The new domain name (used as directory name and class prefix). |
| 142 | """ |
| 143 | scaffold_src = pathlib.Path(__file__).parents[2] / "plugins" / "scaffold" |
| 144 | dest = pathlib.Path(__file__).parents[2] / "plugins" / name |
| 145 | |
| 146 | if dest.exists(): |
| 147 | typer.echo(f"❌ Plugin directory already exists: {dest}", err=True) |
| 148 | raise typer.Exit(1) |
| 149 | |
| 150 | if not scaffold_src.exists(): |
| 151 | typer.echo( |
| 152 | f"❌ Scaffold source not found: {scaffold_src}\n" |
| 153 | "Make sure muse/plugins/scaffold/ exists.", |
| 154 | err=True, |
| 155 | ) |
| 156 | raise typer.Exit(1) |
| 157 | |
| 158 | shutil.copytree(str(scaffold_src), str(dest)) |
| 159 | |
| 160 | class_name = "".join(part.capitalize() for part in name.split("_")) + "Plugin" |
| 161 | |
| 162 | for py_file in dest.glob("*.py"): |
| 163 | text = py_file.read_text() |
| 164 | text = text.replace("ScaffoldPlugin", class_name) |
| 165 | text = text.replace('_DOMAIN_NAME = "scaffold"', f'_DOMAIN_NAME = "{name}"') |
| 166 | text = text.replace( |
| 167 | 'Scaffold domain plugin — copy-paste template for a new Muse domain.', |
| 168 | f'{class_name} — Muse domain plugin for the {name!r} domain.', |
| 169 | ) |
| 170 | py_file.write_text(text) |
| 171 | |
| 172 | typer.echo(f"✅ Scaffolded new domain plugin: muse/plugins/{name}/") |
| 173 | typer.echo(f" Class name: {class_name}") |
| 174 | typer.echo("") |
| 175 | typer.echo("Next steps:") |
| 176 | typer.echo(f" 1. Implement every NotImplementedError in muse/plugins/{name}/plugin.py") |
| 177 | typer.echo(" 2. Register the plugin in muse/plugins/registry.py:") |
| 178 | typer.echo(f' from muse.plugins.{name}.plugin import {class_name}') |
| 179 | typer.echo(f' _REGISTRY["{name}"] = {class_name}()') |
| 180 | typer.echo(f' 3. muse init --domain {name}') |
| 181 | typer.echo(" 4. See docs/guide/plugin-authoring-guide.md for the full walkthrough") |
| 182 | |
| 183 | |
| 184 | # --------------------------------------------------------------------------- |
| 185 | # JSON output |
| 186 | # --------------------------------------------------------------------------- |
| 187 | |
| 188 | def _emit_json(active_domain: str | None) -> None: |
| 189 | """Print all registered domains and their capabilities as JSON. |
| 190 | |
| 191 | Args: |
| 192 | active_domain: The domain of the current repo, or ``None``. |
| 193 | """ |
| 194 | result: list[dict[str, str | list[str] | dict[str, str | list[dict[str, str]]]]] = [] |
| 195 | for domain_name, plugin in sorted(_REGISTRY.items()): |
| 196 | caps = _capabilities(plugin) |
| 197 | entry: dict[str, str | list[str] | dict[str, str | list[dict[str, str]]]] = { |
| 198 | "domain": domain_name, |
| 199 | "capabilities": list(caps), |
| 200 | "active": "true" if domain_name == active_domain else "false", |
| 201 | } |
| 202 | try: |
| 203 | s = plugin.schema() |
| 204 | schema_dict: dict[str, str | list[dict[str, str]]] = { |
| 205 | "schema_version": str(s["schema_version"]), |
| 206 | "merge_mode": s["merge_mode"], |
| 207 | "description": s["description"], |
| 208 | "dimensions": [ |
| 209 | {"name": d["name"], "description": d["description"]} |
| 210 | for d in s["dimensions"] |
| 211 | ], |
| 212 | } |
| 213 | entry["schema"] = schema_dict |
| 214 | except NotImplementedError: |
| 215 | pass |
| 216 | result.append(entry) |
| 217 | typer.echo(json.dumps(result, indent=2)) |
| 218 | |
| 219 | |
| 220 | # --------------------------------------------------------------------------- |
| 221 | # Human-readable dashboard |
| 222 | # --------------------------------------------------------------------------- |
| 223 | |
| 224 | _WIDTH = 62 |
| 225 | |
| 226 | |
| 227 | def _box_line(text: str) -> str: |
| 228 | """Center *text* inside a box line of width ``_WIDTH``.""" |
| 229 | inner = _WIDTH - 2 |
| 230 | padded = text.center(inner) |
| 231 | return f"║{padded}║" |
| 232 | |
| 233 | |
| 234 | def _hr() -> str: |
| 235 | return "─" * _WIDTH |
| 236 | |
| 237 | |
| 238 | def _print_dashboard(active_domain: str | None) -> None: |
| 239 | """Print the human-readable domain dashboard. |
| 240 | |
| 241 | Args: |
| 242 | active_domain: Domain of the current repo (highlighted with ●), or ``None``. |
| 243 | """ |
| 244 | typer.echo("╔" + "═" * (_WIDTH - 2) + "╗") |
| 245 | typer.echo(_box_line("Muse Domain Plugin Dashboard")) |
| 246 | typer.echo("╚" + "═" * (_WIDTH - 2) + "╝") |
| 247 | typer.echo("") |
| 248 | |
| 249 | count = len(_REGISTRY) |
| 250 | typer.echo(f"Registered domains: {count}") |
| 251 | typer.echo(_hr()) |
| 252 | |
| 253 | for domain_name, plugin in sorted(_REGISTRY.items()): |
| 254 | caps = _capabilities(plugin) |
| 255 | is_active = domain_name == active_domain |
| 256 | bullet = "●" if is_active else "○" |
| 257 | module_path = _plugin_module_path(domain_name) |
| 258 | |
| 259 | typer.echo("") |
| 260 | active_suffix = " ← active repo domain" if is_active else "" |
| 261 | typer.echo(f" {bullet} {domain_name}{active_suffix}") |
| 262 | typer.echo(f" Module: plugins/{module_path}") |
| 263 | typer.echo(f" Capabilities: {' · '.join(caps)}") |
| 264 | |
| 265 | try: |
| 266 | s = plugin.schema() |
| 267 | dim_names = [d["name"] for d in s["dimensions"]] |
| 268 | top_kind = s["top_level"]["kind"] |
| 269 | typer.echo( |
| 270 | f" Schema: v{s['schema_version']} · " |
| 271 | f"top_level: {top_kind} · merge_mode: {s['merge_mode']}" |
| 272 | ) |
| 273 | typer.echo(f" Dimensions: {', '.join(dim_names)}") |
| 274 | typer.echo(f" Description: {s['description'][:55]}") |
| 275 | except NotImplementedError: |
| 276 | typer.echo(" Schema: (not declared)") |
| 277 | |
| 278 | typer.echo("") |
| 279 | typer.echo(_hr()) |
| 280 | typer.echo("To scaffold a new domain:") |
| 281 | typer.echo(" muse domains --new <name>") |
| 282 | typer.echo("To see machine-readable output:") |
| 283 | typer.echo(" muse domains --json") |
| 284 | typer.echo("See docs/guide/plugin-authoring-guide.md for the full walkthrough.") |
| 285 | typer.echo(_hr()) |
| 286 | |
| 287 | |
| 288 | # --------------------------------------------------------------------------- |
| 289 | # CLI entry point |
| 290 | # --------------------------------------------------------------------------- |
| 291 | |
| 292 | @app.callback(invoke_without_command=True) |
| 293 | def domains( |
| 294 | ctx: typer.Context, |
| 295 | new: str | None = typer.Option( |
| 296 | None, |
| 297 | "--new", |
| 298 | metavar="NAME", |
| 299 | help="Scaffold a new domain plugin with the given name.", |
| 300 | ), |
| 301 | as_json: bool = typer.Option( |
| 302 | False, |
| 303 | "--json", |
| 304 | help="Emit domain registry as JSON.", |
| 305 | ), |
| 306 | ) -> None: |
| 307 | """Domain plugin dashboard — list registered domains and their capabilities. |
| 308 | |
| 309 | Without flags: prints a human-readable table of all registered domains, |
| 310 | their capability levels (Typed Deltas / Domain Schema / OT Merge / CRDT), |
| 311 | and their declared schemas. |
| 312 | |
| 313 | Use ``--new <name>`` to scaffold a new domain plugin directory from the |
| 314 | scaffold template. |
| 315 | |
| 316 | Use ``--json`` for machine-readable output. |
| 317 | """ |
| 318 | if new is not None: |
| 319 | _scaffold_new_domain(new) |
| 320 | return |
| 321 | |
| 322 | active_domain = _active_domain(_find_repo_root()) |
| 323 | |
| 324 | if as_json: |
| 325 | _emit_json(active_domain) |
| 326 | return |
| 327 | |
| 328 | _print_dashboard(active_domain) |