Skip to content

qass.tools.analyzer.input_observer

Classes:

Name Description
InputObserver

The InputObserver class can observe a Analyzer4D data stream for toggles of input bits.

InputObserver

The InputObserver class can observe a Analyzer4D data stream for toggles of input bits. Its configuration is a list of tuples to observe different input bits. Input bits can be observed for high or low states (True/False). Furthermore an offset after the end of a range can be configured.

When a finished period for an input has been detected the callback function is called.

In each clock cycle the function tick() must be called with the current operator network time. process_start() is called to reset the InputObserver for a new process. process_end() is called to finish all currently detected periods. The callback is then called even if the delay did not expire.

Methods:

Name Description
__init__

Parameters
process_end

This function must be called in the phase eval_process_end() of the operator network.
process_init

The function process_start must be called in the phase eval_process_init() of the operator network.
tick

This function must be called in the phase eval_process_run() of the operator network.
Source code in src/qass/tools/analyzer/input_observer.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class InputObserver:
    """
    The InputObserver class can observe a Analyzer4D data stream for toggles of input bits.
    Its configuration is a list of tuples to observe different input bits.
    Input bits can be observed for high or low states (True/False).
    Furthermore an offset after the end of a range can be configured.

    When a finished period for an input has been detected the callback function is called.

    In each clock cycle the function tick() must be called with the current operator network time.
    process_start() is called to reset the InputObserver for a new process.
    process_end() is called to finish all currently detected periods. The callback is then called even if the delay did not expire.
    """

    def __init__(
        self,
        rti: RunTimeInfo_IF,
        inputs: List[Tuple[int, int, bool, int]],
        callback,
        update_time_ms: Union[int, None] = None,
    ):
        """
        Parameters
        ----------
        rti : RunTimeInfo_IF
            Operator network's runtime info object. This object contains information about the current analysis.
        inputs : List[Tuple[int, int, bool, int]]
            list of input oberserver configurations.
            Each tuple in the list must habe the following form: (byte, bit, state [True/False], delay [in nanoseconds])
        callback : Callable[[tuple[int, int, bool, int], int, int]]
            The callback function is called.
            Its signature must be: ((byte, bit, state, delay), start_time, end_time).
            The times are provided in nanoseconds.
        update_time_ms : int
            declares how often the IO input state is checked for changes.
        """
        self._rti = rti
        self._inputs = inputs
        self._stream = None
        self._current_input_times = None
        self._finished_ranges = None
        self._last_spec = None
        self._callback = callback
        self._spec_duration = None
        self._frq_band_count = None
        self._update_time_ms = update_time_ms
        self._stream_update_specs = None

    def process_init(self, stream: Buffer_Py_IF) -> None:
        """
        The function process_start must be called in the phase eval_process_init() of the operator network.

        Parameters
        ----------
        stream : Buffer_Py_IF
            The stream that should be monitored for input changes.
            Ensure that the stream has an appropriate resolution for your demands.
        """
        self._stream = stream
        self._current_input_times = [None for _ in self._inputs]
        self._last_spec = 0
        self._finished_ranges = []
        self._spec_duration = stream.getSpecDuration()
        self._frq_band_count = stream.getFrqBandCount()
        if self._update_time_ms is None:
            io_specs_updates_raw = 15  # For hardware reasons the Optimizer4D records io changes every 15 raw specs.
            self._stream_update_specs = max(
                1, io_specs_updates_raw / stream.getCompressionTime()
            )
        else:
            self._stream_update_specs = (
                self._update_time_ms * 1e6 / self._spec_duration()
            )

    def tick(self, ignore_getIO_exception: bool = False) -> None:
        """
        This function must be called in the phase eval_process_run() of the operator network.
        It is called for every single spectrum and checks the current input state.

        Parameters
        ----------
        ignore_getIO_exception : bool
            If True exceptions raised by missing IO information in the last specs of a buffer are ignored.
            Otherwise they will be reraised.
        """
        spec_duration = self._spec_duration
        curr_spec = int(self._rti.getCurrentTime() // spec_duration)

        specs = np.arange(self._last_spec, curr_spec, self._stream_update_specs)
        if len(specs):
            self._last_spec = float(specs[-1])
        specs = specs.astype(int)  # [int(s) for s in specs]
        for spec in specs:
            stream_pos = spec * self._frq_band_count
            for idx, (byte, bit, state, delay) in enumerate(self._inputs):
                try:
                    curr_state = self._stream.getIO_inputs(stream_pos, byte, bit)
                except Exception as exc:
                    if ignore_getIO_exception:
                        break
                    else:
                        raise exc

                if curr_state == state and self._current_input_times[idx] is None:
                    self._current_input_times[idx] = spec * spec_duration
                elif curr_state != state and self._current_input_times[idx] is not None:
                    self._finished_ranges.append(
                        (
                            (byte, bit, state, delay),
                            self._current_input_times[idx],
                            spec * spec_duration + delay,
                        )
                    )
                    self._current_input_times[idx] = None

            exec_ranges = [
                (cfg, start, end)
                for cfg, start, end in self._finished_ranges
                if end <= spec * spec_duration
            ]
            if exec_ranges:
                self._finished_ranges = [
                    (cfg, start, end)
                    for cfg, start, end in self._finished_ranges
                    if end > spec * spec_duration
                ]

            for cfg, start, end in exec_ranges:
                self._callback(cfg, start, end)

    def process_end(self) -> None:
        """
        This function must be called in the phase eval_process_end() of the operator network.
        It will finish all currently opened ranges and call the callback for delayed ranges even if the delay has not expired yet.
        """
        self.tick(ignore_getIO_exception=True)

        for cfg, start, end in self._finished_ranges:
            self._callback(cfg, start, self._rti.getCurrentTime())

        for idx, (byte, bit, state, delay) in enumerate(self._inputs):
            if self._current_input_times[idx] is not None:
                self._callback(
                    (byte, bit, state, delay),
                    self._current_input_times[idx],
                    self._rti.getCurrentTime(),
                )

__init__ 

__init__(rti: RunTimeInfo_IF, inputs: List[Tuple[int, int, bool, int]], callback, update_time_ms: Union[int, None] = None)

Parameters:

Name Type Description Default
rti RunTimeInfo_IF

Operator network's runtime info object. This object contains information about the current analysis.

 required
inputs List[Tuple[int, int, bool, int]]

list of input oberserver configurations. Each tuple in the list must habe the following form: (byte, bit, state [True/False], delay [in nanoseconds])

 required
callback Callable[[tuple[int, int, bool, int], int, int]]

The callback function is called. Its signature must be: ((byte, bit, state, delay), start_time, end_time). The times are provided in nanoseconds.

 required
update_time_ms int

declares how often the IO input state is checked for changes.

 None
Source code in src/qass/tools/analyzer/input_observer.py 
40
41
42
43
44
45
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
def __init__(
    self,
    rti: RunTimeInfo_IF,
    inputs: List[Tuple[int, int, bool, int]],
    callback,
    update_time_ms: Union[int, None] = None,
):
    """
    Parameters
    ----------
    rti : RunTimeInfo_IF
        Operator network's runtime info object. This object contains information about the current analysis.
    inputs : List[Tuple[int, int, bool, int]]
        list of input oberserver configurations.
        Each tuple in the list must habe the following form: (byte, bit, state [True/False], delay [in nanoseconds])
    callback : Callable[[tuple[int, int, bool, int], int, int]]
        The callback function is called.
        Its signature must be: ((byte, bit, state, delay), start_time, end_time).
        The times are provided in nanoseconds.
    update_time_ms : int
        declares how often the IO input state is checked for changes.
    """
    self._rti = rti
    self._inputs = inputs
    self._stream = None
    self._current_input_times = None
    self._finished_ranges = None
    self._last_spec = None
    self._callback = callback
    self._spec_duration = None
    self._frq_band_count = None
    self._update_time_ms = update_time_ms
    self._stream_update_specs = None

process_end 

process_end() -> None

This function must be called in the phase eval_process_end() of the operator network. It will finish all currently opened ranges and call the callback for delayed ranges even if the delay has not expired yet.

Source code in src/qass/tools/analyzer/input_observer.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def process_end(self) -> None:
    """
    This function must be called in the phase eval_process_end() of the operator network.
    It will finish all currently opened ranges and call the callback for delayed ranges even if the delay has not expired yet.
    """
    self.tick(ignore_getIO_exception=True)

    for cfg, start, end in self._finished_ranges:
        self._callback(cfg, start, self._rti.getCurrentTime())

    for idx, (byte, bit, state, delay) in enumerate(self._inputs):
        if self._current_input_times[idx] is not None:
            self._callback(
                (byte, bit, state, delay),
                self._current_input_times[idx],
                self._rti.getCurrentTime(),
            )

process_init 

process_init(stream: Buffer_Py_IF) -> None

The function process_start must be called in the phase eval_process_init() of the operator network.

Parameters:

Name Type Description Default
stream Buffer_Py_IF

The stream that should be monitored for input changes. Ensure that the stream has an appropriate resolution for your demands.

 required
Source code in src/qass/tools/analyzer/input_observer.py 
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
def process_init(self, stream: Buffer_Py_IF) -> None:
    """
    The function process_start must be called in the phase eval_process_init() of the operator network.

    Parameters
    ----------
    stream : Buffer_Py_IF
        The stream that should be monitored for input changes.
        Ensure that the stream has an appropriate resolution for your demands.
    """
    self._stream = stream
    self._current_input_times = [None for _ in self._inputs]
    self._last_spec = 0
    self._finished_ranges = []
    self._spec_duration = stream.getSpecDuration()
    self._frq_band_count = stream.getFrqBandCount()
    if self._update_time_ms is None:
        io_specs_updates_raw = 15  # For hardware reasons the Optimizer4D records io changes every 15 raw specs.
        self._stream_update_specs = max(
            1, io_specs_updates_raw / stream.getCompressionTime()
        )
    else:
        self._stream_update_specs = (
            self._update_time_ms * 1e6 / self._spec_duration()
        )

tick 

tick(ignore_getIO_exception: bool = False) -> None

This function must be called in the phase eval_process_run() of the operator network. It is called for every single spectrum and checks the current input state.

Parameters:

Name Type Description Default
ignore_getIO_exception bool

If True exceptions raised by missing IO information in the last specs of a buffer are ignored. Otherwise they will be reraised.

 False
Source code in src/qass/tools/analyzer/input_observer.py 
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def tick(self, ignore_getIO_exception: bool = False) -> None:
    """
    This function must be called in the phase eval_process_run() of the operator network.
    It is called for every single spectrum and checks the current input state.

    Parameters
    ----------
    ignore_getIO_exception : bool
        If True exceptions raised by missing IO information in the last specs of a buffer are ignored.
        Otherwise they will be reraised.
    """
    spec_duration = self._spec_duration
    curr_spec = int(self._rti.getCurrentTime() // spec_duration)

    specs = np.arange(self._last_spec, curr_spec, self._stream_update_specs)
    if len(specs):
        self._last_spec = float(specs[-1])
    specs = specs.astype(int)  # [int(s) for s in specs]
    for spec in specs:
        stream_pos = spec * self._frq_band_count
        for idx, (byte, bit, state, delay) in enumerate(self._inputs):
            try:
                curr_state = self._stream.getIO_inputs(stream_pos, byte, bit)
            except Exception as exc:
                if ignore_getIO_exception:
                    break
                else:
                    raise exc

            if curr_state == state and self._current_input_times[idx] is None:
                self._current_input_times[idx] = spec * spec_duration
            elif curr_state != state and self._current_input_times[idx] is not None:
                self._finished_ranges.append(
                    (
                        (byte, bit, state, delay),
                        self._current_input_times[idx],
                        spec * spec_duration + delay,
                    )
                )
                self._current_input_times[idx] = None

        exec_ranges = [
            (cfg, start, end)
            for cfg, start, end in self._finished_ranges
            if end <= spec * spec_duration
        ]
        if exec_ranges:
            self._finished_ranges = [
                (cfg, start, end)
                for cfg, start, end in self._finished_ranges
                if end > spec * spec_duration
            ]

        for cfg, start, end in exec_ranges:
            self._callback(cfg, start, end)