Source code for semi_cr.core.station.lifecycle

"""Independent, retryable attachment state for live laboratory instruments."""

from __future__ import annotations

import threading
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from collections.abc import Callable




class AttachmentState(StrEnum):
    DETACHED = "detached"
    ATTACHING = "attaching"
    ATTACHED = "attached"
    DETACHING = "detaching"
    FAILED = "failed"





@dataclass(frozen=True)
class InstrumentSnapshot:
    name: str
    state: AttachmentState
    error: str | None
    revision: int





class InstrumentHandle[ResourceT]:
    """Own one live resource without allowing its failure to escape startup.

    Calls can be synchronous for scripts/tests or run on daemon threads for a GUI.
    A generation counter makes an attach result stale when detach is requested while
    the connection attempt is still in flight.
    """

    def __init__(
        self,
        name: str,
        attach: Callable[[], ResourceT],
        detach: Callable[[ResourceT], None],
    ) -> None:
        self.name = name
        self._attach_resource = attach
        self._detach_resource = detach
        self._state = AttachmentState.DETACHED
        self._resource: ResourceT | None = None
        self._error: str | None = None
        self._generation = 0
        self._revision = 0
        self._lock = threading.Lock()
        self._threads: set[threading.Thread] = set()

    @property
    def snapshot(self) -> InstrumentSnapshot:
        with self._lock:
            return self._snapshot_locked()

    @property
    def resource(self) -> ResourceT | None:
        with self._lock:
            return self._resource if self._state is AttachmentState.ATTACHED else None

    @property
    def has_resource(self) -> bool:
        """Whether a resource still needs releasing, including after a failed detach."""
        with self._lock:
            return self._resource is not None



    def require_resource(self) -> ResourceT:
        resource = self.resource
        if resource is None:
            snapshot = self.snapshot
            detail = f": {snapshot.error}" if snapshot.error else ""
            raise RuntimeError(f"{self.name} is {snapshot.state.value}{detail}")
        return resource




    def attach(self) -> InstrumentSnapshot:
        generation = self._begin_attach()
        if generation is None:
            return self.snapshot
        return self._complete_attach(generation)




    def detach(self) -> InstrumentSnapshot:
        pending = self._begin_detach()
        if pending is None:
            return self.snapshot
        generation, resource = pending
        return self._complete_detach(generation, resource)




    def attach_async(self) -> bool:
        generation = self._begin_attach()
        if generation is None:
            return False
        self._start_worker(self._complete_attach, generation)
        return True




    def detach_async(self) -> bool:
        pending = self._begin_detach()
        if pending is None:
            return False
        generation, resource = pending
        self._start_worker(self._complete_detach, generation, resource)
        return True




    def mark_failed(self, error: Exception | str) -> InstrumentSnapshot:
        """Invalidate a live resource after a runtime communication failure."""
        with self._lock:
            self._generation += 1
            resource = self._resource
            self._resource = None
            self._transition_locked(AttachmentState.FAILED, str(error))
            snapshot = self._snapshot_locked()
        if resource is not None:
            self._start_worker(self._release_stale_resource, resource)
        return snapshot




    def wait(self, timeout: float | None = None) -> None:
        """Wait for workers that exist when this method is called."""
        with self._lock:
            threads = tuple(self._threads)
        for thread in threads:
            thread.join(timeout=timeout)


    def _snapshot_locked(self) -> InstrumentSnapshot:
        return InstrumentSnapshot(self.name, self._state, self._error, self._revision)

    def _transition_locked(self, state: AttachmentState, error: str | None = None) -> None:
        self._state = state
        self._error = error
        self._revision += 1

    def _begin_attach(self) -> int | None:
        with self._lock:
            if self._resource is not None or self._state in {AttachmentState.ATTACHING, AttachmentState.DETACHING}:
                return None
            self._generation += 1
            self._transition_locked(AttachmentState.ATTACHING)
            return self._generation

    def _complete_attach(self, generation: int) -> InstrumentSnapshot:
        try:
            resource = self._attach_resource()
        except Exception as error:
            with self._lock:
                if generation == self._generation:
                    self._resource = None
                    self._transition_locked(AttachmentState.FAILED, str(error))
                return self._snapshot_locked()

        with self._lock:
            stale = generation != self._generation
            if not stale:
                self._resource = resource
                self._transition_locked(AttachmentState.ATTACHED)
                return self._snapshot_locked()

        # Detach was requested before the slow attach completed. Release the
        # stale resource without changing the newer state.
        self._release_stale_resource(resource)
        return self.snapshot

    def _begin_detach(self) -> tuple[int, ResourceT] | None:
        with self._lock:
            if self._state is AttachmentState.DETACHING:
                return None
            if self._resource is None and self._state is AttachmentState.DETACHED:
                return None
            self._generation += 1
            generation = self._generation
            resource = self._resource
            self._resource = None
            if resource is None:
                self._transition_locked(AttachmentState.DETACHED)
                return None
            self._transition_locked(AttachmentState.DETACHING)
            return generation, resource

    def _complete_detach(self, generation: int, resource: ResourceT) -> InstrumentSnapshot:
        try:
            self._detach_resource(resource)
        except Exception as error:
            with self._lock:
                if generation == self._generation:
                    self._resource = resource
                    self._transition_locked(AttachmentState.FAILED, f"detach failed: {error}")
                return self._snapshot_locked()

        with self._lock:
            if generation == self._generation:
                self._transition_locked(AttachmentState.DETACHED)
            return self._snapshot_locked()

    def _start_worker(self, target: Callable[..., object], *args: object) -> None:
        def run() -> None:
            try:
                target(*args)
            finally:
                with self._lock:
                    self._threads.discard(thread)

        thread = threading.Thread(target=run, name=f"{self.name}-lifecycle", daemon=True)
        with self._lock:
            self._threads.add(thread)
        thread.start()

    def _release_stale_resource(self, resource: ResourceT) -> None:
        try:
            self._detach_resource(resource)
        except Exception:
            pass