""" Implements a Checkbox node.
Checkbox pauses execution of a Sequencer script until the 'check' command is given to it.
That is to say, until the user confirms some condition and allows the script to continue.
.. warning:: This node is intended to use on interactive Sequencer scripts only. Do not
**seq run** a script that contains it. It will hang forever.
"""
import logging
import asyncio
import attr
from .action import Action
from .state import T_STATE
logger = logging.getLogger(__name__)
[docs]@attr.s
class Checkbox(Action):
"""Checkbox ctor.
Checkbox node waits for user to confirm some condition.
Args:
msg(str): Checkbox message
Keyword Args:
id (str): Unique id. If not provided an unique identifier is assigned.
name (str): Node name. If not provided a name is assigned.
"""
# redefines the f attribute (from Action) to a NOP.
f = attr.ib(default=None, init=False)
checkmark = attr.ib(
init=False, repr=False, default=attr.Factory(asyncio.Event)
)
def __attrs_post_init__(self):
"""
setup object.
Sets name, id and allows node to run.
"""
if self.name is None:
self.name = "Checkbox"
super().__attrs_post_init__()
def _check_function_type(self):
"""Converted to NOP"""
return
async def _f(self):
"""Waits for checkmark event"""
self.checkmark.clear()
await self.checkmark.wait()
return True
async def __call__(self, resume=False):
"""
Executes node action.
If the action is a coroutine a task is created and passed to
the asyncio loop for execution.
"""
await self._execute_preamble()
if self.state is T_STATE.FINISHED: # node skipped in preamble
return None
task = asyncio.create_task(self._f())
r = await self._execute_node_task(task)
return r
[docs] def checked(self):
"""Release the checkbox allowing the Sequencer script to continue its execution"""
if self.state == T_STATE.PAUSED:
raise RuntimeError("Cannot check a PAUSED node")
self.checkmark.set()
[docs] @staticmethod
def create(msg, **kw):
"""
Checkbox node constructor.
Args:
msg(str): Message to display along with the checkbox.
*kw: standard sequencer node keywords.
Keyword Args:
id: Node id
name: node name
"""
return Checkbox(description=msg, **kw)