ykcheungab
484 lines
17 KiB
Python
484 lines
17 KiB
Python
import json
|
|
import shutil
|
|
import sys
|
|
from contextlib import ExitStack
|
|
from dataclasses import dataclass
|
|
from importlib.resources import as_file, files
|
|
from pathlib import Path
|
|
|
|
# Product-facing names that map to on-disk config stems.
|
|
# Public product stems are ``box.json`` and ``finger.json``.
|
|
# Keep older lab names working: ``dd02`` still resolves to ``dd02-ov``.
|
|
CONFIG_ALIASES = {
|
|
"dd02": "dd02-ov",
|
|
}
|
|
|
|
PRODUCT_BUNDLES = ("box", "mini", "finger")
|
|
_RESOURCE_STACK = ExitStack()
|
|
|
|
|
|
def _resource_path(*parts: str) -> Path | None:
|
|
"""Resolve a package resource to a filesystem path for wheel-safe loaders."""
|
|
try:
|
|
resource = files("orisys").joinpath(*parts)
|
|
return _RESOURCE_STACK.enter_context(as_file(resource))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _unique_paths(paths) -> list[Path]:
|
|
"""Ordered, de-duplicated paths (Windows comparison is case-insensitive)."""
|
|
unique: list[Path] = []
|
|
seen: set[str] = set()
|
|
for path in paths:
|
|
try:
|
|
key = str(path.resolve()).casefold()
|
|
except OSError: # pragma: no cover - defensive: unresolvable path
|
|
key = str(path).casefold()
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
unique.append(path)
|
|
return unique
|
|
|
|
|
|
def _is_sdk_runtime_root(path: Path) -> bool:
|
|
"""Whether ``path`` carries the runnable SDK layout.
|
|
|
|
The anchor is the **folder layout**, never the folder's name: a distribution
|
|
folder may be called anything (a release checkout, a renamed copy, an
|
|
unpacked zip), but it always ships ``examples/`` beside ``sensors/``. That is
|
|
what distinguishes a run root from an arbitrary directory, and it is why a
|
|
renamed root keeps working while ``dist/**`` (``sensors/`` without
|
|
``examples/``) is not mistaken for one.
|
|
"""
|
|
try:
|
|
if not path.is_dir():
|
|
return False
|
|
return (path / "examples").is_dir() and (path / "sensors").is_dir()
|
|
except OSError: # pragma: no cover - defensive: unreadable path
|
|
return False
|
|
|
|
|
|
def _runtime_roots() -> list[Path]:
|
|
"""Runnable SDK roots for this process, nearest to the launcher first.
|
|
|
|
Evaluated once at import time, so a process that changes directory later
|
|
keeps the answer it started with.
|
|
"""
|
|
anchors: list[Path] = []
|
|
|
|
def add(path: Path) -> None:
|
|
if path not in anchors:
|
|
anchors.append(path)
|
|
|
|
cwd = Path.cwd()
|
|
add(cwd)
|
|
|
|
# A script run from a subdirectory (``<root>/examples/foo.py``, or a caller
|
|
# that launched from ``<root>/scripts``) still names its root through argv[0].
|
|
script_dir: Path | None = None
|
|
try:
|
|
argv0 = sys.argv[0] if sys.argv else ""
|
|
if argv0:
|
|
candidate = Path(argv0).expanduser()
|
|
script_dir = candidate.resolve().parent if candidate.is_file() else None
|
|
except (OSError, ValueError): # pragma: no cover - defensive
|
|
script_dir = None
|
|
|
|
# Walk up from both, so the root is found however deep the launcher sits.
|
|
for start in (cwd, script_dir):
|
|
if start is None:
|
|
continue
|
|
for candidate in (start, *start.parents):
|
|
add(candidate)
|
|
|
|
# Roots that put themselves on sys.path (``python <root>/examples/foo.py``
|
|
# with the root as cwd, ``PYTHONPATH=<root>``, and similar).
|
|
for entry in list(sys.path):
|
|
if not entry:
|
|
continue
|
|
try:
|
|
add(Path(entry).expanduser())
|
|
except (OSError, ValueError): # pragma: no cover - defensive
|
|
continue
|
|
|
|
anchors = _unique_paths(anchors)
|
|
# The run root is the *nearest* runnable directory, walking up from the
|
|
# launcher: starting at the cwd, then the script's directory, then the
|
|
# entries sys.path. Order carries the meaning here -- containment does not.
|
|
# Scanning every anchor for the shortest match instead picked the shallowest
|
|
# ancestor, so a workspace that carried ``examples/`` alongside ``sensors/``
|
|
# absorbed the real root below it and its bundle shadowed that root's own.
|
|
roots: list[Path] = []
|
|
for path in anchors:
|
|
if not _is_sdk_runtime_root(path):
|
|
continue
|
|
if any(existing.is_relative_to(path) or path.is_relative_to(existing) for existing in roots):
|
|
# Same tree: only the nearest root counts.
|
|
continue
|
|
roots.append(path)
|
|
return roots
|
|
|
|
|
|
def _candidate_config_dirs() -> list[Path]:
|
|
"""Candidate config directories, the running SDK root first.
|
|
|
|
A run root's ``config/`` outranks the packaged copy so a bundle's
|
|
``sensor.json`` and its ``force_calibration.json`` come from the same place.
|
|
"""
|
|
here = Path(__file__).resolve().parent
|
|
return _unique_paths([
|
|
*(root / "config" for root in _runtime_roots()),
|
|
here / "configs", # packaged wheel layout: orisys/configs/*.json
|
|
here.parent.parent / "config", # editable install from repo: repo_root/config/*.json
|
|
here.parent / "config", # fallback
|
|
Path.cwd() / "config", # script launched from repo root
|
|
])
|
|
|
|
|
|
def _candidate_sensor_roots() -> list[Path]:
|
|
"""Candidate roots that contain per-sensor bundle folders.
|
|
|
|
Lookup order:
|
|
|
|
1. the ``sensors/`` of every runnable SDK root found at startup, nearest
|
|
first -- so the bundle under the directory that is actually running wins,
|
|
whatever that directory is called;
|
|
2. the packaged resource ``orisys/sensors`` (wheel installs);
|
|
3. the module-relative candidates (editable checkout, packaged tree);
|
|
4. the current working directory, last, as a safety net.
|
|
"""
|
|
here = Path(__file__).resolve().parent
|
|
candidates = [root / "sensors" for root in _runtime_roots()]
|
|
packaged = _resource_path("sensors")
|
|
if packaged is not None:
|
|
candidates.append(packaged) # packaged wheel layout: orisys/sensors/
|
|
candidates.extend([
|
|
here / "sensors",
|
|
here.parent.parent / "sensors", # editable install: repo_root/sensors/
|
|
here.parent / "sensors",
|
|
Path.cwd() / "sensors",
|
|
])
|
|
return _unique_paths(candidates)
|
|
|
|
|
|
def _get_config_dir() -> Path:
|
|
for candidate in _candidate_config_dirs():
|
|
if candidate.exists() and any(candidate.glob("*.json")):
|
|
return candidate
|
|
# Default to the first candidate so error messages are stable.
|
|
return _candidate_config_dirs()[0]
|
|
|
|
|
|
CONFIG_DIR = _get_config_dir()
|
|
|
|
|
|
def _config_files() -> list[Path]:
|
|
if not CONFIG_DIR.exists():
|
|
return []
|
|
return sorted(CONFIG_DIR.glob("*.json"))
|
|
|
|
|
|
def _load_json(path: Path) -> dict:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SensorBundle:
|
|
"""Resolved paths for one additive per-sensor package folder."""
|
|
|
|
name: str
|
|
root: Path
|
|
sensor_json: Path
|
|
force_calibration_json: Path | None = None
|
|
manifest_json: Path | None = None
|
|
assets_dir: Path | None = None
|
|
|
|
|
|
def resolve_sensor_bundle(name: str) -> SensorBundle | None:
|
|
"""Resolve ``sensors/<name>/`` when present; otherwise return ``None``."""
|
|
requested = CONFIG_ALIASES.get(str(name), str(name))
|
|
for root in _candidate_sensor_roots():
|
|
bundle_root = root / requested
|
|
sensor_json = bundle_root / "sensor.json"
|
|
if not sensor_json.is_file():
|
|
continue
|
|
force_path = bundle_root / "force_calibration.json"
|
|
manifest_path = bundle_root / "manifest.json"
|
|
assets_dir = None
|
|
if manifest_path.is_file():
|
|
try:
|
|
manifest = _load_json(manifest_path)
|
|
except Exception:
|
|
manifest = {}
|
|
rel_assets = manifest.get("assets_dir")
|
|
if isinstance(rel_assets, str) and rel_assets.strip():
|
|
candidate = (bundle_root / rel_assets).resolve()
|
|
if candidate.is_dir():
|
|
assets_dir = candidate
|
|
return SensorBundle(
|
|
name=requested,
|
|
root=bundle_root.resolve(),
|
|
sensor_json=sensor_json.resolve(),
|
|
force_calibration_json=force_path.resolve() if force_path.is_file() else None,
|
|
manifest_json=manifest_path.resolve() if manifest_path.is_file() else None,
|
|
assets_dir=assets_dir,
|
|
)
|
|
return None
|
|
|
|
|
|
def describe_bundle_resolution(name: str) -> dict:
|
|
"""Explain how ``name`` resolved, so a missing calibration stops being silent.
|
|
|
|
A bundle can resolve and still carry no ``force_calibration.json``: the
|
|
lookup stops at the first root that has ``<name>/sensor.json`` and never
|
|
asks whether that root also has the coefficients. Reporting which bundle was
|
|
found -- and which roots were passed over -- is what makes that case
|
|
distinguishable from "no bundle at all", which is otherwise what the caller
|
|
sees.
|
|
|
|
Returns ``name``, the ``searched`` roots in order, and the first root that
|
|
supplied ``<name>/sensor.json`` together with whether it carried a
|
|
``force_calibration.json``.
|
|
"""
|
|
requested = CONFIG_ALIASES.get(str(name), str(name))
|
|
searched = _candidate_sensor_roots()
|
|
for root in searched:
|
|
sensor_json = root / requested / "sensor.json"
|
|
if not sensor_json.is_file():
|
|
continue
|
|
return {
|
|
"name": requested,
|
|
"searched": searched,
|
|
"found_root": sensor_json.parent,
|
|
"sensor_json": sensor_json,
|
|
"force_calibration_json": (
|
|
sensor_json.parent / "force_calibration.json"
|
|
).is_file(),
|
|
}
|
|
return {
|
|
"name": requested,
|
|
"searched": searched,
|
|
"found_root": None,
|
|
"sensor_json": None,
|
|
"force_calibration_json": False,
|
|
}
|
|
|
|
|
|
def resolve_sensor_config_path(name: str) -> Path | None:
|
|
"""Return ``sensor.json`` for a resolved sensor bundle."""
|
|
bundle = resolve_sensor_bundle(name)
|
|
return bundle.sensor_json if bundle is not None else None
|
|
|
|
|
|
def resolve_sensor_asset_dir(name: str) -> Path | None:
|
|
"""Return the default asset directory for a resolved sensor bundle."""
|
|
bundle = resolve_sensor_bundle(name)
|
|
return bundle.assets_dir if bundle is not None else None
|
|
|
|
|
|
def resolve_sensor_resource(name: str, *relative_parts: str) -> Path | None:
|
|
"""Return a path inside one sensor bundle when present."""
|
|
bundle = resolve_sensor_bundle(name)
|
|
if bundle is None:
|
|
return None
|
|
return bundle.root.joinpath(*relative_parts)
|
|
|
|
|
|
def list_sensor_bundles() -> list[str]:
|
|
"""Return product names that have an additive ``sensors/<name>/`` folder."""
|
|
found: list[str] = []
|
|
for name in PRODUCT_BUNDLES:
|
|
if resolve_sensor_bundle(name) is not None:
|
|
found.append(name)
|
|
return found
|
|
|
|
|
|
def load_force_calibration(name: str) -> dict | None:
|
|
"""Load ``force_calibration.json`` from a sensor bundle when available."""
|
|
bundle = resolve_sensor_bundle(name)
|
|
if bundle is None or bundle.force_calibration_json is None:
|
|
return None
|
|
return _load_json(bundle.force_calibration_json)
|
|
|
|
|
|
_BUNDLE_JSON_FILES = ("sensor.json", "force_calibration.json", "manifest.json")
|
|
|
|
|
|
def export_sensor_bundle(
|
|
name: str,
|
|
dest_dir: str | Path,
|
|
*,
|
|
include_assets: bool = False,
|
|
overwrite: bool = True,
|
|
) -> Path:
|
|
"""Copy one resolved ``sensors/<name>/`` bundle into ``dest_dir/<name>/``.
|
|
|
|
By default only JSON metadata is copied (``sensor.json``, and when present
|
|
``force_calibration.json`` / ``manifest.json``). Pass ``include_assets=True``
|
|
to copy the full tree (including ``assets/``) for packaging finger meshes.
|
|
"""
|
|
bundle = resolve_sensor_bundle(name)
|
|
if bundle is None:
|
|
available = ", ".join(list_sensor_bundles()) or "none"
|
|
raise FileNotFoundError(
|
|
f"Sensor bundle not found: {name!r}. Available: {available}"
|
|
)
|
|
|
|
dest_root = Path(dest_dir).expanduser().resolve()
|
|
dest = dest_root / bundle.name
|
|
if dest.exists():
|
|
if not overwrite:
|
|
raise FileExistsError(f"Destination already exists: {dest}")
|
|
if dest.is_dir():
|
|
shutil.rmtree(dest)
|
|
else:
|
|
dest.unlink()
|
|
|
|
if include_assets:
|
|
shutil.copytree(bundle.root, dest)
|
|
return dest
|
|
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
for filename in _BUNDLE_JSON_FILES:
|
|
src = bundle.root / filename
|
|
if src.is_file():
|
|
shutil.copy2(src, dest / filename)
|
|
if not (dest / "sensor.json").is_file():
|
|
raise FileNotFoundError(f"sensor.json missing after export from {bundle.root}")
|
|
return dest
|
|
|
|
|
|
def export_sensor_bundles(
|
|
dest_dir: str | Path,
|
|
*,
|
|
names: list[str] | tuple[str, ...] | None = None,
|
|
include_assets: bool = False,
|
|
overwrite: bool = True,
|
|
) -> list[Path]:
|
|
"""Export several sensor bundles into ``dest_dir``.
|
|
|
|
When ``names`` is omitted, exports every name in ``PRODUCT_BUNDLES`` that
|
|
currently resolves via :func:`resolve_sensor_bundle`.
|
|
"""
|
|
if names is None:
|
|
selected = [name for name in PRODUCT_BUNDLES if resolve_sensor_bundle(name) is not None]
|
|
else:
|
|
selected = [str(name) for name in names]
|
|
if not selected:
|
|
raise FileNotFoundError("No sensor bundles selected for export")
|
|
|
|
exported: list[Path] = []
|
|
for name in selected:
|
|
exported.append(
|
|
export_sensor_bundle(
|
|
name,
|
|
dest_dir,
|
|
include_assets=include_assets,
|
|
overwrite=overwrite,
|
|
)
|
|
)
|
|
return exported
|
|
|
|
|
|
def load_config(config_name_or_path):
|
|
"""
|
|
Load sensor configuration from a JSON file path or config name.
|
|
|
|
Config lookup order:
|
|
1. Exact file path (absolute or relative)
|
|
2. Additive sensor bundle ``sensors/<name>/sensor.json``
|
|
3. File stem under the config directory (e.g. ``box`` -> ``config/box.json``)
|
|
4. JSON field ``"name"`` matching the requested name (keeps ``prototype`` compatible)
|
|
"""
|
|
config_path = Path(config_name_or_path)
|
|
|
|
if config_path.exists() and config_path.is_file():
|
|
if config_path.suffix.lower() != ".json":
|
|
raise ValueError(f"Config file must be a JSON file: {config_path}")
|
|
return _load_json(config_path)
|
|
|
|
if config_path.is_absolute():
|
|
raise FileNotFoundError(f"Config file not found at absolute path: {config_path}")
|
|
|
|
requested = str(config_name_or_path)
|
|
candidates = [requested]
|
|
alias = CONFIG_ALIASES.get(requested)
|
|
if alias and alias not in candidates:
|
|
candidates.append(alias)
|
|
|
|
for name in candidates:
|
|
bundle = resolve_sensor_bundle(name)
|
|
if bundle is not None:
|
|
return _load_json(bundle.sensor_json)
|
|
|
|
for name in candidates:
|
|
direct = CONFIG_DIR / f"{name}.json"
|
|
if direct.exists():
|
|
return _load_json(direct)
|
|
|
|
# Backward-compatible alias resolution: support old names via JSON "name" field.
|
|
for candidate in _config_files():
|
|
try:
|
|
data = _load_json(candidate)
|
|
except Exception:
|
|
continue
|
|
if data.get("name") in candidates:
|
|
return data
|
|
|
|
available = [f.stem for f in _config_files()]
|
|
available_str = ", ".join(available) if available else "none"
|
|
raise FileNotFoundError(
|
|
f"Input config file not found with name or path: {config_name_or_path}\n"
|
|
f" Tried as file path: {config_path} (not found)\n"
|
|
f" Tried as config files: {', '.join(candidates)} (not found)\n"
|
|
f" Available config files: {available_str}"
|
|
)
|
|
|
|
|
|
def list_config():
|
|
"""
|
|
List all available JSON configuration files by file stem.
|
|
"""
|
|
stems = [f.stem for f in _config_files()]
|
|
for name in list_sensor_bundles():
|
|
if name not in stems:
|
|
stems.append(name)
|
|
return sorted(stems)
|
|
|
|
|
|
def export_config(config_name=None, path="./config/exported_config.json"):
|
|
"""
|
|
Export a configuration to a JSON file with readable formatting.
|
|
"""
|
|
config = load_config(config_name)
|
|
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
formatted = _format_json_value(config)
|
|
f.write(formatted)
|
|
f.write("\n")
|
|
|
|
print(f"Config name: {config_name} exported to {path}")
|
|
return config
|
|
|
|
|
|
def _format_json_value(value, indent_level=0, indent_str=" "):
|
|
if isinstance(value, dict):
|
|
if not value:
|
|
return "{}"
|
|
items = []
|
|
for k, v in value.items():
|
|
key_str = json.dumps(k, ensure_ascii=False)
|
|
value_str = _format_json_value(v, indent_level + 1, indent_str)
|
|
items.append(f"{indent_str * (indent_level + 1)}{key_str}: {value_str}")
|
|
return "{\n" + ",\n".join(items) + "\n" + indent_str * indent_level + "}"
|
|
if isinstance(value, list):
|
|
if not value:
|
|
return "[]"
|
|
items = [json.dumps(item, ensure_ascii=False) for item in value]
|
|
return "[" + ", ".join(items) + "]"
|
|
return json.dumps(value, ensure_ascii=False)
|