initial commit
This commit is contained in:
ykcheungab
@@ -0,0 +1,3 @@
|
||||
"""Orisys nominal sensor design tooling (minimal v1)."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,281 @@
|
||||
"""orisys-design CLI — validate, export, manifest."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from orisys_sensor_design import __version__
|
||||
from orisys_sensor_design.manifest import (
|
||||
RevisionPaths,
|
||||
build_export_job,
|
||||
build_manifest,
|
||||
default_manifest_path,
|
||||
export_revision_dir,
|
||||
export_output_dir,
|
||||
load_profile,
|
||||
repo_root,
|
||||
validate_revision_metadata,
|
||||
write_export_job,
|
||||
write_manifest,
|
||||
)
|
||||
|
||||
|
||||
def _blender_from_lock(root: Path) -> Path | None:
|
||||
lock_path = root / "blender.lock.json"
|
||||
if not lock_path.is_file():
|
||||
return None
|
||||
try:
|
||||
data = json.loads(lock_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
raw = data.get("executable")
|
||||
if not raw:
|
||||
return None
|
||||
path = Path(str(raw)).expanduser()
|
||||
if path.is_file():
|
||||
return path.resolve()
|
||||
return None
|
||||
|
||||
|
||||
def _lock_preferred_versions(root: Path | None) -> list[str]:
|
||||
"""Return preferred version tokens from blender.lock.json (lts then full)."""
|
||||
if root is None:
|
||||
return []
|
||||
lock_path = root / "blender.lock.json"
|
||||
if not lock_path.is_file():
|
||||
return []
|
||||
try:
|
||||
data = json.loads(lock_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return []
|
||||
preferred: list[str] = []
|
||||
for key in ("blender_lts", "blender_version"):
|
||||
value = data.get(key)
|
||||
if isinstance(value, str) and value.strip():
|
||||
preferred.append(value.strip())
|
||||
return preferred
|
||||
|
||||
|
||||
def _blender_install_candidates() -> list[Path]:
|
||||
"""Collect blender.exe under Program Files / Blender Foundation."""
|
||||
roots: list[Path] = []
|
||||
for env_name in ("ProgramFiles", "ProgramFiles(x86)"):
|
||||
raw = os.environ.get(env_name)
|
||||
if raw:
|
||||
roots.append(Path(raw))
|
||||
# Fallbacks when env is missing (some sandboxed shells).
|
||||
for fallback in (r"C:\Program Files", r"C:\Program Files (x86)"):
|
||||
path = Path(fallback)
|
||||
if path not in roots:
|
||||
roots.append(path)
|
||||
|
||||
found: list[Path] = []
|
||||
seen: set[Path] = set()
|
||||
for root in roots:
|
||||
base = root / "Blender Foundation"
|
||||
if not base.is_dir():
|
||||
continue
|
||||
for child in sorted(base.glob("Blender *"), key=lambda p: p.name, reverse=True):
|
||||
exe = child / "blender.exe"
|
||||
if not exe.is_file():
|
||||
continue
|
||||
resolved = exe.resolve()
|
||||
if resolved not in seen:
|
||||
seen.add(resolved)
|
||||
found.append(resolved)
|
||||
return found
|
||||
|
||||
|
||||
def _blender_from_program_files(root: Path | None = None) -> Path | None:
|
||||
"""Pick a Blender install from Program Files; prefer lock version, else newest."""
|
||||
candidates = _blender_install_candidates()
|
||||
if not candidates:
|
||||
return None
|
||||
|
||||
for preferred in _lock_preferred_versions(root):
|
||||
for exe in candidates:
|
||||
# Folder names look like "Blender 4.2" or "Blender 5.0".
|
||||
if preferred in exe.parent.name:
|
||||
return exe
|
||||
|
||||
# Newest by folder name (already sorted descending in candidates).
|
||||
return candidates[0]
|
||||
|
||||
|
||||
def resolve_blender_executable(root: Path | None = None) -> Path | None:
|
||||
"""Find Blender without hard-coding a machine path.
|
||||
|
||||
Order: ORISYS_BLENDER / BLENDER_EXE → PATH → Program Files scan →
|
||||
optional blender.lock.json executable.
|
||||
"""
|
||||
for env_name in ("ORISYS_BLENDER", "BLENDER_EXE"):
|
||||
raw = os.environ.get(env_name)
|
||||
if raw:
|
||||
path = Path(raw).expanduser()
|
||||
if path.is_file():
|
||||
return path.resolve()
|
||||
for name in ("blender", "blender.exe"):
|
||||
found = shutil.which(name)
|
||||
if found:
|
||||
return Path(found).resolve()
|
||||
from_pf = _blender_from_program_files(root)
|
||||
if from_pf is not None:
|
||||
return from_pf
|
||||
if root is not None:
|
||||
return _blender_from_lock(root)
|
||||
return None
|
||||
|
||||
|
||||
def _run_blender_script(script: str, blend: Path, extra_args: list[str]) -> int:
|
||||
root = repo_root()
|
||||
blender = resolve_blender_executable(root)
|
||||
script_path = root / "blender" / "scripts" / script
|
||||
if blender is None:
|
||||
manual = (
|
||||
f"blender -b {blend} -P {script_path} -- {' '.join(extra_args)}"
|
||||
)
|
||||
print(
|
||||
"Blender not found. Set ORISYS_BLENDER (or BLENDER_EXE) to blender.exe, "
|
||||
"add Blender to PATH, install under Program Files\\Blender Foundation, "
|
||||
"or set optional 'executable' in blender.lock.json.\n"
|
||||
f"Manual command:\n {manual}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
cmd = [
|
||||
str(blender),
|
||||
"-b",
|
||||
str(blend),
|
||||
"-P",
|
||||
str(script_path),
|
||||
"--",
|
||||
*extra_args,
|
||||
]
|
||||
print("Running:", " ".join(cmd))
|
||||
return subprocess.call(cmd)
|
||||
|
||||
|
||||
def cmd_validate(revision_dir: Path, blender: bool) -> int:
|
||||
paths = RevisionPaths.from_revision_dir(revision_dir)
|
||||
errors = validate_revision_metadata(paths)
|
||||
if errors:
|
||||
for message in errors:
|
||||
print(f"ERROR: {message}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(f"OK: metadata for {paths.root}")
|
||||
if not blender:
|
||||
print("Tip: pass --blender to run in-scene collection checks.")
|
||||
return 0
|
||||
return _run_blender_script("validate.py", paths.source_blend, [str(revision_dir.resolve())])
|
||||
|
||||
|
||||
def cmd_export(revision_dir: Path, profile: str) -> int:
|
||||
root = repo_root(revision_dir)
|
||||
paths = RevisionPaths.from_revision_dir(revision_dir)
|
||||
errors = validate_revision_metadata(paths)
|
||||
if errors:
|
||||
for message in errors:
|
||||
print(f"ERROR: {message}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
load_profile(paths, profile)
|
||||
output_dir = export_output_dir(root, paths, profile)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
job = build_export_job(paths, output_dir, profile)
|
||||
job_path = write_export_job(job, export_revision_dir(root, paths, profile))
|
||||
print(f"Wrote export job: {job_path}")
|
||||
|
||||
code = _run_blender_script(
|
||||
"export.py",
|
||||
paths.source_blend,
|
||||
["--job", str(job_path)],
|
||||
)
|
||||
if code != 0:
|
||||
return code
|
||||
|
||||
print(f"Export target: {output_dir}")
|
||||
missing = []
|
||||
for name in load_profile(paths, profile).get("artifacts", []):
|
||||
if not (output_dir / name).is_file():
|
||||
missing.append(name)
|
||||
if missing:
|
||||
print(
|
||||
f"ERROR: export finished but artifacts are missing: {', '.join(missing)}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 3
|
||||
print("OK: all profile artifacts present")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_manifest(revision_dir: Path, out: Path | None, profile: str) -> int:
|
||||
root = repo_root(revision_dir)
|
||||
paths = RevisionPaths.from_revision_dir(revision_dir)
|
||||
errors = validate_revision_metadata(paths, require_blend=False)
|
||||
if errors:
|
||||
for message in errors:
|
||||
print(f"ERROR: {message}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
manifest = build_manifest(root=root, revision_dir=revision_dir, profile=profile)
|
||||
if out is None:
|
||||
out = default_manifest_path(root, paths, profile)
|
||||
write_manifest(out, manifest)
|
||||
print(f"Wrote {out}")
|
||||
if "bundle_sha256" in manifest:
|
||||
print(f"bundle_sha256: {manifest['bundle_sha256']}")
|
||||
else:
|
||||
print("WARN: export directory empty — run export before promoting", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> None:
|
||||
parser = argparse.ArgumentParser(prog="orisys-design")
|
||||
parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
|
||||
sub = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
validate_p = sub.add_parser("validate", help="Check revision metadata and optional Blender scene")
|
||||
validate_p.add_argument("revision_dir", type=Path)
|
||||
validate_p.add_argument(
|
||||
"--blender",
|
||||
action="store_true",
|
||||
help="Run blender/scripts/validate.py headless",
|
||||
)
|
||||
|
||||
export_p = sub.add_parser("export", help="Export nominal geometry for a profile")
|
||||
export_p.add_argument("revision_dir", type=Path)
|
||||
export_p.add_argument("--profile", default="compute", choices=["compute"])
|
||||
|
||||
manifest_p = sub.add_parser("manifest", help="Write release manifest with file hashes")
|
||||
manifest_p.add_argument("revision_dir", type=Path)
|
||||
manifest_p.add_argument(
|
||||
"--out",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="Manifest path (default: build/export/<sensor>/<revision>/manifest.json)",
|
||||
)
|
||||
manifest_p.add_argument("--profile", default="compute")
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
if args.command == "validate":
|
||||
code = cmd_validate(args.revision_dir, args.blender)
|
||||
elif args.command == "export":
|
||||
code = cmd_export(args.revision_dir, args.profile)
|
||||
elif args.command == "manifest":
|
||||
code = cmd_manifest(args.revision_dir, args.out, args.profile)
|
||||
else:
|
||||
parser.error(f"Unknown command: {args.command}")
|
||||
return
|
||||
|
||||
raise SystemExit(code)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,229 @@
|
||||
"""Release manifest and revision metadata helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RevisionPaths:
|
||||
root: Path
|
||||
design_yaml: Path
|
||||
source_blend: Path
|
||||
profiles_dir: Path
|
||||
|
||||
@classmethod
|
||||
def from_revision_dir(cls, revision_dir: Path) -> RevisionPaths:
|
||||
revision_dir = revision_dir.resolve()
|
||||
design_yaml = revision_dir / "design.yaml"
|
||||
if not design_yaml.is_file():
|
||||
raise FileNotFoundError(f"Missing design.yaml: {design_yaml}")
|
||||
design = load_yaml(design_yaml)
|
||||
blend_rel = design.get("source_blend", "source/assembly.blend")
|
||||
return cls(
|
||||
root=revision_dir,
|
||||
design_yaml=design_yaml,
|
||||
source_blend=(revision_dir / blend_rel).resolve(),
|
||||
profiles_dir=revision_dir / "profiles",
|
||||
)
|
||||
|
||||
|
||||
def repo_root(start: Path | None = None) -> Path:
|
||||
here = (start or Path.cwd()).resolve()
|
||||
for candidate in [here, *here.parents]:
|
||||
if (candidate / "pyproject.toml").is_file() and (candidate / "sensors").is_dir():
|
||||
return candidate
|
||||
raise FileNotFoundError("Could not locate orisys-sensor-design repository root")
|
||||
|
||||
|
||||
def load_yaml(path: Path) -> dict[str, Any]:
|
||||
with path.open(encoding="utf-8") as handle:
|
||||
data = yaml.safe_load(handle)
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(f"Expected mapping in {path}")
|
||||
return data
|
||||
|
||||
|
||||
def git_head_commit(root: Path) -> str | None:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "-C", str(root), "rev-parse", "HEAD"],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
except (OSError, subprocess.CalledProcessError):
|
||||
return None
|
||||
return result.stdout.strip() or None
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
GENERATED_EXPORT_METADATA = frozenset({"_export_job.json", "manifest.json"})
|
||||
|
||||
|
||||
def _is_generated_metadata(path: Path) -> bool:
|
||||
return path.name in GENERATED_EXPORT_METADATA
|
||||
|
||||
|
||||
def sha256_directory(directory: Path) -> str:
|
||||
if not directory.is_dir():
|
||||
raise FileNotFoundError(f"Export directory not found: {directory}")
|
||||
parts: list[str] = []
|
||||
for path in sorted(directory.rglob("*")):
|
||||
if path.is_file() and not _is_generated_metadata(path):
|
||||
rel = path.relative_to(directory).as_posix()
|
||||
parts.append(f"{rel}\0{sha256_file(path)}")
|
||||
digest = hashlib.sha256()
|
||||
digest.update("\n".join(parts).encode("utf-8"))
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def validate_revision_metadata(paths: RevisionPaths, *, require_blend: bool = True) -> list[str]:
|
||||
errors: list[str] = []
|
||||
design = load_yaml(paths.design_yaml)
|
||||
|
||||
for key in ("sensor", "revision", "meters_per_source_unit"):
|
||||
if key not in design:
|
||||
errors.append(f"design.yaml missing required key: {key}")
|
||||
|
||||
if require_blend and not paths.source_blend.is_file():
|
||||
errors.append(f"Source blend not found: {paths.source_blend}")
|
||||
|
||||
if not paths.profiles_dir.is_dir():
|
||||
errors.append(f"Profiles directory not found: {paths.profiles_dir}")
|
||||
else:
|
||||
for name in ("compute.yaml", "studio.yaml"):
|
||||
if not (paths.profiles_dir / name).is_file():
|
||||
errors.append(f"Missing profile: {paths.profiles_dir / name}")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def load_profile(paths: RevisionPaths, profile: str) -> dict[str, Any]:
|
||||
profile_path = paths.profiles_dir / f"{profile}.yaml"
|
||||
if not profile_path.is_file():
|
||||
raise FileNotFoundError(f"Unknown profile: {profile_path}")
|
||||
data = load_yaml(profile_path)
|
||||
if data.get("profile") != profile:
|
||||
raise ValueError(f"Profile name mismatch in {profile_path}")
|
||||
return data
|
||||
|
||||
|
||||
DEFAULT_EXPORT_OBJECTS = {
|
||||
"sensing_surface": "sensing_surface",
|
||||
"cam_sensor": "cam_sensor",
|
||||
"origin_sensor": "origin_sensor",
|
||||
}
|
||||
|
||||
|
||||
def export_object_names(paths: RevisionPaths) -> dict[str, str]:
|
||||
"""Resolve Blender object names for headless export (design.yaml overrides)."""
|
||||
design = load_yaml(paths.design_yaml)
|
||||
objects = design.get("objects") or {}
|
||||
if not isinstance(objects, dict):
|
||||
raise ValueError("design.yaml 'objects' must be a mapping")
|
||||
return {
|
||||
key: str(objects.get(key, default))
|
||||
for key, default in DEFAULT_EXPORT_OBJECTS.items()
|
||||
}
|
||||
|
||||
|
||||
EXPORT_JOB_SCHEMA = "orisys-sensor-design/export-job-v1"
|
||||
EXPORT_JOB_FILENAME = "_export_job.json"
|
||||
MANIFEST_FILENAME = "manifest.json"
|
||||
|
||||
|
||||
def build_export_job(
|
||||
paths: RevisionPaths,
|
||||
output_dir: Path,
|
||||
profile: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Build the JSON job dict that Blender's export.py consumes (stdlib JSON)."""
|
||||
return {
|
||||
"schema": EXPORT_JOB_SCHEMA,
|
||||
"output": str(output_dir.resolve()),
|
||||
"profile": profile,
|
||||
"objects": export_object_names(paths),
|
||||
}
|
||||
|
||||
|
||||
def export_revision_dir(root: Path, paths: RevisionPaths, profile: str) -> Path:
|
||||
"""Return the revision export folder that contains profile payload dirs."""
|
||||
return export_output_dir(root, paths, profile).parent
|
||||
|
||||
|
||||
def default_manifest_path(root: Path, paths: RevisionPaths, profile: str) -> Path:
|
||||
"""Default release manifest path beside profile payload directories."""
|
||||
return export_revision_dir(root, paths, profile) / MANIFEST_FILENAME
|
||||
|
||||
|
||||
def write_export_job(job: dict[str, Any], export_root: Path) -> Path:
|
||||
"""Write _export_job.json beside profile output dirs; return its path."""
|
||||
export_root.mkdir(parents=True, exist_ok=True)
|
||||
job_path = export_root / EXPORT_JOB_FILENAME
|
||||
job_path.write_text(json.dumps(job, indent=2) + "\n", encoding="utf-8")
|
||||
return job_path
|
||||
|
||||
|
||||
def export_output_dir(root: Path, paths: RevisionPaths, profile: str) -> Path:
|
||||
profile_data = load_profile(paths, profile)
|
||||
output_rel = profile_data.get("output_dir")
|
||||
if not output_rel:
|
||||
raise ValueError(f"Profile {profile} missing output_dir")
|
||||
return (root / output_rel).resolve()
|
||||
|
||||
|
||||
def build_manifest(
|
||||
*,
|
||||
root: Path,
|
||||
revision_dir: Path,
|
||||
profile: str = "compute",
|
||||
) -> dict[str, Any]:
|
||||
paths = RevisionPaths.from_revision_dir(revision_dir)
|
||||
design = load_yaml(paths.design_yaml)
|
||||
profile_data = load_profile(paths, profile)
|
||||
export_dir = export_output_dir(root, paths, profile)
|
||||
|
||||
files: dict[str, str] = {}
|
||||
if export_dir.is_dir():
|
||||
for path in sorted(export_dir.rglob("*")):
|
||||
if path.is_file() and not _is_generated_metadata(path):
|
||||
rel = path.relative_to(export_dir).as_posix()
|
||||
files[rel] = sha256_file(path)
|
||||
|
||||
manifest: dict[str, Any] = {
|
||||
"schema": "orisys-sensor-design/manifest-v1",
|
||||
"generated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
|
||||
"design_repo_commit": git_head_commit(root),
|
||||
"sensor": design.get("sensor"),
|
||||
"revision": design.get("revision"),
|
||||
"profile": profile,
|
||||
"meters_per_source_unit": design.get("meters_per_source_unit"),
|
||||
"source_blend": str(paths.source_blend.relative_to(root)).replace("\\", "/"),
|
||||
"export_dir": str(export_dir.relative_to(root)).replace("\\", "/"),
|
||||
"expected_artifacts": profile_data.get("artifacts", []),
|
||||
"files": files,
|
||||
}
|
||||
if export_dir.is_dir():
|
||||
manifest["bundle_sha256"] = sha256_directory(export_dir)
|
||||
return manifest
|
||||
|
||||
|
||||
def write_manifest(path: Path, manifest: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
|
||||
Reference in New Issue
Block a user