"""Interactive action helpers for companion plugins."""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional
from .._action_kind import ActionKind
from .state import CompanionState
if TYPE_CHECKING:
from .viewer_proxy import ViewerProxy
[docs]
class MultiStepAction:
"""Wrap a viewer interactive action that spans multiple input steps."""
def __init__(
self,
action_id: str | ActionKind,
step_hints: list[str],
state: Optional[CompanionState] = None,
runtime: 'ViewerProxy | None' = None,
) -> None:
if not step_hints:
raise ValueError('step_hints must contain at least one entry')
[docs]
self._action_id_raw: str | ActionKind = action_id
[docs]
self.step_hints: list[str] = list(step_hints)
[docs]
self.state: CompanionState = state if state is not None else CompanionState()
[docs]
self._runtime: 'ViewerProxy | None' = runtime
[docs]
self._active: bool = False
[docs]
def _resolve_runtime(
self, runtime: 'ViewerProxy | None'
) -> 'ViewerProxy':
rt = runtime if runtime is not None else self._runtime
if rt is None:
raise RuntimeError(
f"MultiStepAction('{self.action_id}'): no runtime available. "
'Pass runtime= to the constructor or to the method call.'
)
return rt
@property
[docs]
def action_id(self) -> str:
"""Resolved action id (namespaced when runtime is available)."""
if self._runtime is not None:
return self._runtime.resolve_action_id(self._action_id_raw)
raw = self._action_id_raw
if isinstance(raw, ActionKind):
return raw.value
return str(raw)
@property
[docs]
def is_active(self) -> bool:
return self._active
@property
[docs]
def current_step(self) -> int:
return self.state.step
@property
[docs]
def current_hint(self) -> str:
idx = self.state.step
if 0 <= idx < len(self.step_hints):
return self.step_hints[idx]
return ''
@property
[docs]
def total_steps(self) -> int:
return len(self.step_hints)
[docs]
def start(self, runtime: 'ViewerProxy | None' = None, message: str = '') -> None:
rt = self._resolve_runtime(runtime)
self.state.reset()
self._active = True
rt.start_action(self._action_id_raw, message or self.current_hint)
[docs]
def advance(self, runtime: 'ViewerProxy | None' = None) -> bool:
rt = self._resolve_runtime(runtime)
self.state.advance()
if self.state.step >= len(self.step_hints):
self._active = False
return False
rt.start_action(self._action_id_raw, self.current_hint)
return True
[docs]
def cancel(self, runtime: 'ViewerProxy | None' = None, message: str = '') -> None:
rt = self._resolve_runtime(runtime)
self.state.reset()
self._active = False
rt.end_action(message)