#!/usr/bin/env python
"""Packaged command-line facade for the high-quality camera generator.

Machine-readable job events are written as one compact JSON object per line,
prefixed with ``CWAI_EVENT ``.  Human diagnostics remain on stderr so Unity can
parse stdout without scraping prose emitted by the historical scripts.
"""

from __future__ import annotations

import argparse
import contextlib
import dataclasses
import _thread
import importlib
import io
import json
import os
import platform
import sys
import threading
import time
import traceback
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, TextIO

import cw_camera_runtime


WORKER_VERSION = cw_camera_runtime.WORKER_VERSION
EVENT_SCHEMA_VERSION = "cw-camera-worker-event-v1"
PROTOCOL_VERSION = "1"
EVENT_PREFIX = "CWAI_EVENT "
EXECUTABLE_NAME = "CWCameraWorker.exe"
DEFAULT_HEARTBEAT_SECONDS = 5.0

GENERATION_STAGES = (
    "catalog_and_policy",
    "preparation_cache_lookup",
    "target_preparation",
    "template_preparation",
    "model_audio_preparation",
    "shot_planning",
    "candidate_cache_lookup",
    "candidate_retarget_and_safety",
    "candidate_cache_publish",
    "output_serialization",
    "metadata_assembly",
    "metadata_write",
)
PREPARATION_STAGES = (
    "catalog_and_policy",
    "cache_lookup",
    "load_prepared_songs",
    "build_template_library",
    "fit_shot_classifier",
    "fit_hierarchical_planner",
    "build_beat_onset_cache",
    "verify_build_inputs_stable",
    "publish_and_verify",
)


class WorkerUsageError(ValueError):
    """Raised when wrapper-level command arguments are malformed."""


class WorkerCancelled(RuntimeError):
    """Raised when Unity creates the cancellation sentinel for this job."""


def _encoding_safe_text(stream: TextIO, value: object) -> str:
    """Return text that a strict, locale-encoded stream can always write."""

    text = str(value)
    encoding = getattr(stream, "encoding", None) or "ascii"
    try:
        return text.encode(encoding, errors="backslashreplace").decode(encoding)
    except LookupError:
        return text.encode("ascii", errors="backslashreplace").decode("ascii")


def _write_text_safely(stream: TextIO, value: object) -> None:
    """Write diagnostics without letting the host code page hide an error."""

    text = str(value)
    try:
        stream.write(text)
    except UnicodeEncodeError:
        stream.write(_encoding_safe_text(stream, text))
    stream.flush()


def _write_line_safely(stream: TextIO, value: object) -> None:
    _write_text_safely(stream, str(value) + "\n")


class CancellationGuard:
    def __init__(self, cancel_file: str = "") -> None:
        self.path = Path(cancel_file).resolve() if cancel_file else None
        self.requested = False

    def poll(self) -> bool:
        if self.requested:
            return True
        self.requested = bool(self.path is not None and self.path.is_file())
        return self.requested

    def raise_if_cancelled(self) -> None:
        if self.poll():
            raise WorkerCancelled("Camera worker cancellation was requested.")


class EventEmitter:
    """Write ordered, flush-on-line JSON events for one worker job."""

    def __init__(
        self,
        command: str,
        job_id: str,
        stream: TextIO | None = None,
        clock: Callable[[], float] = time.perf_counter,
    ) -> None:
        self.command = str(command)
        self.job_id = str(job_id)
        self.stream = stream if stream is not None else sys.stdout
        self._clock = clock
        self._started = float(clock())
        self._sequence = 0
        self._lock = threading.Lock()

    @property
    def elapsed_seconds(self) -> float:
        return max(0.0, float(self._clock()) - self._started)

    def emit(self, event_type: str, **fields: Any) -> dict[str, Any]:
        with self._lock:
            if event_type == "progress":
                if "phase" not in fields and "stage" in fields:
                    fields["phase"] = fields["stage"]
                if "progress" not in fields and "fraction" in fields:
                    fields["progress"] = fields["fraction"]
            payload = {
                "schemaVersion": EVENT_SCHEMA_VERSION,
                "protocolVersion": PROTOCOL_VERSION,
                "sequence": self._sequence,
                "jobId": self.job_id,
                "command": self.command,
                "type": str(event_type),
                "timestampUtc": datetime.now(timezone.utc).isoformat(),
                "elapsedSeconds": self.elapsed_seconds,
                **fields,
            }
            self._sequence += 1
            line = EVENT_PREFIX + json.dumps(
                payload,
                # JSON escapes preserve the original Unicode value while
                # keeping the wire protocol safe for Windows code pages such
                # as CP949. Unity's JSON parser restores the escaped text.
                ensure_ascii=True,
                allow_nan=False,
                separators=(",", ":"),
            )
            _write_line_safely(self.stream, line)
            return payload


class Heartbeat:
    """Emit liveness events while a long native-library stage is silent."""

    def __init__(
        self,
        emitter: EventEmitter,
        interval_seconds: float,
        cancellation: CancellationGuard | None = None,
    ) -> None:
        self._emitter = emitter
        self._interval = max(0.2, float(interval_seconds))
        self._cancellation = cancellation
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def __enter__(self) -> "Heartbeat":
        self._thread = threading.Thread(
            target=self._run,
            name="CWCameraWorkerHeartbeat",
            daemon=True,
        )
        self._thread.start()
        return self

    def __exit__(self, *_: object) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=self._interval + 1.0)

    def _run(self) -> None:
        while not self._stop.wait(self._interval):
            if self._cancellation is not None and self._cancellation.poll():
                _thread.interrupt_main()
                return
            self._emitter.emit("heartbeat")


@contextlib.contextmanager
def temporary_argv(program: str, values: Iterable[str]):
    previous = sys.argv
    sys.argv = [program, *list(values)]
    try:
        yield
    finally:
        sys.argv = previous


def install_legacy_pickle_aliases(generator_module: Any) -> None:
    """Allow trusted caches written by direct ``generate_hybrid.py`` runs.

    Older automatic preparation bundles pickle ``ShotTemplate`` as
    ``__main__.ShotTemplate``.  The packaged entry point has a different
    ``__main__`` module, so expose generator dataclasses there before loading
    the already hash-validated local cache.
    """

    main_module = sys.modules.get("__main__")
    if main_module is None:
        return
    for name, value in vars(generator_module).items():
        if not isinstance(value, type) or not dataclasses.is_dataclass(value):
            continue
        if getattr(value, "__module__", "") not in {
            generator_module.__name__,
            "__main__",
        }:
            continue
        setattr(main_module, name, value)


def discover_runtime_root(explicit: str = "") -> Path:
    """Find the sidecar data/model root without embedding machine paths."""

    candidates: list[Path] = []
    if explicit:
        candidates.append(Path(explicit))
    environment = os.environ.get("CWAI_RUNTIME_ROOT", "").strip()
    if environment:
        candidates.append(Path(environment))
    executable = Path(sys.executable).resolve()
    candidates.extend((Path.cwd(), executable.parent))
    # Frozen utility layout: CameraAI~/Tools~/CWCameraWorker/CWCameraWorker.exe
    # ships its read-only RuntimeData~ sibling two levels above the executable.
    candidates.extend(
        (
            executable.parent.parent.parent / "RuntimeData~",
            executable.parent.parent / "RuntimeData~",
        )
    )
    if not cw_camera_runtime.is_frozen_runtime():
        candidates.append(Path(__file__).resolve().parents[2])
    resolved: list[Path] = []
    for candidate in candidates:
        path = candidate.resolve()
        if path not in resolved:
            resolved.append(path)
    for path in resolved:
        if (
            (path / "DatasetExports").is_dir()
            or (path / "reports" / "training_index.json").is_file()
        ):
            return path
    return resolved[0] if resolved else Path.cwd().resolve()


def default_cache_root(runtime_root: Path) -> Path:
    override = os.environ.get("CWAI_CACHE_ROOT", "").strip()
    if override:
        return Path(override).resolve()
    resolved = Path(runtime_root).resolve()
    if resolved.name == "RuntimeData~":
        return Path.cwd().resolve() / ".cw-ai-cache"
    return resolved / ".cache"


def has_option(values: list[str], option: str) -> bool:
    return any(value == option or value.startswith(option + "=") for value in values)


def option_value(values: list[str], option: str) -> str | None:
    result: str | None = None
    for index, value in enumerate(values):
        if value.startswith(option + "="):
            result = value.split("=", 1)[1]
        elif value == option and index + 1 < len(values):
            result = values[index + 1]
    return result


def augment_generate_arguments(values: list[str], runtime_root: Path) -> list[str]:
    """Supply portable sidecar defaults while preserving explicit options."""

    result = list(values)
    cache_root = default_cache_root(runtime_root)
    defaults = (
        ("--training-index", runtime_root / "reports" / "training_index.json"),
        ("--cut-ranker-model", runtime_root / "models" / "cut_ranker_v2.json"),
        ("--cache-dir", cache_root / "camera_director"),
        ("--preparation-cache-root", cache_root / "hybrid_preparation"),
        ("--candidate-cache-root", cache_root / "hybrid_candidates"),
        (
            "--single-shot-cache-root",
            cache_root / "hybrid_single_shot_candidates",
        ),
    )
    for option, path in defaults:
        if not has_option(result, option):
            result.extend((option, str(path)))
    dataset_root = runtime_root / "DatasetExports"
    if not has_option(result, "--data") and dataset_root.is_dir():
        result.extend(("--data", str(dataset_root)))
    return result


def augment_prepare_arguments(values: list[str], runtime_root: Path) -> list[str]:
    result = list(values)
    cache_root = default_cache_root(runtime_root)
    defaults = (
        ("--training-index", runtime_root / "reports" / "training_index.json"),
        ("--cache-dir", cache_root / "camera_director"),
        ("--output-root", cache_root / "hybrid_preparation"),
    )
    for option, path in defaults:
        if not has_option(result, option):
            result.extend((option, str(path)))
    dataset_root = runtime_root / "DatasetExports"
    if not has_option(result, "--data") and dataset_root.is_dir():
        result.extend(("--data", str(dataset_root)))
    return result


def extract_worker_options(values: list[str]) -> tuple[dict[str, Any], list[str]]:
    """Remove worker-only options before forwarding generator arguments."""

    options: dict[str, Any] = {
        "job_id": uuid.uuid4().hex,
        "runtime_root": "",
        "heartbeat_seconds": DEFAULT_HEARTBEAT_SECONDS,
        "cancel_file": "",
        "event_format": "jsonl",
    }
    remaining: list[str] = []
    aliases = {
        "--job-id": "job_id",
        "--worker-job-id": "job_id",
        "--runtime-root": "runtime_root",
        "--heartbeat-seconds": "heartbeat_seconds",
        "--cancel-file": "cancel_file",
        "--event-format": "event_format",
    }
    index = 0
    while index < len(values):
        value = values[index]
        matched = False
        for flag, name in aliases.items():
            if value == flag:
                if index + 1 >= len(values):
                    raise WorkerUsageError(f"{flag} requires a value.")
                options[name] = values[index + 1]
                index += 2
                matched = True
                break
            if value.startswith(flag + "="):
                options[name] = value.split("=", 1)[1]
                index += 1
                matched = True
                break
        if matched:
            continue
        remaining.append(value)
        index += 1
    try:
        options["heartbeat_seconds"] = float(options["heartbeat_seconds"])
    except (TypeError, ValueError) as error:
        raise WorkerUsageError("--heartbeat-seconds must be numeric.") from error
    if options["heartbeat_seconds"] <= 0:
        raise WorkerUsageError("--heartbeat-seconds must be positive.")
    if not str(options["job_id"]).strip():
        raise WorkerUsageError("--job-id must not be empty.")
    if str(options["event_format"]).strip().lower() != "jsonl":
        raise WorkerUsageError("--event-format currently supports only 'jsonl'.")
    return options, remaining


def _forward_legacy_tail(captured: io.StringIO) -> None:
    lines = [line for line in captured.getvalue().splitlines() if line.strip()]
    if lines:
        _write_line_safely(sys.stderr, "[legacy] " + lines[-1])


def _reporting_generation_timer(
    generator: Any,
    emitter: EventEmitter,
    cancellation: CancellationGuard,
) -> type:
    original = generator.SequentialStageTimer

    class ReportingSequentialStageTimer(original):
        def mark(self, completed_stage: str) -> float:
            elapsed = super().mark(completed_stage)
            cancellation.raise_if_cancelled()
            try:
                index = GENERATION_STAGES.index(completed_stage) + 1
            except ValueError:
                index = 0
            emitter.emit(
                "progress",
                stage=completed_stage,
                stageSeconds=elapsed,
                completedStages=index,
                totalStages=len(GENERATION_STAGES),
                fraction=(index / len(GENERATION_STAGES) if index else None),
            )
            return elapsed

    return ReportingSequentialStageTimer


def run_generate(values: list[str]) -> int:
    if values in (("--help",), ("-h",)):
        values = list(values)
    options, forwarded = extract_worker_options(list(values))
    runtime_root = discover_runtime_root(str(options["runtime_root"]))
    forwarded = augment_generate_arguments(forwarded, runtime_root)
    emitter = EventEmitter("generate", str(options["job_id"]))
    cancellation = CancellationGuard(str(options["cancel_file"]))
    emitter.emit(
        "started",
        workerVersion=WORKER_VERSION,
        runtimeRoot=str(runtime_root),
    )
    original_timer: type | None = None
    captured = io.StringIO()
    try:
        cancellation.raise_if_cancelled()
        with Heartbeat(
            emitter,
            float(options["heartbeat_seconds"]),
            cancellation,
        ):
            emitter.emit("progress", stage="loading_runtime", fraction=0.0)
            generator = importlib.import_module("generate_hybrid")
            install_legacy_pickle_aliases(generator)
            original_timer = generator.SequentialStageTimer
            generator.SequentialStageTimer = _reporting_generation_timer(
                generator,
                emitter,
                cancellation,
            )
            with temporary_argv("generate_hybrid.py", forwarded):
                with contextlib.redirect_stdout(captured):
                    generator.main()
        _forward_legacy_tail(captured)
        output_value = option_value(forwarded, "--output")
        output = Path(output_value).resolve() if output_value else None
        metadata: dict[str, Any] = {}
        if output is not None and (output / "metadata.json").is_file():
            metadata = json.loads(
                (output / "metadata.json").read_text(encoding="utf-8")
            )
        emitter.emit(
            "result",
            success=True,
            outputPath=str(output) if output is not None else "",
            outputDirectory=str(output) if output is not None else "",
            metadataPath=(
                str(output / "metadata.json") if output is not None else ""
            ),
            frameCount=metadata.get("frameCount"),
            shotCount=metadata.get("shotCount"),
            seed=metadata.get("seed"),
        )
        return 0
    except SystemExit as error:
        code = int(error.code or 0)
        if code == 0:
            emitter.emit("result", success=True, help=True)
            return 0
        emitter.emit(
            "error",
            success=False,
            code="invalid_arguments",
            message=f"Generator argument parsing failed with exit code {code}.",
        )
        return code
    except (KeyboardInterrupt, WorkerCancelled):
        emitter.emit(
            "cancelled",
            success=False,
            message="Camera generation was cancelled.",
        )
        return 130
    except Exception as error:
        if captured.getvalue():
            _forward_legacy_tail(captured)
        emitter.emit(
            "error",
            success=False,
            code=type(error).__name__,
            message=str(error),
        )
        _write_text_safely(sys.stderr, traceback.format_exc())
        return 1
    finally:
        if original_timer is not None:
            generator.SequentialStageTimer = original_timer


def _reporting_preparation_timer(
    cache_module: Any,
    emitter: EventEmitter,
    cancellation: CancellationGuard,
) -> type:
    original = cache_module.StageTimer

    class ReportingStageTimer(original):
        @contextlib.contextmanager
        def stage(self, name: str):
            with super().stage(name):
                yield
            cancellation.raise_if_cancelled()
            try:
                index = PREPARATION_STAGES.index(name) + 1
            except ValueError:
                index = 0
            emitter.emit(
                "progress",
                stage=name,
                completedStages=index,
                totalStages=len(PREPARATION_STAGES),
                fraction=(index / len(PREPARATION_STAGES) if index else None),
            )

    return ReportingStageTimer


def run_prepare_cache(values: list[str]) -> int:
    options, forwarded = extract_worker_options(list(values))
    runtime_root = discover_runtime_root(str(options["runtime_root"]))
    forwarded = augment_prepare_arguments(forwarded, runtime_root)
    emitter = EventEmitter("prepare-cache", str(options["job_id"]))
    cancellation = CancellationGuard(str(options["cancel_file"]))
    emitter.emit(
        "started",
        workerVersion=WORKER_VERSION,
        runtimeRoot=str(runtime_root),
    )
    original_timer: type | None = None
    try:
        cancellation.raise_if_cancelled()
        with Heartbeat(
            emitter,
            float(options["heartbeat_seconds"]),
            cancellation,
        ):
            emitter.emit("progress", stage="loading_runtime", fraction=0.0)
            generator = importlib.import_module("generate_hybrid")
            install_legacy_pickle_aliases(generator)
            builder = importlib.import_module("build_hybrid_preparation_cache")
            cache_module = builder.preparation_cache
            original_timer = cache_module.StageTimer
            cache_module.StageTimer = _reporting_preparation_timer(
                cache_module,
                emitter,
                cancellation,
            )
            with temporary_argv(
                "build_hybrid_preparation_cache.py",
                forwarded,
            ):
                args = builder.parse_args()
                report = builder.run(args)
        emitter.emit(
            "result",
            success=True,
            cacheHit=bool(report.get("cacheHit")),
            cacheKey=report.get("cacheKey"),
            cachePath=report.get("cachePath"),
            targetSongId=report.get("targetSongId"),
        )
        return 0
    except SystemExit as error:
        code = int(error.code or 0)
        if code == 0:
            emitter.emit("result", success=True, help=True)
            return 0
        emitter.emit(
            "error",
            success=False,
            code="invalid_arguments",
            message=f"Cache builder argument parsing failed with exit code {code}.",
        )
        return code
    except (KeyboardInterrupt, WorkerCancelled):
        emitter.emit(
            "cancelled",
            success=False,
            message="Preparation-cache build was cancelled.",
        )
        return 130
    except Exception as error:
        emitter.emit(
            "error",
            success=False,
            code=type(error).__name__,
            message=str(error),
        )
        _write_text_safely(sys.stderr, traceback.format_exc())
        return 1
    finally:
        if original_timer is not None:
            cache_module.StageTimer = original_timer


def version_payload() -> dict[str, Any]:
    build_identity = cw_camera_runtime.load_build_identity()
    return {
        "name": "CWCameraWorker",
        "executableName": EXECUTABLE_NAME,
        "workerVersion": WORKER_VERSION,
        "eventSchemaVersion": EVENT_SCHEMA_VERSION,
        "protocolVersion": PROTOCOL_VERSION,
        "python": platform.python_version(),
        "frozen": cw_camera_runtime.is_frozen_runtime(),
        "buildIdentitySchemaVersion": build_identity.get("schemaVersion"),
        "buildCreatedUtc": build_identity.get("createdUtc"),
    }


def run_version(values: list[str]) -> int:
    parser = argparse.ArgumentParser(prog="CWCameraWorker version")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(values)
    payload = version_payload()
    if args.json:
        _write_line_safely(
            sys.stdout,
            json.dumps(
                payload,
                ensure_ascii=True,
                allow_nan=False,
                separators=(",", ":"),
            ),
        )
    else:
        _write_line_safely(
            sys.stdout,
            f"CWCameraWorker {payload['workerVersion']}",
        )
    return 0


def dependency_checks() -> list[dict[str, Any]]:
    checks: list[dict[str, Any]] = []
    for name in (
        "numpy",
        "scipy",
        "sklearn",
        "librosa",
        "joblib",
        "soundfile",
    ):
        try:
            module = importlib.import_module(name)
            checks.append(
                {
                    "name": "dependency:" + name,
                    "ok": True,
                    "detail": str(getattr(module, "__version__", "available")),
                }
            )
        except Exception as error:
            checks.append(
                {
                    "name": "dependency:" + name,
                    "ok": False,
                    "detail": f"{type(error).__name__}: {error}",
                }
            )
    try:
        generator = importlib.import_module("generate_hybrid")
        install_legacy_pickle_aliases(generator)
        importlib.import_module("build_hybrid_preparation_cache")
        checks.append(
            {
                "name": "workerModules",
                "ok": True,
                "detail": "generator and cache builder imported",
            }
        )
    except Exception as error:
        checks.append(
            {
                "name": "workerModules",
                "ok": False,
                "detail": f"{type(error).__name__}: {error}",
            }
        )
    return checks


def doctor_payload(runtime_root: Path) -> dict[str, Any]:
    runtime_root = Path(runtime_root).resolve()
    training_index_path = runtime_root / "reports" / "training_index.json"
    resources = (
        ("datasetRoot", runtime_root / "DatasetExports", "directory"),
        (
            "trainingIndex",
            training_index_path,
            "file",
        ),
        (
            "cutRankerModel",
            runtime_root / "models" / "cut_ranker_v2.json",
            "file",
        ),
    )
    checks = dependency_checks()
    for name, path, kind in resources:
        ok = path.is_dir() if kind == "directory" else path.is_file()
        checks.append({"name": name, "ok": ok, "detail": str(path)})
    if training_index_path.is_file():
        try:
            generator = importlib.import_module("generate_hybrid")
            training_index = json.loads(
                training_index_path.read_text(encoding="utf-8")
            )
            catalog_path = generator.validate_training_index_catalog_freshness(
                training_index_path,
                training_index,
            )
        except Exception as error:
            checks.append(
                {
                    "name": "trainingIndexCatalogFreshness",
                    "ok": False,
                    "detail": f"{type(error).__name__}: {error}",
                }
            )
        else:
            checks.append(
                {
                    "name": "trainingIndexCatalogFreshness",
                    "ok": True,
                    "detail": str(catalog_path),
                }
            )
    if cw_camera_runtime.is_frozen_runtime():
        identity = cw_camera_runtime.load_build_identity()
        checks.append(
            {
                "name": "buildIdentity",
                "ok": bool(identity),
                "detail": str(identity.get("schemaVersion") or "missing"),
            }
        )
    return {
        **version_payload(),
        "runtimeRoot": str(runtime_root),
        "ok": all(bool(check["ok"]) for check in checks),
        "checks": checks,
    }


def run_doctor(values: list[str]) -> int:
    parser = argparse.ArgumentParser(prog="CWCameraWorker doctor")
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--runtime-root", default="")
    args = parser.parse_args(values)
    payload = doctor_payload(discover_runtime_root(args.runtime_root))
    if args.json:
        _write_line_safely(
            sys.stdout,
            json.dumps(
                payload,
                ensure_ascii=True,
                allow_nan=False,
                separators=(",", ":"),
            ),
        )
    else:
        _write_line_safely(
            sys.stdout,
            "CWCameraWorker doctor: " + ("OK" if payload["ok"] else "FAILED"),
        )
        for check in payload["checks"]:
            marker = "OK" if check["ok"] else "FAIL"
            _write_line_safely(
                sys.stdout,
                f"[{marker}] {check['name']}: {check['detail']}",
            )
    return 0 if payload["ok"] else 2


def print_help() -> None:
    print(
        "Usage: CWCameraWorker <command> [options]\n\n"
        "Commands:\n"
        "  generate       Generate a complete camera edit or one-shot composite.\n"
        "  prepare-cache  Build the high-quality preparation cache.\n"
        "  version        Print worker version (`version --json` for JSON).\n"
        "  doctor         Validate dependencies and runtime sidecar assets.\n\n"
        "Worker options for generate/prepare-cache:\n"
        "  --runtime-root PATH       DatasetExports/models/reports sidecar root.\n"
        "  --job-id ID               Stable Unity job correlation ID.\n"
        "  --cancel-file PATH         Sentinel created by Unity to cancel.\n"
        "  --heartbeat-seconds N     Liveness event interval (default 5)."
    )


def main(argv: list[str] | None = None) -> int:
    values = list(sys.argv[1:] if argv is None else argv)
    if not values or values[0] in {"-h", "--help", "help"}:
        print_help()
        return 0
    command, command_values = values[0], values[1:]
    try:
        if command == "generate":
            return run_generate(command_values)
        if command == "prepare-cache":
            return run_prepare_cache(command_values)
        if command == "version":
            return run_version(command_values)
        if command == "doctor":
            return run_doctor(command_values)
        raise WorkerUsageError(f"Unknown command: {command}")
    except WorkerUsageError as error:
        _write_line_safely(sys.stderr, f"CWCameraWorker: {error}")
        return 2


if __name__ == "__main__":
    raise SystemExit(main())
