Leopoldadmin/hms-mediaengine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Phase 5: Projekt-/Media-/Plugin-API im Control Core (§23.1, §13.3)

Browse Source 
- Projekte: CRUD + Activate (State-Store-Revision), §23.1/§24.2
- Medien: Import mit Duplikaterkennung (§13.3), Liste mit
  Missing-Markierung (§13.2)
- Plugins: Listing nach Lifecycle-Status gruppiert (§14.5),
  Enable mit Lifecycle-Advancement, 404 bei unbekanntem Plugin
- Art-Net-Status-Endpunkt (§23.1)
- Diagnostics erweitert um State-/Projekt-Revisionen (§28.2)
- Commands schreiben jetzt auch in den State-Store (§6.4)
- SQLite check_same_thread=False für FastAPI async handlers
- python-multipart für UploadFile hinzugefügt
- Discovery-Fix: Builtin-Plugins nach Kategorie-Unterverzeichnissen
  (generators/, filters/) durchsuchen statt nur Direktkinder
- 8 neue Integrationstests: Projekt-CRUD-Roundtrip, Activate,
  Media-Import mit Duplikat, Plugin-Listing (10+14 verifiziert),
  Enable-404, Diagnostics-Revisionen
- Gesamtsuite 557 grün, Ruff grün
This commit is contained in:
HMS MediaEngine Agent
2026-09-11 03:13:43 +02:00
parent bae05ea8ad
commit 82c79ec485
6 changed files with 384 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/control_server/hms_control_server/app.py
+228 -12
View File
@@ -9,6 +9,11 @@ Endpunkte:
- POST /api/v1/commands/{command_id}/release
- GET /api/v1/diagnostics
- GET /api/v1/cluster/nodes
- GET /api/v1/projects, POST/GET/PUT/DELETE /api/v1/projects/{id}
- POST /api/v1/media/import
- GET /api/v1/media
- GET /api/v1/plugins
- GET /api/v1/artnet/status
- WS /ws (State-Snapshot + Updates)
Commands folgen §23.2: command_id, type, expected_revision, actor, payload.
@@ -20,16 +25,23 @@ from __future__ import annotations
import asyncio
import uuid
from pathlib import Path
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from fastapi import FastAPI, HTTPException, UploadFile, WebSocket, WebSocketDisconnect
from hms_capabilities.probe import CapabilityReport
from hms_cluster.registry import NodeRegistry
from hms_domain.identity import NodeIdentity, NodeRole
from hms_domain.model import (
Project,
)
from hms_media import ImportStatus, MediaLibrary
from hms_parameter.engine import (
ControlSource,
ParameterEngine,
RevisionConflict,
)
from hms_persistence import Database
from hms_persistence.state_store import ProjectStateStore
from hms_protocol.idempotency import IdempotencyRegistry
from pydantic import BaseModel, Field
@@ -48,14 +60,51 @@ class ReleaseCommand(BaseModel):
source: str = "web"
class CreateProjectBody(BaseModel):
name: str = "Neues Projekt"
def _builtin_plugin_dir() -> Path:
"""Löst das Builtin-Plugin-Verzeichnis robust auf (§14.2).
Sucht vom aktuellen Arbeitsverzeichnis aus über bekannte Kandidaten;
im Entwickungsbaum ist das Repo-Root die Basis.
"""
candidates = [
Path.cwd() / "plugins" / "builtin",
Path(__file__).resolve().parents[3] / "plugins" / "builtin",
]
for candidate in candidates:
if candidate.is_dir():
return candidate
return candidates[0]
def _discover_builtin_plugins(mgr, builtin_dir: Path) -> list[str]:
"""Entdeckt Plugins je Kategorie-Unterverzeichnis (§14.2)."""
errors: list[str] = []
for category in sorted(builtin_dir.iterdir()):
if not category.is_dir():
continue
errors.extend(mgr.discover(category))
return errors
class _State:
def __init__(self, identity: NodeIdentity) -> None:
def __init__(self, identity: NodeIdentity, data_dir: Path | None = None) -> None:
self.identity = identity
self.engine = ParameterEngine()
self.registry = IdempotencyRegistry()
self.report = CapabilityReport()
self.nodes = NodeRegistry()
self.subscribers: list[asyncio.Queue] = []
# Persistenz und Projekt-Verwaltung (§24, §6.4)
self.data_dir = data_dir or Path("/tmp/hms-dev-data")
self.database = Database(self.data_dir / "database" / "hms.db")
self.database.open()
self.database.migrate(backup_dir=self.data_dir / "backups")
self.state_store = ProjectStateStore()
self.media_library = MediaLibrary(self.data_dir / "media")
def create_app(identity: NodeIdentity | None = None) -> FastAPI:
@@ -82,11 +131,13 @@ def create_app(identity: NodeIdentity | None = None) -> FastAPI:
for queue in list(state.subscribers):
queue.put_nowait(event)
# ---------- System (§23.1) ----------
@app.get("/api/v1/system/health")
async def health() -> dict:
return {
"status": "ok",
"phase": 1,
"phase": 5,
"node_id": state.identity.node_id,
"revision": state.engine.revision,
}
@@ -111,6 +162,19 @@ def create_app(identity: NodeIdentity | None = None) -> FastAPI:
snap = state.engine.snapshot()
return {"revision": snap.revision, "values": snap.as_dict()}
@app.get("/api/v1/diagnostics")
async def diagnostics() -> dict:
return {
"renderer": "not_connected",
"artnet": "not_started",
"revision": state.engine.revision,
"node_id": state.identity.node_id,
"state_revision": state.state_store.state_revision,
"project_revision": state.state_store.project_revision,
}
# ---------- Commands (§23.2) ----------
@app.post("/api/v1/commands")
async def post_command(cmd: SetParameterCommand) -> dict:
if cmd.type != "parameter.set":
@@ -131,6 +195,7 @@ def create_app(identity: NodeIdentity | None = None) -> FastAPI:
source=ControlSource.WEB,
expected_revision=cmd.expected_revision,
)
state.state_store.set_value(path, float(value))
except RevisionConflict as exc:
raise HTTPException(
status_code=409,
@@ -161,17 +226,9 @@ def create_app(identity: NodeIdentity | None = None) -> FastAPI:
@app.post("/api/v1/commands/{command_id}/release")
async def release_override(command_id: str) -> dict:
# Release nach §11.3; command_id referenziert den ursprünglichen Command.
return {"status": "not_implemented_in_phase0"}
@app.get("/api/v1/diagnostics")
async def diagnostics() -> dict:
return {
"renderer": "not_connected", # IPC-Handshake folgt (ADR-0003)
"artnet": "not_started",
"revision": state.engine.revision,
"node_id": state.identity.node_id,
}
# ---------- Cluster (§23.1) ----------
@app.get("/api/v1/cluster/nodes")
async def cluster_nodes() -> dict:
@@ -190,6 +247,165 @@ def create_app(identity: NodeIdentity | None = None) -> FastAPI:
)
return {"self": state.identity.node_id, "nodes": nodes}
# ---------- Projekte (§23.1, §24.2) ----------
@app.get("/api/v1/projects")
async def list_projects() -> dict:
return {"projects": state.database.list_projects()}
@app.post("/api/v1/projects")
async def create_project(body: CreateProjectBody) -> dict:
project = Project(name=body.name)
state.database.save_project(project.model_dump(mode="json"))
return {"status": "ack", "project": project.model_dump(mode="json")}
@app.get("/api/v1/projects/{project_id}")
async def get_project(project_id: str) -> dict:
data = state.database.load_project(project_id)
if data is None:
raise HTTPException(status_code=404, detail="Projekt nicht gefunden")
return data
@app.put("/api/v1/projects/{project_id}")
async def update_project(project_id: str, body: dict) -> dict:
existing = state.database.load_project(project_id)
if existing is None:
raise HTTPException(status_code=404, detail="Projekt nicht gefunden")
for key, value in body.items():
if key in existing:
existing[key] = value
updated = Project.model_validate(existing)
state.database.save_project(updated.model_dump(mode="json"))
state.state_store.bump_project_revision()
return {"status": "ack"}
@app.delete("/api/v1/projects/{project_id}")
async def delete_project(project_id: str) -> dict:
state.database.delete_project(project_id)
return {"status": "ack"}
@app.post("/api/v1/projects/{project_id}/activate")
async def activate_project(project_id: str) -> dict:
"""Aktiviert ein Projekt als autoritative Basis (§6.4)."""
data = state.database.load_project(project_id)
if data is None:
raise HTTPException(status_code=404, detail="Projekt nicht gefunden")
project = Project.model_validate(data)
revision = state.state_store.activate_project(project)
return {"status": "ack", "state_revision": revision}
# ---------- Medien (§13.3, §23.1) ----------
@app.post("/api/v1/media/import")
async def import_media(file: UploadFile) -> dict:
"""Importiert eine Mediendatei mit Duplikaterkennung (§13.3).
Import blockiert niemals den Renderthread (§13.3).
"""
media_root = state.data_dir / "media"
media_root.mkdir(parents=True, exist_ok=True)
dest = media_root / file.filename
content = await file.read()
dest.write_bytes(content)
outcome = state.media_library.import_file(dest)
if outcome.status is ImportStatus.IMPORTED and outcome.asset:
state.database.save_project({
"id": str(uuid.uuid4()),
"name": file.filename,
"schema_version": 1,
"created_at": "2026-09-11T00:00:00+00:00",
"updated_at": "2026-09-11T00:00:00+00:00",
"data": outcome.asset.model_dump(mode="json"),
})
return {
"status": outcome.status.value,
"asset": outcome.asset.model_dump(mode="json") if outcome.asset else None,
"duplicate_of": outcome.duplicate_of,
}
@app.get("/api/v1/media")
async def list_media() -> dict:
"""Medienliste mit Metadaten (§13.2)."""
missing = state.media_library.mark_missing()
assets = []
for asset in state.media_library.all():
assets.append({
**asset.model_dump(mode="json"),
"missing": asset.id in state.media_library.missing_asset_ids,
})
return {"media": assets, "missing_count": len(missing)}
# ---------- Plugins (§23.1) ----------
@app.get("/api/v1/plugins")
async def list_plugins() -> dict:
"""Installierte/verfügbare Plugins nach Status gruppiert (§14.5)."""
from hms_plugin_sdk import PluginLifecycleManager
builtin_dir = _builtin_plugin_dir()
mgr = PluginLifecycleManager()
errors = _discover_builtin_plugins(mgr, builtin_dir)
plugins_by_state: dict[str, list[dict]] = {}
for record in mgr.all():
plugins_by_state.setdefault(record.state.value, []).append(
{
"plugin_id": record.plugin_id,
"version": record.version,
"kind": record.manifest.get("kind", "unknown"),
"name": record.manifest.get("name", record.plugin_id),
"last_error": record.last_error,
}
)
return {
"plugins_by_state": plugins_by_state,
"total": len(mgr.all()),
"discovery_errors": errors,
}
@app.post("/api/v1/plugins/{plugin_id}/enable")
async def enable_plugin(plugin_id: str) -> dict:
"""Aktiviert ein Plugin über den Lifecycle (§14.5)."""
from hms_plugin_sdk import (
InvalidTransitionError,
LifecycleState,
PluginLifecycleManager,
)
builtin_dir = _builtin_plugin_dir()
mgr = PluginLifecycleManager()
_discover_builtin_plugins(mgr, builtin_dir)
if mgr.get(plugin_id) is None:
raise HTTPException(status_code=404, detail=f"unbekanntes Plugin {plugin_id!r}")
try:
for target in (
LifecycleState.INSTALLED,
LifecycleState.ENABLED,
LifecycleState.COMPILED,
LifecycleState.ACTIVE,
):
record = mgr.get(plugin_id)
if record is None:
break
if record.state is target:
break
mgr.advance(plugin_id, target)
return {"status": "ack", "plugin_id": plugin_id}
except InvalidTransitionError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
# ---------- Art-Net (§23.1) ----------
@app.get("/api/v1/artnet/status")
async def artnet_status() -> dict:
return {
"running": False,
"universes": [],
"telemetry": {},
"note": "Art-Net-Receiver startet mit dem Launcher (Phase 1 vollstaendig nach Gate 1)",
}
# ---------- WebSocket (§23.3) ----------
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket) -> None:
await ws.accept()