"""Closed, profile-bound xArm Lite6 vacuum sampling-tool adapter.

This module is deliberately not a GPIO service.  Channels, polarities, timing,
tool identity, payload and TCP geometry come exclusively from one root-owned,
hash-pinned commissioning profile.  The public surface exposes only typed
``grip``, ``release``, safe-state and diagnostic operations.
"""

from __future__ import annotations

import hashlib
import json
import math
import os
import re
import stat
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping, Optional, Sequence, Tuple

_PROFILE_SCHEMA = "labcobot.sampling-tool-profile.v1"
_PROFILE_KEYS = frozenset(
    {
        "schema",
        "tool_model",
        "tool_serial",
        "controller_serial",
        "safety_profile_sha256",
        "tcp_offset_xyz_mm_rpy_degrees",
        "tcp_load_kg",
        "tcp_center_of_gravity_mm",
        "vacuum_output",
        "grasp_feedback_input",
        "identity_input",
        "blowoff_output",
        "grasp_timeout_seconds",
        "release_timeout_seconds",
        "poll_interval_seconds",
        "blowoff_duration_seconds",
        "outputs_deenergized_safe_on_reboot",
    }
)
_CHANNEL_KEYS = frozenset({"channel", "active_level"})
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:+-]{0,63}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")


class SamplingToolFault(RuntimeError):
    """One bounded production-tool rejection with a closed fault code."""

    def __init__(self, code: str, operation: str) -> None:
        self.code = str(code)[:48]
        self.operation = str(operation)[:64]
        super().__init__("{}: {}".format(self.code, self.operation))


@dataclass(frozen=True)
class DigitalChannel:
    channel: int
    active_level: int

    @property
    def inactive_level(self) -> int:
        return 1 - self.active_level


@dataclass(frozen=True)
class SamplingToolProfile:
    """Immutable physical tool, wiring, feedback and load contract."""

    profile_sha256: str
    tool_model: str
    tool_serial: str
    controller_serial: str
    safety_profile_sha256: str
    tcp_offset_xyz_mm_rpy_degrees: Tuple[float, ...]
    tcp_load_kg: float
    tcp_center_of_gravity_mm: Tuple[float, ...]
    vacuum_output: DigitalChannel
    grasp_feedback_input: DigitalChannel
    identity_input: DigitalChannel
    blowoff_output: Optional[DigitalChannel]
    grasp_timeout_seconds: float
    release_timeout_seconds: float
    poll_interval_seconds: float
    blowoff_duration_seconds: float

    @classmethod
    def from_document(
        cls,
        document: Mapping[str, Any],
        *,
        profile_sha256: str,
    ) -> "SamplingToolProfile":
        if not isinstance(document, Mapping) or set(document) != _PROFILE_KEYS:
            raise ValueError("sampling-tool profile schema is invalid")
        if document.get("schema") != _PROFILE_SCHEMA:
            raise ValueError("sampling-tool profile version is invalid")
        if document.get("outputs_deenergized_safe_on_reboot") is not True:
            raise ValueError("sampling-tool reboot state is not fail-safe")
        if not isinstance(profile_sha256, str) or not _SHA256.fullmatch(profile_sha256):
            raise ValueError("sampling-tool profile SHA-256 is invalid")

        identifiers = {}
        for name in ("tool_model", "tool_serial", "controller_serial"):
            value = document.get(name)
            if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value):
                raise ValueError("sampling-tool {} is invalid".format(name))
            identifiers[name] = value
        safety_digest = document.get("safety_profile_sha256")
        if not isinstance(safety_digest, str) or not _SHA256.fullmatch(safety_digest):
            raise ValueError("sampling-tool safety-profile binding is invalid")

        tcp_offset = _finite_vector(
            document.get("tcp_offset_xyz_mm_rpy_degrees"), 6, "TCP offset"
        )
        tcp_load = _finite_number(document.get("tcp_load_kg"), "TCP load")
        tcp_cog = _finite_vector(
            document.get("tcp_center_of_gravity_mm"), 3, "TCP load center"
        )
        if not 0.0 <= tcp_load <= 2.0:
            raise ValueError("sampling-tool TCP load is outside the reviewed limit")
        if any(abs(value) > 500.0 for value in tcp_offset[:3]) or any(
            abs(value) > 360.0 for value in tcp_offset[3:]
        ):
            raise ValueError("sampling-tool TCP offset is outside reviewed bounds")
        if any(abs(value) > 500.0 for value in tcp_cog):
            raise ValueError("sampling-tool TCP load center is outside reviewed bounds")

        vacuum = _channel(document.get("vacuum_output"), "vacuum output")
        feedback = _channel(
            document.get("grasp_feedback_input"), "grasp feedback input"
        )
        identity = _channel(document.get("identity_input"), "identity input")
        blowoff_raw = document.get("blowoff_output")
        blowoff = (
            None if blowoff_raw is None else _channel(blowoff_raw, "blow-off output")
        )
        if blowoff is not None and blowoff.channel == vacuum.channel:
            raise ValueError("sampling-tool outputs must use distinct channels")
        if identity.channel == feedback.channel:
            raise ValueError("sampling-tool inputs must use distinct channels")

        grasp_timeout = _bounded_duration(
            document.get("grasp_timeout_seconds"),
            "grasp timeout",
            0.1,
            10.0,
        )
        release_timeout = _bounded_duration(
            document.get("release_timeout_seconds"),
            "release timeout",
            0.1,
            10.0,
        )
        poll_interval = _bounded_duration(
            document.get("poll_interval_seconds"),
            "poll interval",
            0.005,
            0.1,
        )
        blowoff_duration = _bounded_duration(
            document.get("blowoff_duration_seconds"),
            "blow-off duration",
            0.0,
            0.5,
        )
        if blowoff is None and blowoff_duration != 0.0:
            raise ValueError("blow-off duration requires a commissioned output")
        if blowoff is not None and blowoff_duration <= 0.0:
            raise ValueError("commissioned blow-off requires a positive duration")

        return cls(
            profile_sha256=profile_sha256,
            safety_profile_sha256=safety_digest,
            tcp_offset_xyz_mm_rpy_degrees=tcp_offset,
            tcp_load_kg=tcp_load,
            tcp_center_of_gravity_mm=tcp_cog,
            vacuum_output=vacuum,
            grasp_feedback_input=feedback,
            identity_input=identity,
            blowoff_output=blowoff,
            grasp_timeout_seconds=grasp_timeout,
            release_timeout_seconds=release_timeout,
            poll_interval_seconds=poll_interval,
            blowoff_duration_seconds=blowoff_duration,
            **identifiers,
        )


def load_sampling_tool_profile(
    path: str,
    *,
    expected_sha256: str,
) -> SamplingToolProfile:
    """Load one bounded, root-owned profile from ``/etc/labcobot``."""

    if not isinstance(expected_sha256, str) or not _SHA256.fullmatch(expected_sha256):
        raise ValueError("sampling-tool profile SHA-256 is invalid")
    source = Path(path)
    root = Path("/etc/labcobot")
    if not source.is_absolute() or ".." in source.parts:
        raise ValueError("sampling-tool profile path is unsafe")
    try:
        relative = source.relative_to(root)
    except ValueError as exc:
        raise ValueError("sampling-tool profile must be below /etc/labcobot") from exc
    if not relative.parts:
        raise ValueError("sampling-tool profile must be below /etc/labcobot")
    metadata = source.lstat()
    if (
        stat.S_ISLNK(metadata.st_mode)
        or not stat.S_ISREG(metadata.st_mode)
        or metadata.st_uid != 0
        or metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH)
    ):
        raise ValueError("sampling-tool profile must be a protected root-owned file")
    descriptor = os.open(str(source), os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    try:
        raw = os.read(descriptor, 32 * 1024 + 1)
        opened = os.fstat(descriptor)
        if len(raw) > 32 * 1024:
            raise ValueError("sampling-tool profile is too large")
        if (
            opened.st_dev != metadata.st_dev
            or opened.st_ino != metadata.st_ino
            or not stat.S_ISREG(opened.st_mode)
        ):
            raise ValueError("sampling-tool profile changed while opening")
    finally:
        os.close(descriptor)
    actual_sha256 = hashlib.sha256(raw).hexdigest()
    if not _constant_time_equal(actual_sha256, expected_sha256):
        raise ValueError("sampling-tool profile SHA-256 does not match")
    try:
        document = json.loads(
            raw.decode("utf-8"),
            object_pairs_hook=_unique_json_object,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("sampling-tool profile is not valid JSON") from exc
    return SamplingToolProfile.from_document(
        document,
        profile_sha256=actual_sha256,
    )


class SamplingToolAdapter:
    """Typed vacuum/grip controller sharing the canonical Lite6 SDK owner."""

    _MOTION_ALLOWED_STATES = frozenset({"safe_idle", "grasp_confirmed"})

    def __init__(
        self,
        motion_owner: Any,
        profile: SamplingToolProfile,
        *,
        safety_profile_sha256: str,
        monotonic: Callable[[], float] = time.monotonic,
        wait: Optional[Callable[[float], bool]] = None,
    ) -> None:
        metric_source = getattr(motion_owner, "metric_source", None)
        if (
            metric_source is None
            or not callable(getattr(metric_source, "is_motion_owner", None))
            or metric_source.is_motion_owner(motion_owner) is not True
        ):
            raise RuntimeError("sampling tool requires the canonical motion owner")
        if not isinstance(profile, SamplingToolProfile):
            raise RuntimeError("sampling-tool profile is invalid")
        if not _constant_time_equal(
            profile.safety_profile_sha256,
            str(safety_profile_sha256),
        ):
            raise RuntimeError("sampling tool does not match the safety profile")
        self.motion_owner = motion_owner
        self.metric_source = metric_source
        self.arm = metric_source.arm
        self.profile = profile
        self._sdk_lock = metric_source._lock
        self._operation_lock = threading.Lock()
        self._state_lock = threading.Lock()
        self._monotonic = monotonic
        self._wait = wait or self._cancel_event_wait
        self._cancel_event = threading.Event()
        self._stop_generation = 0
        self._state = "unavailable"
        self._fault_code = ""
        self._fault_operation = ""
        self._sequence = 0
        self._started = False
        self._process_owner: Any = None
        self._process_control: Optional[SamplingToolProcessControl] = None

    def claim_process_control(self, owner: Any) -> "SamplingToolProcessControl":
        """Issue one in-process typed control to one trusted workflow owner."""

        if owner is None:
            raise RuntimeError("sampling-tool process owner is invalid")
        with self._state_lock:
            if self._process_owner not in (None, owner):
                raise RuntimeError("sampling-tool process control is already claimed")
            if self._process_control is None:
                self._process_owner = owner
                self._process_control = SamplingToolProcessControl(self, owner)
            return self._process_control

    def start(self) -> Mapping[str, Any]:
        """De-energize outputs, then prove exact tool and released feedback."""

        with self._operation_lock:
            generation = self._begin_operation()
            try:
                with self._sdk_lock:
                    self._require_identity_locked()
                    self._safe_outputs_locked(expected_generation=generation)
                    self._require_feedback_locked(grasped=False)
            except Exception as exc:
                self._latch_fault("startup_failed", "start")
                self._best_effort_safe_outputs()
                raise SamplingToolFault("startup_failed", "start") from exc
            with self._state_lock:
                self._started = True
                self._state = "safe_idle"
                self._fault_code = ""
                self._fault_operation = ""
                self._sequence += 1
            return self.status()

    def _grip(self, owner: Any) -> Mapping[str, Any]:
        """Energize vacuum and return only after commissioned feedback."""

        self._require_process_owner(owner)
        with self._operation_lock:
            generation = self._begin_operation(require_started=True)
            with self._state_lock:
                if self._fault_code:
                    raise SamplingToolFault(self._fault_code, self._fault_operation)
                self._state = "acquiring"
                self._sequence += 1
            try:
                with self._sdk_lock:
                    self._require_identity_locked()
                    self._write_output_locked(
                        self.profile.vacuum_output,
                        active=True,
                        expected_generation=generation,
                    )
                self._wait_for_feedback(
                    grasped=True,
                    timeout=self.profile.grasp_timeout_seconds,
                    generation=generation,
                )
                self._mark_operation_complete(generation, "grasp_confirmed")
            except _ToolCancelled as exc:
                self._best_effort_safe_outputs()
                with self._state_lock:
                    if not self._fault_code:
                        self._state = "safe_idle"
                        self._sequence += 1
                raise SamplingToolFault("cancelled", "grip") from exc
            except Exception as exc:
                self._latch_fault("grasp_not_confirmed", "grip")
                self._best_effort_safe_outputs()
                raise SamplingToolFault("grasp_not_confirmed", "grip") from exc
            return self.status()

    def _release(self, owner: Any) -> Mapping[str, Any]:
        """Release, optionally pulse blow-off, and prove feedback inactive."""

        self._require_process_owner(owner)
        with self._operation_lock:
            generation = self._begin_operation(require_started=True)
            with self._state_lock:
                if self._fault_code:
                    raise SamplingToolFault(self._fault_code, self._fault_operation)
                self._state = "releasing"
                self._sequence += 1
            try:
                with self._sdk_lock:
                    self._require_identity_locked()
                    self._write_output_locked(
                        self.profile.vacuum_output,
                        active=False,
                        expected_generation=generation,
                    )
                    if self.profile.blowoff_output is not None:
                        self._write_output_locked(
                            self.profile.blowoff_output,
                            active=True,
                            expected_generation=generation,
                        )
                if self.profile.blowoff_output is not None:
                    if self._wait(self.profile.blowoff_duration_seconds):
                        raise _ToolCancelled()
                    with self._sdk_lock:
                        self._write_output_locked(
                            self.profile.blowoff_output,
                            active=False,
                            expected_generation=generation,
                        )
                self._wait_for_feedback(
                    grasped=False,
                    timeout=self.profile.release_timeout_seconds,
                    generation=generation,
                )
                self._mark_operation_complete(generation, "safe_idle")
            except _ToolCancelled as exc:
                self._best_effort_safe_outputs()
                raise SamplingToolFault("cancelled", "release") from exc
            except Exception as exc:
                self._latch_fault("release_not_confirmed", "release")
                self._best_effort_safe_outputs()
                raise SamplingToolFault("release_not_confirmed", "release") from exc
            return self.status()

    def request_stop(self) -> None:
        """Fence active/future operations without waiting for controller I/O."""

        with self._state_lock:
            self._stop_generation += 1
            self._cancel_event.set()

    def emergency_safe_idle(self) -> Mapping[str, Any]:
        """De-energize outputs for STOP, fault and shutdown paths."""

        self.request_stop()
        error = None
        with self._sdk_lock:
            try:
                self._safe_outputs_locked()
            except Exception as exc:  # The latch is the important durable result.
                error = exc
        with self._state_lock:
            if error is not None:
                self._fault_code = "safe_state_failed"
                self._fault_operation = "emergency_safe_idle"
                self._state = "fault"
            elif not self._fault_code:
                self._state = "safe_idle" if self._started else "unavailable"
            self._sequence += 1
        if error is not None:
            raise SamplingToolFault(
                "safe_state_failed", "emergency_safe_idle"
            ) from error
        return self.status()

    def shutdown(self) -> Mapping[str, Any]:
        self.request_stop()
        # Wait for the cancelled typed operation to leave its bounded owner
        # before publishing CLOSED; no operation can overwrite that final mark.
        with self._operation_lock:
            result = self.emergency_safe_idle()
            with self._state_lock:
                self._started = False
                if not self._fault_code:
                    self._state = "closed"
                self._sequence += 1
            return {**result, **self.status()}

    def acknowledge_supervised_recovery(self) -> Mapping[str, Any]:
        """Clear the latch only after safe outputs and released feedback prove."""

        with self._operation_lock:
            generation = self._begin_operation()
            try:
                with self._sdk_lock:
                    self._require_identity_locked()
                    self._safe_outputs_locked(expected_generation=generation)
                    self._require_feedback_locked(grasped=False)
            except Exception as exc:
                self._latch_fault("recovery_failed", "supervised_recovery")
                raise SamplingToolFault(
                    "recovery_failed", "supervised_recovery"
                ) from exc
            with self._state_lock:
                self._started = True
                self._state = "safe_idle"
                self._fault_code = ""
                self._fault_operation = ""
                self._sequence += 1
            return self.status()

    def require_motion_allowed_locked(self) -> None:
        """Veto motion under the already-held canonical SDK lock."""

        with self._state_lock:
            state = self._state
            fault = self._fault_code
            started = self._started
            process_owner_bound = self._process_owner is not None
        if (
            not started
            or not process_owner_bound
            or fault
            or state not in self._MOTION_ALLOWED_STATES
        ):
            raise SamplingToolFault(fault or "tool_not_ready", "motion_gate")
        try:
            self._require_identity_locked()
            if state == "grasp_confirmed":
                self._require_feedback_locked(grasped=True)
        except Exception as exc:
            code = (
                "grasp_lost"
                if state == "grasp_confirmed"
                else getattr(exc, "code", "tool_gate_failed")
            )
            self._latch_fault(code, "motion_gate")
            try:
                self._safe_outputs_locked()
            except Exception as safe_error:
                self._latch_fault("safe_state_failed", "motion_gate")
                raise SamplingToolFault(
                    "safe_state_failed", "motion_gate"
                ) from safe_error
            raise SamplingToolFault(code, "motion_gate") from exc

    def status(self) -> Mapping[str, Any]:
        motion_owner = self.metric_source.is_motion_owner(self.motion_owner)
        with self._state_lock:
            return {
                "schema_version": 1,
                "configured": True,
                "started": self._started,
                "state": self._state,
                "ready": bool(
                    self._started
                    and self._process_owner is not None
                    and not self._fault_code
                    and self._state in self._MOTION_ALLOWED_STATES
                ),
                "grasp_confirmed": self._state == "grasp_confirmed",
                "fault_code": self._fault_code,
                "fault_operation": self._fault_operation,
                "sequence": self._sequence,
                "tool_model": self.profile.tool_model,
                "tool_serial": self.profile.tool_serial,
                "profile_sha256": self.profile.profile_sha256,
                "motion_owner": motion_owner,
                "process_owner_bound": self._process_owner is not None,
            }

    def _require_process_owner(self, owner: Any) -> None:
        with self._state_lock:
            if owner is None or owner is not self._process_owner:
                raise SamplingToolFault("workflow_owner_mismatch", "tool actuation")

    def _begin_operation(self, *, require_started: bool = False) -> int:
        with self._state_lock:
            if require_started and not self._started:
                raise SamplingToolFault("tool_not_started", "operation")
            generation = self._stop_generation
            self._cancel_event.clear()
            return generation

    def _cancel_event_wait(self, seconds: float) -> bool:
        return self._cancel_event.wait(max(0.0, float(seconds)))

    def _cancelled(self, generation: int) -> bool:
        with self._state_lock:
            return generation != self._stop_generation

    def _mark_operation_complete(self, generation: int, state: str) -> None:
        """Linearize STOP generation with the final successful state mark."""

        with self._state_lock:
            if generation != self._stop_generation:
                raise _ToolCancelled()
            self._state = state
            self._sequence += 1

    def _wait_for_feedback(
        self,
        *,
        grasped: bool,
        timeout: float,
        generation: int,
    ) -> None:
        deadline = float(self._monotonic()) + float(timeout)
        while True:
            if self._cancelled(generation):
                raise _ToolCancelled()
            with self._sdk_lock:
                if self._cancelled(generation):
                    raise _ToolCancelled()
                self._require_identity_locked()
                if self._feedback_locked() is grasped:
                    return
            remaining = deadline - float(self._monotonic())
            if remaining <= 0.0:
                raise TimeoutError("sampling-tool feedback timeout")
            if self._wait(min(self.profile.poll_interval_seconds, remaining)):
                raise _ToolCancelled()

    def _require_identity_locked(self) -> None:
        if (
            not bool(getattr(self.arm, "connected", False))
            or getattr(self.arm, "axis", None) != 6
            or getattr(self.arm, "device_type", None) != 9
            or not _constant_time_equal(
                str(getattr(self.arm, "sn", "")), self.profile.controller_serial
            )
        ):
            raise SamplingToolFault("identity_mismatch", "controller")
        if self._read_input_locked(self.profile.identity_input) is not True:
            raise SamplingToolFault("identity_mismatch", "tool feedback")
        tcp_offset = _runtime_vector(getattr(self.arm, "tcp_offset", None), 6)
        tcp_load = getattr(self.arm, "tcp_load", None)
        if not isinstance(tcp_load, (tuple, list)) or len(tcp_load) != 2:
            raise SamplingToolFault("profile_mismatch", "TCP load")
        load_kg = _finite_number(tcp_load[0], "runtime TCP load")
        load_cog = _runtime_vector(tcp_load[1], 3)
        if (
            not _vectors_match(tcp_offset, self.profile.tcp_offset_xyz_mm_rpy_degrees)
            or not math.isclose(
                load_kg, self.profile.tcp_load_kg, rel_tol=0.0, abs_tol=1e-4
            )
            or not _vectors_match(load_cog, self.profile.tcp_center_of_gravity_mm)
        ):
            raise SamplingToolFault("profile_mismatch", "TCP profile")

    def _feedback_locked(self) -> bool:
        return self._read_input_locked(self.profile.grasp_feedback_input)

    def _require_feedback_locked(self, *, grasped: bool) -> None:
        if self._feedback_locked() is not grasped:
            raise SamplingToolFault("feedback_mismatch", "grasp feedback")

    def _read_input_locked(self, channel: DigitalChannel) -> bool:
        callback = getattr(self.arm, "get_tgpio_digital", None)
        if not callable(callback):
            raise SamplingToolFault("io_unavailable", "read digital input")
        result = callback(ionum=channel.channel)
        if not isinstance(result, (tuple, list)) or len(result) != 2:
            raise SamplingToolFault("io_invalid_response", "read digital input")
        code, value = result
        if type(code) is not int or code != 0 or type(value) not in (int, bool):
            raise SamplingToolFault("io_read_failed", "read digital input")
        level = int(value)
        if level not in (0, 1):
            raise SamplingToolFault("io_invalid_response", "read digital input")
        return level == channel.active_level

    def _write_output_locked(
        self,
        channel: DigitalChannel,
        *,
        active: bool,
        expected_generation: Optional[int] = None,
    ) -> None:
        if expected_generation is not None and self._cancelled(expected_generation):
            raise _ToolCancelled()
        callback = getattr(self.arm, "set_tgpio_digital", None)
        if not callable(callback):
            raise SamplingToolFault("io_unavailable", "write digital output")
        value = channel.active_level if active else channel.inactive_level
        result = callback(ionum=channel.channel, value=value)
        code = result[0] if isinstance(result, (tuple, list)) and result else result
        if type(code) is not int or code != 0:
            raise SamplingToolFault("io_write_failed", "write digital output")

    def _safe_outputs_locked(
        self, *, expected_generation: Optional[int] = None
    ) -> None:
        if self.profile.blowoff_output is not None:
            self._write_output_locked(
                self.profile.blowoff_output,
                active=False,
                expected_generation=expected_generation,
            )
        self._write_output_locked(
            self.profile.vacuum_output,
            active=False,
            expected_generation=expected_generation,
        )

    def _safe_idle_from_owner_locked(self) -> None:
        """Apply safe outputs/state while the motion owner holds the SDK lock."""

        self._safe_outputs_locked()
        with self._state_lock:
            if not self._fault_code:
                if self._started:
                    self._state = "safe_idle"
                elif self._state != "closed":
                    self._state = "unavailable"
            self._sequence += 1

    def _best_effort_safe_outputs(self) -> None:
        try:
            with self._sdk_lock:
                self._safe_outputs_locked()
        except Exception:
            with self._state_lock:
                self._fault_code = "safe_state_failed"
                self._fault_operation = "fault_cleanup"
                self._state = "fault"
                self._sequence += 1

    def _latch_fault(self, code: str, operation: str) -> None:
        with self._state_lock:
            self._stop_generation += 1
            self._cancel_event.set()
            self._fault_code = str(code)[:48]
            self._fault_operation = str(operation)[:64]
            self._state = "fault"
            self._sequence += 1


class _ToolCancelled(RuntimeError):
    pass


class SamplingToolProcessControl:
    """Capability object exposing only production-level tool operations."""

    def __init__(self, adapter: SamplingToolAdapter, owner: Any) -> None:
        self.__adapter = adapter
        self.__owner = owner

    def grip(self) -> Mapping[str, Any]:
        return self.__adapter._grip(self.__owner)

    def release(self) -> Mapping[str, Any]:
        return self.__adapter._release(self.__owner)

    def status(self) -> Mapping[str, Any]:
        return self.__adapter.status()


def _channel(value: Any, name: str) -> DigitalChannel:
    if not isinstance(value, Mapping) or set(value) != _CHANNEL_KEYS:
        raise ValueError("sampling-tool {} is invalid".format(name))
    channel = value.get("channel")
    level = value.get("active_level")
    if type(channel) is not int or not 0 <= channel <= 4:
        raise ValueError("sampling-tool {} channel is invalid".format(name))
    if type(level) is not int or level not in (0, 1):
        raise ValueError("sampling-tool {} polarity is invalid".format(name))
    return DigitalChannel(channel=channel, active_level=level)


def _finite_number(value: Any, name: str) -> float:
    if isinstance(value, bool):
        raise ValueError("{} is invalid".format(name))
    try:
        result = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise ValueError("{} is invalid".format(name)) from exc
    if not math.isfinite(result):
        raise ValueError("{} is invalid".format(name))
    return result


def _finite_vector(value: Any, length: int, name: str) -> Tuple[float, ...]:
    if not isinstance(value, list) or len(value) != length:
        raise ValueError("{} must contain {} values".format(name, length))
    return tuple(_finite_number(item, name) for item in value)


def _runtime_vector(value: Any, length: int) -> Tuple[float, ...]:
    if not isinstance(value, (tuple, list)) or len(value) != length:
        raise SamplingToolFault("profile_mismatch", "runtime vector")
    try:
        result = tuple(float(item) for item in value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise SamplingToolFault("profile_mismatch", "runtime vector") from exc
    if not all(math.isfinite(item) for item in result):
        raise SamplingToolFault("profile_mismatch", "runtime vector")
    return result


def _bounded_duration(value: Any, name: str, minimum: float, maximum: float) -> float:
    result = _finite_number(value, name)
    if not minimum <= result <= maximum:
        raise ValueError("{} is outside reviewed bounds".format(name))
    return result


def _vectors_match(left: Sequence[float], right: Sequence[float]) -> bool:
    return len(left) == len(right) and all(
        math.isclose(a, b, rel_tol=0.0, abs_tol=1e-4) for a, b in zip(left, right)
    )


def _constant_time_equal(left: str, right: str) -> bool:
    import secrets

    return secrets.compare_digest(str(left), str(right))


def _unique_json_object(pairs: Sequence[Tuple[str, Any]]) -> Mapping[str, Any]:
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("sampling-tool profile contains duplicate keys")
        result[key] = value
    return result


__all__ = [
    "DigitalChannel",
    "SamplingToolAdapter",
    "SamplingToolFault",
    "SamplingToolProfile",
    "SamplingToolProcessControl",
    "load_sampling_tool_profile",
]
