thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
290 lines
13 KiB
Python
290 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from app.core.database import SessionLocal
|
|
from app.fixtures.normalize import normalize_fixture
|
|
from app.fixtures.ofl import OflClient
|
|
from app.models.entities import FixtureDefinition, FixtureInstance, FixtureSource
|
|
|
|
|
|
class FixtureService:
|
|
def __init__(
|
|
self,
|
|
session_factory: async_sessionmaker[AsyncSession] = SessionLocal,
|
|
client: OflClient | None = None,
|
|
) -> None:
|
|
self._session_factory = session_factory
|
|
self.client = client or OflClient()
|
|
self.test_data_path = Path("test-data/fixtures")
|
|
|
|
async def search_ofl(self, query: str) -> list[dict[str, object]]:
|
|
if not query.strip():
|
|
return []
|
|
return await self.client.search(query)
|
|
|
|
async def preview_ofl(self, manufacturer_key: str, fixture_key: str) -> dict[str, object]:
|
|
fixture = await self.client.fetch_fixture(manufacturer_key, fixture_key)
|
|
return normalize_fixture(fixture, manufacturer_key, fixture_key)
|
|
|
|
async def import_ofl(self, manufacturer_key: str, fixture_key: str) -> dict[str, object]:
|
|
source_payload = await self.client.fetch_fixture(manufacturer_key, fixture_key)
|
|
normalized = normalize_fixture(source_payload, manufacturer_key, fixture_key)
|
|
return await self._upsert_fixture(normalized, source_payload)
|
|
|
|
async def import_payload(self, payload: dict[str, object]) -> dict[str, object]:
|
|
normalized = self._normalize_import_payload(payload)
|
|
return await self._upsert_fixture(normalized, payload)
|
|
|
|
async def list_fixtures(self) -> list[dict[str, object]]:
|
|
async with self._session_factory() as session:
|
|
result = await session.execute(select(FixtureDefinition).order_by(FixtureDefinition.created_at, FixtureDefinition.id))
|
|
definitions = result.scalars().all()
|
|
return [self._serialize_definition(definition) for definition in definitions]
|
|
|
|
async def get_fixture(self, fixture_id: int) -> dict[str, object]:
|
|
async with self._session_factory() as session:
|
|
definition = await session.get(FixtureDefinition, fixture_id)
|
|
if definition is None:
|
|
raise LookupError("Fixture not found")
|
|
return self._serialize_definition(definition)
|
|
|
|
async def update_fixture(self, fixture_id: int, payload: dict[str, object]) -> dict[str, object]:
|
|
async with self._session_factory() as session:
|
|
definition = await session.get(FixtureDefinition, fixture_id)
|
|
if definition is None:
|
|
raise LookupError("Fixture not found")
|
|
|
|
existing = dict(definition.normalized_data)
|
|
merged = dict(existing)
|
|
for key in ("manufacturer", "model", "short_name", "categories", "modes", "schema_version", "source", "warnings"):
|
|
if key in payload:
|
|
merged[key] = payload[key]
|
|
|
|
definition.manufacturer = str(merged.get("manufacturer", definition.manufacturer))
|
|
definition.model = str(merged.get("model", definition.model))
|
|
definition.short_name = self._optional_str(merged.get("short_name"))
|
|
definition.categories = self._string_list(merged.get("categories", []))
|
|
definition.normalized_data = merged
|
|
await session.commit()
|
|
await session.refresh(definition)
|
|
return self._serialize_definition(definition)
|
|
|
|
async def delete_fixture(self, fixture_id: int) -> None:
|
|
async with self._session_factory() as session:
|
|
definition = await session.get(FixtureDefinition, fixture_id)
|
|
if definition is None:
|
|
raise LookupError("Fixture not found")
|
|
|
|
await session.execute(delete(FixtureInstance).where(FixtureInstance.definition_id == fixture_id))
|
|
source_id = definition.source_id
|
|
await session.delete(definition)
|
|
await session.flush()
|
|
if source_id is not None:
|
|
result = await session.execute(
|
|
select(FixtureDefinition).where(
|
|
FixtureDefinition.source_id == source_id,
|
|
FixtureDefinition.id != fixture_id,
|
|
).limit(1)
|
|
)
|
|
if result.scalar_one_or_none() is None:
|
|
source = await session.get(FixtureSource, source_id)
|
|
if source is not None:
|
|
await session.delete(source)
|
|
await session.commit()
|
|
|
|
async def _upsert_fixture(
|
|
self,
|
|
normalized: dict[str, object],
|
|
source_payload: dict[str, object],
|
|
) -> dict[str, object]:
|
|
source = normalized.get("source", {})
|
|
manufacturer_key = str(source.get("manufacturer_key", "custom")) if isinstance(source, dict) else "custom"
|
|
fixture_key = str(source.get("fixture_key", "custom")) if isinstance(source, dict) else "custom"
|
|
slug = f"{manufacturer_key}/{fixture_key}"
|
|
payload_hash = hashlib.sha256(json.dumps(source_payload, sort_keys=True).encode("utf-8")).hexdigest()
|
|
schema_ref = str(normalized.get("schema_version", "unknown"))
|
|
source_url = str(source_payload.get("oflURL", "")) if isinstance(source_payload, dict) else ""
|
|
|
|
async with self._session_factory() as session:
|
|
fixture_source = await self._upsert_source(
|
|
session,
|
|
manufacturer_key=manufacturer_key,
|
|
fixture_key=fixture_key,
|
|
schema_ref=schema_ref,
|
|
source_url=source_url,
|
|
payload=source_payload,
|
|
payload_hash=payload_hash,
|
|
)
|
|
|
|
result = await session.execute(select(FixtureDefinition).where(FixtureDefinition.slug == slug))
|
|
definition = result.scalar_one_or_none()
|
|
if definition is None:
|
|
definition = FixtureDefinition(
|
|
slug=slug,
|
|
manufacturer=str(normalized.get("manufacturer", manufacturer_key)),
|
|
model=str(normalized.get("model", fixture_key)),
|
|
short_name=self._optional_str(normalized.get("short_name")),
|
|
categories=self._string_list(normalized.get("categories", [])),
|
|
normalized_data=normalized,
|
|
source_id=fixture_source.id,
|
|
)
|
|
session.add(definition)
|
|
else:
|
|
definition.manufacturer = str(normalized.get("manufacturer", definition.manufacturer))
|
|
definition.model = str(normalized.get("model", definition.model))
|
|
definition.short_name = self._optional_str(normalized.get("short_name"))
|
|
definition.categories = self._string_list(normalized.get("categories", []))
|
|
definition.normalized_data = normalized
|
|
definition.source_id = fixture_source.id
|
|
|
|
await session.commit()
|
|
await session.refresh(definition)
|
|
return self._serialize_definition(definition)
|
|
|
|
async def _upsert_source(
|
|
self,
|
|
session: AsyncSession,
|
|
*,
|
|
manufacturer_key: str,
|
|
fixture_key: str,
|
|
schema_ref: str,
|
|
source_url: str,
|
|
payload: dict[str, object],
|
|
payload_hash: str,
|
|
) -> FixtureSource:
|
|
result = await session.execute(
|
|
select(FixtureSource).where(
|
|
FixtureSource.manufacturer_key == manufacturer_key,
|
|
FixtureSource.fixture_key == fixture_key,
|
|
)
|
|
)
|
|
fixture_source = result.scalar_one_or_none()
|
|
if fixture_source is None:
|
|
fixture_source = FixtureSource(
|
|
manufacturer_key=manufacturer_key,
|
|
fixture_key=fixture_key,
|
|
schema_ref=schema_ref,
|
|
source_url=source_url,
|
|
payload=payload,
|
|
payload_hash=payload_hash,
|
|
)
|
|
session.add(fixture_source)
|
|
await session.flush()
|
|
return fixture_source
|
|
|
|
fixture_source.schema_ref = schema_ref
|
|
fixture_source.source_url = source_url
|
|
fixture_source.payload = payload
|
|
fixture_source.payload_hash = payload_hash
|
|
await session.flush()
|
|
return fixture_source
|
|
|
|
def _serialize_definition(self, definition: FixtureDefinition) -> dict[str, object]:
|
|
normalized = dict(definition.normalized_data)
|
|
return {
|
|
"id": definition.id,
|
|
"slug": definition.slug,
|
|
"manufacturer": definition.manufacturer,
|
|
"model": definition.model,
|
|
"short_name": definition.short_name,
|
|
"categories": definition.categories,
|
|
"modes": normalized.get("modes", []),
|
|
"schema_version": normalized.get("schema_version", "unknown"),
|
|
"source": normalized.get("source", {}),
|
|
"warnings": normalized.get("warnings", []),
|
|
}
|
|
|
|
def _normalize_import_payload(self, payload: dict[str, object]) -> dict[str, object]:
|
|
if self._looks_like_ofl_fixture(payload):
|
|
source_payload = payload.get("source", {})
|
|
source = source_payload if isinstance(source_payload, dict) else {}
|
|
manufacturer_key = str(source.get("manufacturer_key", payload.get("manufacturer_key", "custom")))
|
|
fixture_key = str(
|
|
source.get(
|
|
"fixture_key",
|
|
payload.get("fixture_key", str(payload.get("name", "custom-fixture")).lower().replace(" ", "-")),
|
|
)
|
|
)
|
|
normalized = normalize_fixture(payload, manufacturer_key, fixture_key)
|
|
manufacturer_name = payload.get("manufacturer")
|
|
if manufacturer_name not in (None, ""):
|
|
normalized["manufacturer"] = str(manufacturer_name)
|
|
return normalized
|
|
return self._normalize_custom_payload(payload)
|
|
|
|
def _normalize_custom_payload(self, payload: dict[str, object]) -> dict[str, object]:
|
|
manufacturer = str(payload.get("manufacturer", "custom"))
|
|
model = str(payload.get("model", payload.get("name", "Custom fixture")))
|
|
source_payload = payload.get("source", {})
|
|
source = source_payload if isinstance(source_payload, dict) else {}
|
|
manufacturer_key = str(source.get("manufacturer_key", manufacturer.lower().replace(" ", "-")))
|
|
fixture_key = str(source.get("fixture_key", model.lower().replace(" ", "-")))
|
|
raw_modes = payload.get("modes", [])
|
|
modes: list[dict[str, object]] = []
|
|
for index, mode in enumerate(raw_modes if isinstance(raw_modes, list) else [], start=1):
|
|
if not isinstance(mode, dict):
|
|
continue
|
|
raw_channels = mode.get("channels", [])
|
|
channels: list[dict[str, object]] = []
|
|
for channel_index, channel in enumerate(raw_channels if isinstance(raw_channels, list) else [], start=1):
|
|
if isinstance(channel, dict):
|
|
channels.append(
|
|
{
|
|
"index": int(channel.get("index", channel_index)),
|
|
"key": str(channel.get("key", channel.get("display_name", f"channel-{channel_index}"))),
|
|
"display_name": str(channel.get("display_name", channel.get("key", f"Channel {channel_index}"))),
|
|
"precedence": str(channel.get("precedence", "ltp")).lower(),
|
|
"resolution": int(channel.get("resolution", 8)),
|
|
"capabilities": channel.get("capabilities", []),
|
|
}
|
|
)
|
|
elif isinstance(channel, str):
|
|
channels.append(
|
|
{
|
|
"index": channel_index,
|
|
"key": channel,
|
|
"display_name": channel,
|
|
"precedence": "htp" if "dim" in channel.lower() else "ltp",
|
|
"resolution": 8,
|
|
"capabilities": [],
|
|
}
|
|
)
|
|
modes.append(
|
|
{
|
|
"key": str(mode.get("key", mode.get("name", f"mode-{index}"))),
|
|
"channel_count": int(mode.get("channel_count", len(channels))),
|
|
"channels": channels,
|
|
}
|
|
)
|
|
return {
|
|
"manufacturer": manufacturer,
|
|
"model": model,
|
|
"short_name": payload.get("short_name"),
|
|
"categories": self._string_list(payload.get("categories", [])),
|
|
"modes": modes,
|
|
"schema_version": str(payload.get("schema_version", "custom")),
|
|
"source": {
|
|
"manufacturer_key": manufacturer_key,
|
|
"fixture_key": fixture_key,
|
|
},
|
|
"warnings": payload.get("warnings", []),
|
|
}
|
|
|
|
def _string_list(self, values: object) -> list[str]:
|
|
if not isinstance(values, list):
|
|
return []
|
|
return [str(value) for value in values]
|
|
|
|
def _optional_str(self, value: object) -> str | None:
|
|
return str(value) if value not in (None, "") else None
|
|
|
|
def _looks_like_ofl_fixture(self, payload: dict[str, object]) -> bool:
|
|
return isinstance(payload.get("availableChannels"), dict) and isinstance(payload.get("modes"), list)
|