Skip to content

deep.processor.context.action_context

Handling for action context.

ActionContext

Bases: ABC

A context for the processing of an action.

Source code in deep/processor/context/action_context.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class ActionContext(abc.ABC):
    """A context for the processing of an action."""

    def __init__(self, parent: 'TriggerContext', action: 'LocationAction'):
        """
        Create a new action context.

        :param parent: the parent trigger
        :param action: the action config
        """
        self.trigger_context: 'TriggerContext' = parent
        self.location_action: 'LocationAction' = action
        self._triggered = False

    def __enter__(self):
        """Enter and open the context."""
        return self

    def __exit__(self, exception_type, exception_value, exception_traceback):
        """Exit and close the context."""
        if self.has_triggered():
            self.location_action.record_triggered(self.trigger_context.ts)

    def eval_watch(self, watch: str, source: str) -> Tuple[WatchResult, Dict[str, Variable], str]:
        """
        Evaluate an expression in the current frame.

        :param source: The watch source.
        :param watch: The watch expression to evaluate.
        :return: Tuple with WatchResult, collected variables, and the log string for the expression
        """
        var_processor = VariableSetProcessor({}, self.trigger_context.var_cache)

        try:
            result = self.trigger_context.evaluate_expression(watch)
            variable_id, log_str = var_processor.process_variable(watch, result)

            return WatchResult(source, watch, variable_id), var_processor.var_lookup, log_str
        except BaseException as e:
            logging.exception("Error evaluating watch %s", watch)
            return WatchResult(source, watch, None, str(e)), {}, str(e)

    def process_capture_variable(self, name: str, variable: any) -> Tuple[WatchResult, Dict[str, Variable], str]:
        """
        Process a captured variable (exception or return), into a variable set.

        :param name: the name to use (raised or returned)
        :param variable: the value to process
        :return: Tuple with WatchResult, collected variables, and the log string for the expression
        """
        var_processor = VariableSetProcessor({}, self.trigger_context.var_cache)
        variable_id, log_str = var_processor.process_variable(name, variable)

        return WatchResult(WATCH_SOURCE_CAPTURE, name, variable_id), var_processor.var_lookup, log_str

    def process(self):
        """Process the action."""
        try:
            return self._process_action()
        finally:
            self._triggered = True

    @abc.abstractmethod
    def _process_action(self):
        pass

    def has_triggered(self):
        """
        Check if we have triggerd during this context.

        :return: True, if the trigger has been fired.
        """
        return self._triggered

    def can_trigger(self) -> bool:
        """
        Check if the action can trigger.

        Combine checks for rate limits, windows and condition.
        :return: True, if the trigger can be triggered.
        """
        if not self.location_action.can_trigger(self.trigger_context.ts):
            return False
        if self.location_action.condition is None or len(self.location_action.condition.strip()) == 0:
            return True
        result = self.trigger_context.evaluate_expression(self.location_action.condition)
        return str2bool(str(result))

__enter__()

Enter and open the context.

Source code in deep/processor/context/action_context.py 
60
61
62
def __enter__(self):
    """Enter and open the context."""
    return self

__exit__(exception_type, exception_value, exception_traceback)

Exit and close the context.

Source code in deep/processor/context/action_context.py 
64
65
66
67
def __exit__(self, exception_type, exception_value, exception_traceback):
    """Exit and close the context."""
    if self.has_triggered():
        self.location_action.record_triggered(self.trigger_context.ts)

__init__(parent, action)

Create a new action context.

:param parent: the parent trigger :param action: the action config

Source code in deep/processor/context/action_context.py 
49
50
51
52
53
54
55
56
57
58
def __init__(self, parent: 'TriggerContext', action: 'LocationAction'):
    """
    Create a new action context.

    :param parent: the parent trigger
    :param action: the action config
    """
    self.trigger_context: 'TriggerContext' = parent
    self.location_action: 'LocationAction' = action
    self._triggered = False

can_trigger()

Check if the action can trigger.

Combine checks for rate limits, windows and condition. :return: True, if the trigger can be triggered.

Source code in deep/processor/context/action_context.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
def can_trigger(self) -> bool:
    """
    Check if the action can trigger.

    Combine checks for rate limits, windows and condition.
    :return: True, if the trigger can be triggered.
    """
    if not self.location_action.can_trigger(self.trigger_context.ts):
        return False
    if self.location_action.condition is None or len(self.location_action.condition.strip()) == 0:
        return True
    result = self.trigger_context.evaluate_expression(self.location_action.condition)
    return str2bool(str(result))

eval_watch(watch, source)

Evaluate an expression in the current frame.

:param source: The watch source. :param watch: The watch expression to evaluate. :return: Tuple with WatchResult, collected variables, and the log string for the expression

Source code in deep/processor/context/action_context.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def eval_watch(self, watch: str, source: str) -> Tuple[WatchResult, Dict[str, Variable], str]:
    """
    Evaluate an expression in the current frame.

    :param source: The watch source.
    :param watch: The watch expression to evaluate.
    :return: Tuple with WatchResult, collected variables, and the log string for the expression
    """
    var_processor = VariableSetProcessor({}, self.trigger_context.var_cache)

    try:
        result = self.trigger_context.evaluate_expression(watch)
        variable_id, log_str = var_processor.process_variable(watch, result)

        return WatchResult(source, watch, variable_id), var_processor.var_lookup, log_str
    except BaseException as e:
        logging.exception("Error evaluating watch %s", watch)
        return WatchResult(source, watch, None, str(e)), {}, str(e)

has_triggered()

Check if we have triggerd during this context.

:return: True, if the trigger has been fired.

Source code in deep/processor/context/action_context.py 
112
113
114
115
116
117
118
def has_triggered(self):
    """
    Check if we have triggerd during this context.

    :return: True, if the trigger has been fired.
    """
    return self._triggered

process()

Process the action.

Source code in deep/processor/context/action_context.py 
101
102
103
104
105
106
def process(self):
    """Process the action."""
    try:
        return self._process_action()
    finally:
        self._triggered = True

process_capture_variable(name, variable)

Process a captured variable (exception or return), into a variable set.

:param name: the name to use (raised or returned) :param variable: the value to process :return: Tuple with WatchResult, collected variables, and the log string for the expression

Source code in deep/processor/context/action_context.py 
88
89
90
91
92
93
94
95
96
97
98
99
def process_capture_variable(self, name: str, variable: any) -> Tuple[WatchResult, Dict[str, Variable], str]:
    """
    Process a captured variable (exception or return), into a variable set.

    :param name: the name to use (raised or returned)
    :param variable: the value to process
    :return: Tuple with WatchResult, collected variables, and the log string for the expression
    """
    var_processor = VariableSetProcessor({}, self.trigger_context.var_cache)
    variable_id, log_str = var_processor.process_variable(name, variable)

    return WatchResult(WATCH_SOURCE_CAPTURE, name, variable_id), var_processor.var_lookup, log_str

NoActionContext

Bases: ActionContext

Default context if no action can be determined.

Source code in deep/processor/context/action_context.py 
135
136
137
138
139
class NoActionContext(ActionContext):
    """Default context if no action can be determined."""

    def _process_action(self):
        deep.logging.error("Unsupported action type: %s", self.location_action)