# coding: utf-8
from time import time, sleep
from typing import Callable, Union, Dict, Optional
from re import split, IGNORECASE, match
import logging
from multiprocessing import current_process
from .meta_path import MetaPath
ConditionType = Callable[[Dict[str, list]], bool]
[docs]
class Path(metaclass=MetaPath):
"""Base class for all the Generator Path objects.
The Path object are used by the :class:`~crappy.blocks.Generator` Block to
generate signals.
.. versionadded:: 1.4.0
"""
t0: Optional[float] = None
last_cmd: Optional[float] = None
[docs]
def __init__(self, *_, **__) -> None:
"""Here, the arguments given to the Path should be handled.
If the Path accepts one or more conditions, (e.g. stop conditions for
switching to the next
:class:`~crappy.blocks.generator_path.meta_path.Path`), they can be parsed
here using the :meth:`parse_condition` method. See the code of
:class:`~crappy.blocks.generator_path.Constant` for an example.
The ``self.t0`` attribute stores the time when the last command of the
previous :class:`~crappy.blocks.generator_path.meta_path.Path` was sent,
and the ``self.last_cmd`` stores the value of the last command of the
previous :class:`~crappy.blocks.generator_path.meta_path.Path`.
.. versionchanged:: 1.5.10 renamed *time* argument to *_last_time*
.. versionchanged:: 1.5.10 renamed *cmd* argument to *_last_cmd*
.. versionremoved:: 2.0.0 *_last_time* and *_last_cmd* arguments
"""
self._logger: Optional[logging.Logger] = None
[docs]
def get_cmd(self, data: Dict[str, list]) -> Optional[float]:
"""This method is called by the :class:`~crappy.blocks.Generator` Block to
get the next command to send.
It takes as input a :obj:`dict` containing the data received by the
Generator Block since the last command was sent. Refer to
:meth:`~crappy.blocks.Block.recv_all_data` for more information on the
format of this dict.
This method should output the next command that will be sent by the
Generator Block, as a numeric value. This value can be calculated based on
one or several of :
* the input data (see the code of
:class:`~crappy.blocks.generator_path.Conditional`)
* the current time and the ``self.t0`` attribute (see the code of
:class:`~crappy.blocks.generator_path.Ramp`)
* the last sent value, using the ``self.last_cmd`` attribute
* any other criteria
Alternatively, if the :class:`~crappy.blocks.generator_path.meta_path.Path`
is done and should hand over to the next one, it must raise a
:exc:`StopIteration` exception. Again, the choice to raise that exception
can be motivated by the current time, a condition as generated by
:meth:`parse_condition`, or any other criteria.
It is also fine for this method to return :obj:`None` if no value should be
output by the Generator Block for this loop.
"""
self.log(logging.WARNING, "The get_cmd was called but is not defined ! "
"Please define a get_cmd method for your "
"Generator path ! Returning the last sent "
"command")
sleep(1)
return self.last_cmd
[docs]
def log(self, level: int, msg: str) -> None:
"""Records log messages for the Path.
Also instantiates the :obj:`~logging.Logger` when logging the first
message.
Args:
level: An :obj:`int` indicating the logging level of the message.
msg: The message to log, as a :obj:`str`.
.. versionadded:: 2.0.0
"""
if self._logger is None:
self._logger = logging.getLogger(
f"{current_process().name}.{type(self).__name__}")
self._logger.log(level, msg)
[docs]
def parse_condition(self,
condition: Optional[Union[str, ConditionType]]
) -> ConditionType:
"""This method returns a function allowing to check whether a given
condition is met or not.
This returned function takes as an input a :obj:`dict` containing the data
received by the :class:`~crappy.blocks.Generator` Block since it sent the
last command. See :meth:`~crappy.blocks.Block.recv_all_data` for
information on the structure of this data. Based on the input data, it
returns :obj:`True` if the condition is met, and :obj:`False` otherwise.
The condition can be given already as a function (or a callable), as
:obj:`None` or more conveniently as a :obj:`str` to parse. If it is given
as a function / callable, this callable is directly returned. If it is
given as :obj:`None`, a function always returning :obj:`False` is returned.
If the condition is given as a string, the supported condition types are :
::
'<var> > <threshold>'
'<var> < <threshold>'
'delay = <your_delay>'
With ``<var>``, ``<threshold>`` and ``<your_delay>`` to be replaced
respectively with the label on which the condition applies, the threshold
for the condition to become :obj:`True`, and the delay before switching to
the next :class:`~crappy.blocks.generator_path.meta_path.Path`.
In the case when a :obj:`str` to parse is given as the condition, a
function performing the check is generated and returned. This way, the user
doesn't have to understand the internals of data transfers in Crappy to
handle custom conditions.
"""
if not isinstance(condition, str):
# First case, the condition is None
if condition is None:
self.log(logging.DEBUG, "Condition is None")
def cond(_: Dict[str, list]) -> bool:
"""Condition always returning False."""
return False
return cond
# Second case, the condition is already a Callable
elif isinstance(condition, Callable):
self.log(logging.DEBUG, "Condition is a callable")
return condition
# Third case, the condition is a string containing '<'
if '<' in condition:
self.log(logging.DEBUG, "Condition is of type var < thresh")
var, thresh = split(r'\s*<\s*', condition)
# Return a function that checks if received data is inferior to threshold
def cond(data: Dict[str, list]) -> bool:
"""Condition checking that the label values are below a given
threshold."""
if var in data:
return any((val < float(thresh) for val in data[var]))
return False
return cond
# Fourth case, the condition is a string containing '>'
elif '>' in condition:
self.log(logging.DEBUG, "Condition is of type var > thresh")
var, thresh = split(r'\s*>\s*', condition)
# Return a function that checks if received data is superior to threshold
def cond(data: Dict[str, list]) -> bool:
"""Condition checking that the label values are above a given
threshold."""
if var in data:
return any((val > float(thresh) for val in data[var]))
return False
return cond
# Fifth case, it is a delay condition
elif match(r'delay', condition, IGNORECASE) is not None:
self.log(logging.DEBUG, "Condition is of type delay=xx")
delay = float(split(r'=\s*', condition)[1])
# Return a function that checks if the delay is expired
def cond(_: Dict[str, list]) -> bool:
"""Condition checking if a given delay is expired."""
return time() - self.t0 > delay
return cond
# Otherwise, it's an invalid syntax
else:
raise ValueError("Wrong syntax for the condition, please refer to the "
"documentation")