|
|
|
@@ -0,0 +1,567 @@
|
|
|
|
|
"""The sensor configuration catalog: which bundles exist, and what each one is.
|
|
|
|
|
|
|
|
|
|
Studio used to offer a hard-coded product list and to treat any ``*.json`` beside a
|
|
|
|
|
``sensor.json`` as another configuration. Both are replaced here by discovery over the two
|
|
|
|
|
roots the app actually has:
|
|
|
|
|
|
|
|
|
|
* the **shipped** root — ``config/sensors``, resolved exactly as
|
|
|
|
|
:func:`orisys_app_shared.paths.sensors_dir` resolves it today;
|
|
|
|
|
* a **writable user** root *beside* it — ``config/sensors-user``, because the app ships as a
|
|
|
|
|
portable folder and the operator's own bundles belong next to it, never inside the shipped
|
|
|
|
|
tree and never in Documents. When that directory cannot be written (a read-only install
|
|
|
|
|
medium, a locked-down install directory) the OS per-user area is used instead and the
|
|
|
|
|
catalog says which root is in use.
|
|
|
|
|
|
|
|
|
|
Discovery is the SDK's, in this order: a parseable ``manifest.json`` descriptor, then the
|
|
|
|
|
compatibility adapter for a legacy ``sensor.json`` bundle, then an entry retained as
|
|
|
|
|
**invalid** carrying the SDK's own explanation. This module never decides a processing
|
|
|
|
|
backend and never reads a folder name as evidence of one; it reports what
|
|
|
|
|
:func:`orisys.sensors.resolve_bundle` decided.
|
|
|
|
|
|
|
|
|
|
A folder holding only an arbitrary JSON document — ``force_calibration.json`` — is not a
|
|
|
|
|
configuration. It is still listed, marked invalid, and its reason says so, because silently
|
|
|
|
|
hiding it is how the same file ends up offered as a sensor config.
|
|
|
|
|
|
|
|
|
|
Two bundles that declare one ``profile_id`` are both listed and each names the other. Nothing
|
|
|
|
|
is shadowed, and the ambiguity is available to a caller that must refuse to start.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
import contextlib
|
|
|
|
|
import json
|
|
|
|
|
import os
|
|
|
|
|
import uuid
|
|
|
|
|
from dataclasses import dataclass, replace
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
|
|
|
|
|
|
from orisys_app_shared.paths import (
|
|
|
|
|
fallback_user_sensors_dir,
|
|
|
|
|
sensors_dir,
|
|
|
|
|
user_sensors_dir,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if TYPE_CHECKING: # pragma: no cover - typing only
|
|
|
|
|
from orisys.sensors import ResolvedSensorProfile
|
|
|
|
|
|
|
|
|
|
#: Which root an entry came from.
|
|
|
|
|
ORIGIN_SHIPPED = "shipped"
|
|
|
|
|
ORIGIN_USER = "user"
|
|
|
|
|
ORIGIN_FALLBACK = "fallback"
|
|
|
|
|
|
|
|
|
|
#: Which physical location the writable user root resolved to.
|
|
|
|
|
ROOT_KIND_SIBLING = "sibling"
|
|
|
|
|
ROOT_KIND_FALLBACK = "fallback"
|
|
|
|
|
ROOT_KIND_EXPLICIT = "explicit"
|
|
|
|
|
|
|
|
|
|
#: Family bucket for an entry that cannot be attributed to a product family. It is deliberately
|
|
|
|
|
#: not a product name: a folder of stray JSON has no product, and inventing one is how a
|
|
|
|
|
#: calibration file becomes a sensor.
|
|
|
|
|
FAMILY_UNIDENTIFIED = "__unidentified__"
|
|
|
|
|
|
|
|
|
|
#: The order the picker offers families in. The families the shipped tree has always provided
|
|
|
|
|
#: come first, in the order the picker has always used them, so replacing the hard-coded product
|
|
|
|
|
#: list does not reshuffle the picker for the products that exist today. This is presentation
|
|
|
|
|
#: order only — which families are offered comes from what is installed.
|
|
|
|
|
KNOWN_FAMILY_ORDER: tuple[str, ...] = ("box", "mini", "finger", "finger-mems")
|
|
|
|
|
|
|
|
|
|
#: ``CatalogEntry.state`` for a folder no bundle could be read from at all. Every other state is
|
|
|
|
|
#: the SDK's own :attr:`ValidationReport.state` (``ok`` / ``degraded`` / ``failed``).
|
|
|
|
|
STATE_UNREADABLE = "invalid"
|
|
|
|
|
|
|
|
|
|
_WRITE_PROBE_PREFIX = ".orisys-write-probe"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _sdk():
|
|
|
|
|
"""The SDK surface this module needs, imported on first use.
|
|
|
|
|
|
|
|
|
|
``orisys_app_shared.paths`` already reaches for the SDK lazily, so a shared module stays
|
|
|
|
|
importable without it; this keeps that behaviour instead of making the SDK a hard import of
|
|
|
|
|
the shared sensor package.
|
|
|
|
|
"""
|
|
|
|
|
from orisys import sensors
|
|
|
|
|
|
|
|
|
|
return sensors
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _sort_key(path: Path) -> str:
|
|
|
|
|
return path.name.lower()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
|
|
|
class UserRootDecision:
|
|
|
|
|
"""Where the writable user bundles live, and why."""
|
|
|
|
|
|
|
|
|
|
path: Path
|
|
|
|
|
#: ``sibling`` (the portable ``config/sensors-user``), ``fallback`` (the OS per-user area)
|
|
|
|
|
#: or ``explicit`` (named by the caller, as a test does).
|
|
|
|
|
kind: str
|
|
|
|
|
#: Whether a file can actually be created there right now.
|
|
|
|
|
writable: bool
|
|
|
|
|
reason: str
|
|
|
|
|
|
|
|
|
|
def __str__(self) -> str:
|
|
|
|
|
return self.reason
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
|
|
|
class CatalogRoot:
|
|
|
|
|
"""One directory the catalog scans."""
|
|
|
|
|
|
|
|
|
|
origin: str
|
|
|
|
|
path: Path
|
|
|
|
|
exists: bool
|
|
|
|
|
|
|
|
|
|
def __str__(self) -> str:
|
|
|
|
|
return f"{self.origin}: {self.path}" + ("" if self.exists else " (absent)")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
|
|
|
class CatalogEntry:
|
|
|
|
|
"""One bundle folder, as the catalog sees it.
|
|
|
|
|
|
|
|
|
|
``valid`` is the SDK's verdict (its validation report is usable), never a judgement made
|
|
|
|
|
here, so an entry the SDK rejects is shown with the reason instead of being dropped.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
#: Selection key: the configuration file the app already passes around, or the bundle folder
|
|
|
|
|
#: when no ``sensor.json`` could be found. Unique per entry.
|
|
|
|
|
key: str
|
|
|
|
|
folder: str
|
|
|
|
|
bundle_path: Path
|
|
|
|
|
origin: str
|
|
|
|
|
display_name: str
|
|
|
|
|
product_family: str
|
|
|
|
|
profile_id: str
|
|
|
|
|
sensor_json_path: Path | None
|
|
|
|
|
processing_backend: str
|
|
|
|
|
valid: bool
|
|
|
|
|
state: str
|
|
|
|
|
reason: str = ""
|
|
|
|
|
errors: tuple[str, ...] = ()
|
|
|
|
|
warnings: tuple[str, ...] = ()
|
|
|
|
|
diagnostics: tuple[str, ...] = ()
|
|
|
|
|
#: ``origin/folder`` of every other bundle declaring this entry's ``profile_id``.
|
|
|
|
|
duplicates: tuple[str, ...] = ()
|
|
|
|
|
profile: ResolvedSensorProfile | None = None
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def ref(self) -> str:
|
|
|
|
|
"""``origin/folder`` — readable, and unique across the two roots."""
|
|
|
|
|
return f"{self.origin}/{self.folder}"
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def family(self) -> str:
|
|
|
|
|
"""The family group this entry is listed under; never empty."""
|
|
|
|
|
return self.product_family or FAMILY_UNIDENTIFIED
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def duplicate(self) -> bool:
|
|
|
|
|
return bool(self.duplicates)
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def start_blocked_reason(self) -> str | None:
|
|
|
|
|
"""Why no session may be started with this entry, or ``None`` when one may.
|
|
|
|
|
|
|
|
|
|
The verdict belongs to the SDK: a profile its validation rejected is not startable, and
|
|
|
|
|
two bundles declaring one ``profile_id`` make choosing between them ambiguous. The entry
|
|
|
|
|
stays listed and explainable either way — that is the point of returning a reason.
|
|
|
|
|
"""
|
|
|
|
|
if not self.valid:
|
|
|
|
|
return self.reason or f"the SDK rejected this profile ({self.state})"
|
|
|
|
|
if self.duplicates:
|
|
|
|
|
return (
|
|
|
|
|
f"profile_id {self.profile_id!r} is also provided by "
|
|
|
|
|
f"{', '.join(self.duplicates)}; selecting between them is ambiguous"
|
|
|
|
|
)
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def label(self) -> str:
|
|
|
|
|
"""One line for a picker item: family, profile, origin and validity."""
|
|
|
|
|
return f"{self.family} · {self.profile_id or self.folder} · {self.origin} · {self.state}"
|
|
|
|
|
|
|
|
|
|
def tooltip(self) -> str:
|
|
|
|
|
"""The facts behind the label, including why the entry is not startable."""
|
|
|
|
|
lines = [
|
|
|
|
|
f"{self.display_name} ({self.ref})",
|
|
|
|
|
f"profile_id: {self.profile_id or 'unknown'}",
|
|
|
|
|
f"product family: {self.product_family or 'unknown'}",
|
|
|
|
|
f"processing backend: {self.processing_backend or 'unknown'}",
|
|
|
|
|
f"origin: {self.origin} ({self.bundle_path})",
|
|
|
|
|
f"state: {self.state}",
|
|
|
|
|
]
|
|
|
|
|
blocked = self.start_blocked_reason
|
|
|
|
|
if blocked:
|
|
|
|
|
lines.append(f"cannot start: {blocked}")
|
|
|
|
|
# The reason already quotes the errors it was built from; repeating them here would
|
|
|
|
|
# bury the one line that explains the entry.
|
|
|
|
|
lines.extend(
|
|
|
|
|
f"error: {text}" for text in self.errors if text not in self.reason
|
|
|
|
|
)
|
|
|
|
|
lines.extend(f"warning: {text}" for text in self.warnings)
|
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
|
|
|
class SensorCatalog:
|
|
|
|
|
"""Every configuration bundle the roots provide, with the roots themselves."""
|
|
|
|
|
|
|
|
|
|
roots: tuple[CatalogRoot, ...]
|
|
|
|
|
entries: tuple[CatalogEntry, ...]
|
|
|
|
|
user_root_decision: UserRootDecision
|
|
|
|
|
|
|
|
|
|
# ------------------------------------------------------------------ roots
|
|
|
|
|
@property
|
|
|
|
|
def shipped_root(self) -> Path:
|
|
|
|
|
return self.roots[0].path
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def user_root(self) -> Path:
|
|
|
|
|
return self.roots[1].path
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def user_root_exists(self) -> bool:
|
|
|
|
|
return self.roots[1].exists
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def user_root_kind(self) -> str:
|
|
|
|
|
return self.user_root_decision.kind
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def user_root_reason(self) -> str:
|
|
|
|
|
return self.user_root_decision.reason
|
|
|
|
|
|
|
|
|
|
# ------------------------------------------------------------------ entries
|
|
|
|
|
def entries_for_origin(self, origin: str) -> tuple[CatalogEntry, ...]:
|
|
|
|
|
wanted = str(origin or "")
|
|
|
|
|
return tuple(entry for entry in self.entries if entry.origin == wanted)
|
|
|
|
|
|
|
|
|
|
def entries_for_family(self, family: str) -> tuple[CatalogEntry, ...]:
|
|
|
|
|
"""The entries one family group lists; :data:`FAMILY_UNIDENTIFIED` is a group too."""
|
|
|
|
|
wanted = str(family or "")
|
|
|
|
|
if wanted == FAMILY_UNIDENTIFIED:
|
|
|
|
|
return tuple(entry for entry in self.entries if not entry.product_family)
|
|
|
|
|
return tuple(entry for entry in self.entries if entry.product_family == wanted)
|
|
|
|
|
|
|
|
|
|
def families(self) -> list[str]:
|
|
|
|
|
"""The family groups to offer, in the picker's order, unidentified last."""
|
|
|
|
|
present: list[str] = []
|
|
|
|
|
for entry in self.entries:
|
|
|
|
|
if entry.family not in present:
|
|
|
|
|
present.append(entry.family)
|
|
|
|
|
known = [name for name in KNOWN_FAMILY_ORDER if name in present]
|
|
|
|
|
others = sorted(
|
|
|
|
|
name
|
|
|
|
|
for name in present
|
|
|
|
|
if name not in KNOWN_FAMILY_ORDER and name != FAMILY_UNIDENTIFIED
|
|
|
|
|
)
|
|
|
|
|
tail = [FAMILY_UNIDENTIFIED] if FAMILY_UNIDENTIFIED in present else []
|
|
|
|
|
return known + others + tail
|
|
|
|
|
|
|
|
|
|
def invalid_entries(self) -> tuple[CatalogEntry, ...]:
|
|
|
|
|
return tuple(entry for entry in self.entries if not entry.valid)
|
|
|
|
|
|
|
|
|
|
def by_key(self, key: str) -> CatalogEntry | None:
|
|
|
|
|
wanted = str(key or "")
|
|
|
|
|
for entry in self.entries:
|
|
|
|
|
if entry.key == wanted:
|
|
|
|
|
return entry
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def by_ref(self, ref: str) -> CatalogEntry | None:
|
|
|
|
|
wanted = str(ref or "")
|
|
|
|
|
for entry in self.entries:
|
|
|
|
|
if entry.ref == wanted:
|
|
|
|
|
return entry
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def start_blocked(self, key_or_entry: CatalogEntry | str) -> str | None:
|
|
|
|
|
"""Why the named entry must not be started; ``None`` when it may be.
|
|
|
|
|
|
|
|
|
|
This is the whole answer a caller needs to refuse to start: it is ``None`` only for an
|
|
|
|
|
entry the SDK validated *and* that no other bundle duplicates.
|
|
|
|
|
"""
|
|
|
|
|
entry = (
|
|
|
|
|
key_or_entry
|
|
|
|
|
if isinstance(key_or_entry, CatalogEntry)
|
|
|
|
|
else self.by_key(str(key_or_entry))
|
|
|
|
|
)
|
|
|
|
|
return None if entry is None else entry.start_blocked_reason
|
|
|
|
|
|
|
|
|
|
def summary_lines(self) -> list[str]:
|
|
|
|
|
"""A short block for a log header or a diagnostics panel."""
|
|
|
|
|
lines = [
|
|
|
|
|
f"shipped root: {self.shipped_root}"
|
|
|
|
|
+ ("" if self.roots[0].exists else " (absent)"),
|
|
|
|
|
f"user root: {self.user_root} ({self.user_root_kind})"
|
|
|
|
|
+ ("" if self.user_root_exists else " (absent)"),
|
|
|
|
|
]
|
|
|
|
|
for entry in self.entries:
|
|
|
|
|
blocked = entry.start_blocked_reason
|
|
|
|
|
lines.append(
|
|
|
|
|
f" {entry.ref}: {entry.label()}" + (f" - {blocked}" if blocked else "")
|
|
|
|
|
)
|
|
|
|
|
return lines
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def resolve_user_root() -> UserRootDecision:
|
|
|
|
|
"""The writable user bundle root, preferring the portable sibling.
|
|
|
|
|
|
|
|
|
|
Returns the decision rather than just a path, because which root is in use is a fact the
|
|
|
|
|
app has to be able to report: a user who cannot find their bundles needs to be told where
|
|
|
|
|
they went, not shown a path the app silently chose.
|
|
|
|
|
"""
|
|
|
|
|
sibling = user_sensors_dir()
|
|
|
|
|
if _is_writable(sibling):
|
|
|
|
|
return UserRootDecision(
|
|
|
|
|
path=sibling,
|
|
|
|
|
kind=ROOT_KIND_SIBLING,
|
|
|
|
|
writable=True,
|
|
|
|
|
reason=f"{sibling} (sibling of the shipped root; writable)",
|
|
|
|
|
)
|
|
|
|
|
fallback = fallback_user_sensors_dir()
|
|
|
|
|
return UserRootDecision(
|
|
|
|
|
path=fallback,
|
|
|
|
|
kind=ROOT_KIND_FALLBACK,
|
|
|
|
|
writable=_is_writable(fallback),
|
|
|
|
|
reason=f"{sibling} is not writable; using {fallback}",
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _is_writable(target: Path) -> bool:
|
|
|
|
|
"""Whether a file can be created in ``target``, or beside it when it does not exist.
|
|
|
|
|
|
|
|
|
|
A real write is the only honest test on Windows, where a read-only directory still answers
|
|
|
|
|
``os.access(..., os.W_OK)``. The directory itself is never created: reporting where user
|
|
|
|
|
bundles *would* go must not have the side effect of creating that place, and a catalogue
|
|
|
|
|
scan must not change the tree it is describing.
|
|
|
|
|
"""
|
|
|
|
|
directory = target if target.is_dir() else target.parent
|
|
|
|
|
if not directory.is_dir():
|
|
|
|
|
return False
|
|
|
|
|
probe = directory / f"{_WRITE_PROBE_PREFIX}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
|
|
|
|
|
try:
|
|
|
|
|
probe.write_text("", encoding="utf-8")
|
|
|
|
|
except OSError:
|
|
|
|
|
return False
|
|
|
|
|
# A cleanup failure is not a writability finding: the file was created, which is the answer,
|
|
|
|
|
# and one empty probe file left behind is harmless.
|
|
|
|
|
with contextlib.suppress(OSError):
|
|
|
|
|
probe.unlink()
|
|
|
|
|
return True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def scan_sensor_catalog(
|
|
|
|
|
*,
|
|
|
|
|
shipped_root: str | Path | None = None,
|
|
|
|
|
user_root: str | Path | None = None,
|
|
|
|
|
) -> SensorCatalog:
|
|
|
|
|
"""Discover every bundle under the shipped and writable user roots.
|
|
|
|
|
|
|
|
|
|
``shipped_root`` defaults to :func:`orisys_app_shared.paths.sensors_dir` and ``user_root``
|
|
|
|
|
to whatever :func:`resolve_user_root` decided. Both are parameters so a test — or a caller
|
|
|
|
|
with an unusual layout — can point the catalog at another tree without touching the real
|
|
|
|
|
one. A root that does not exist contributes no entries and is not an error.
|
|
|
|
|
"""
|
|
|
|
|
shipped = Path(shipped_root) if shipped_root is not None else sensors_dir()
|
|
|
|
|
if user_root is None:
|
|
|
|
|
decision = resolve_user_root()
|
|
|
|
|
user = decision.path
|
|
|
|
|
user_origin = (
|
|
|
|
|
ORIGIN_FALLBACK if decision.kind == ROOT_KIND_FALLBACK else ORIGIN_USER
|
|
|
|
|
)
|
|
|
|
|
else:
|
|
|
|
|
user = Path(user_root)
|
|
|
|
|
decision = UserRootDecision(
|
|
|
|
|
path=user,
|
|
|
|
|
kind=ROOT_KIND_EXPLICIT,
|
|
|
|
|
writable=_is_writable(user),
|
|
|
|
|
reason=f"{user} (named by the caller)",
|
|
|
|
|
)
|
|
|
|
|
user_origin = ORIGIN_USER
|
|
|
|
|
|
|
|
|
|
roots = (
|
|
|
|
|
CatalogRoot(origin=ORIGIN_SHIPPED, path=shipped, exists=shipped.is_dir()),
|
|
|
|
|
CatalogRoot(origin=user_origin, path=user, exists=user.is_dir()),
|
|
|
|
|
)
|
|
|
|
|
entries: list[CatalogEntry] = []
|
|
|
|
|
for root in roots:
|
|
|
|
|
if root.exists:
|
|
|
|
|
entries.extend(_scan_root(root))
|
|
|
|
|
return SensorCatalog(
|
|
|
|
|
roots=roots,
|
|
|
|
|
entries=tuple(_mark_duplicates(entries)),
|
|
|
|
|
user_root_decision=decision,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _scan_root(root: CatalogRoot) -> list[CatalogEntry]:
|
|
|
|
|
"""Every bundle folder directly under one root, in a stable order."""
|
|
|
|
|
try:
|
|
|
|
|
children = sorted(
|
|
|
|
|
(path for path in root.path.iterdir() if path.is_dir()), key=_sort_key
|
|
|
|
|
)
|
|
|
|
|
except OSError:
|
|
|
|
|
# An unreadable root lists nothing; the root itself is still reported, so the reason a
|
|
|
|
|
# family is missing remains visible above the (empty) entry list.
|
|
|
|
|
return []
|
|
|
|
|
found: list[CatalogEntry] = []
|
|
|
|
|
for folder in children:
|
|
|
|
|
entry = _entry_for(folder, root.origin)
|
|
|
|
|
if entry is not None:
|
|
|
|
|
found.append(entry)
|
|
|
|
|
return found
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _entry_for(folder: Path, origin: str) -> CatalogEntry | None:
|
|
|
|
|
"""One child folder, or ``None`` when it is not a bundle candidate at all.
|
|
|
|
|
|
|
|
|
|
A candidate holds at least one top-level JSON document, because both a descriptor and a
|
|
|
|
|
configuration are JSON. Whether it *is* a usable configuration is then the SDK's answer,
|
|
|
|
|
never this module's.
|
|
|
|
|
"""
|
|
|
|
|
json_files = _json_files(folder)
|
|
|
|
|
if not json_files:
|
|
|
|
|
return None
|
|
|
|
|
try:
|
|
|
|
|
profile = _sdk().resolve_bundle(folder)
|
|
|
|
|
except Exception as exc: # noqa: BLE001 - one broken bundle must not hide the others
|
|
|
|
|
return _unreadable_entry(folder, origin, exc)
|
|
|
|
|
return _resolved_entry(folder, origin, profile)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _resolved_entry(
|
|
|
|
|
folder: Path, origin: str, profile: ResolvedSensorProfile
|
|
|
|
|
) -> CatalogEntry:
|
|
|
|
|
"""An entry the SDK resolved, valid or rejected, with its verdict kept whole."""
|
|
|
|
|
validation = profile.validation
|
|
|
|
|
errors = tuple(str(issue) for issue in validation.errors)
|
|
|
|
|
return CatalogEntry(
|
|
|
|
|
key=str(profile.sensor_json_path),
|
|
|
|
|
folder=folder.name,
|
|
|
|
|
bundle_path=folder.resolve(),
|
|
|
|
|
origin=origin,
|
|
|
|
|
display_name=profile.display_name,
|
|
|
|
|
product_family=profile.product_family,
|
|
|
|
|
profile_id=profile.profile_id,
|
|
|
|
|
sensor_json_path=profile.sensor_json_path,
|
|
|
|
|
processing_backend=profile.processing_backend,
|
|
|
|
|
valid=validation.usable,
|
|
|
|
|
state=validation.state,
|
|
|
|
|
reason="; ".join(errors),
|
|
|
|
|
errors=errors,
|
|
|
|
|
warnings=tuple(str(issue) for issue in validation.warnings),
|
|
|
|
|
diagnostics=tuple(profile.diagnostics),
|
|
|
|
|
profile=profile,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _unreadable_entry(folder: Path, origin: str, exc: Exception) -> CatalogEntry:
|
|
|
|
|
"""An entry no bundle could be read from, with the SDK's explanation as its reason."""
|
|
|
|
|
display_name, family, profile_id = _best_effort_identity(folder)
|
|
|
|
|
sensor_json = folder / "sensor.json"
|
|
|
|
|
has_config = sensor_json.is_file()
|
|
|
|
|
return CatalogEntry(
|
|
|
|
|
key=str(sensor_json if has_config else folder),
|
|
|
|
|
folder=folder.name,
|
|
|
|
|
bundle_path=folder.resolve(),
|
|
|
|
|
origin=origin,
|
|
|
|
|
display_name=display_name,
|
|
|
|
|
product_family=family,
|
|
|
|
|
profile_id=profile_id,
|
|
|
|
|
sensor_json_path=sensor_json if has_config else None,
|
|
|
|
|
processing_backend="",
|
|
|
|
|
valid=False,
|
|
|
|
|
state=STATE_UNREADABLE,
|
|
|
|
|
reason=_unreadable_reason(folder, exc),
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _unreadable_reason(folder: Path, exc: Exception) -> str:
|
|
|
|
|
"""The SDK's own explanation, prefixed when the folder holds no configuration at all.
|
|
|
|
|
|
|
|
|
|
"Not a built-in bundle" is true of a folder of stray JSON but does not say *why* it is not
|
|
|
|
|
a configuration. Naming the files that are there does, and it is the sentence that stops
|
|
|
|
|
``force_calibration.json`` from being offered as a sensor.
|
|
|
|
|
"""
|
|
|
|
|
manifest = folder / _sdk().DESCRIPTOR_FILENAME
|
|
|
|
|
note = ""
|
|
|
|
|
if not manifest.is_file() and not (folder / "sensor.json").is_file():
|
|
|
|
|
names = ", ".join(path.name for path in _json_files(folder))
|
|
|
|
|
note = (
|
|
|
|
|
f"only {names} present; a sensor configuration needs sensor.json or a "
|
|
|
|
|
"manifest.json descriptor, so this folder is not offered as one. "
|
|
|
|
|
)
|
|
|
|
|
return f"{folder.name}: {note}the SDK rejected it: {exc}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _best_effort_identity(folder: Path) -> tuple[str, str, str]:
|
|
|
|
|
"""``(display_name, family, profile_id)`` a broken bundle still declares.
|
|
|
|
|
|
|
|
|
|
A bundle whose resolution failed should not reduce to a bare folder name when the fields
|
|
|
|
|
that *are* readable identify it. Only ``sensor.json`` and ``manifest.json`` are consulted:
|
|
|
|
|
a calibration document's keys are not a profile's identity. The entry stays invalid.
|
|
|
|
|
"""
|
|
|
|
|
manifest = folder / _sdk().DESCRIPTOR_FILENAME
|
|
|
|
|
sensor_json = folder / "sensor.json"
|
|
|
|
|
data: dict[str, Any] = {}
|
|
|
|
|
for path in (sensor_json, manifest):
|
|
|
|
|
parsed = _read_json_object(path)
|
|
|
|
|
if parsed:
|
|
|
|
|
data.update(parsed)
|
|
|
|
|
built_in = _sdk().BUILT_IN_PROFILES.get(folder.name, {})
|
|
|
|
|
display = _text(data.get("display_name")) or folder.name
|
|
|
|
|
family = _text(data.get("product_family")) or _text(built_in.get("product_family"))
|
|
|
|
|
return display, family, _text(data.get("profile_id"))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _text(value: Any) -> str:
|
|
|
|
|
"""A non-empty stripped string, or ``""``."""
|
|
|
|
|
return value.strip() if isinstance(value, str) and value.strip() else ""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _read_json_object(path: Path) -> dict[str, Any] | None:
|
|
|
|
|
"""A JSON object, or ``None`` when the file is unreadable or is not an object."""
|
|
|
|
|
try:
|
|
|
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
except (OSError, ValueError):
|
|
|
|
|
return None
|
|
|
|
|
return data if isinstance(data, dict) else None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _json_files(folder: Path) -> tuple[Path, ...]:
|
|
|
|
|
"""The top-level JSON documents in ``folder``, sorted by name."""
|
|
|
|
|
try:
|
|
|
|
|
found = [
|
|
|
|
|
path
|
|
|
|
|
for path in folder.iterdir()
|
|
|
|
|
if path.is_file() and path.suffix.lower() == ".json"
|
|
|
|
|
]
|
|
|
|
|
except OSError:
|
|
|
|
|
return ()
|
|
|
|
|
return tuple(sorted(found, key=_sort_key))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _mark_duplicates(entries: list[CatalogEntry]) -> list[CatalogEntry]:
|
|
|
|
|
"""Flag every entry sharing a ``profile_id`` with another, naming the others.
|
|
|
|
|
|
|
|
|
|
Nothing is hidden or shadowed: both sides stay listed and each says which other bundles
|
|
|
|
|
declare the same profile, so a caller can refuse an ambiguous selection instead of
|
|
|
|
|
silently starting one of two profiles that claim to be the same thing.
|
|
|
|
|
"""
|
|
|
|
|
by_profile: dict[str, list[CatalogEntry]] = {}
|
|
|
|
|
for entry in entries:
|
|
|
|
|
if entry.profile_id:
|
|
|
|
|
by_profile.setdefault(entry.profile_id, []).append(entry)
|
|
|
|
|
if not any(len(group) > 1 for group in by_profile.values()):
|
|
|
|
|
return entries
|
|
|
|
|
marked: list[CatalogEntry] = []
|
|
|
|
|
for entry in entries:
|
|
|
|
|
others = tuple(
|
|
|
|
|
other.ref
|
|
|
|
|
for other in by_profile.get(entry.profile_id, [])
|
|
|
|
|
if other.key != entry.key
|
|
|
|
|
)
|
|
|
|
|
marked.append(replace(entry, duplicates=others) if others else entry)
|
|
|
|
|
return marked
|