examples/a121/algo/vibration/processor.py

examples/a121/algo/vibration/processor.py#

 1# Copyright (c) Acconeer AB, 2022-2026
 2# All rights reserved
 3
 4from __future__ import annotations
 5
 6import pyqtgraph as pg
 7
 8import acconeer.exptool as et
 9from acconeer.exptool import a121
10from acconeer.exptool.a121.algo._utils import APPROX_BASE_STEP_LENGTH_M
11from acconeer.exptool.a121.algo.vibration import (
12    ExampleApp,
13    Processor,
14    ProcessorConfig,
15    ProcessorContext,
16    ProcessorResult,
17    get_high_frequency_config,
18)
19from acconeer.exptool.a121.algo.vibration._processor import RANGE_SUBSWEEP
20from acconeer.exptool.a121.algo.vibration.plot import VibrationPlot
21
22
23def main():
24    args = a121.ExampleArgumentParser().parse_args()
25    et.utils.config_logging(args)
26
27    sensor_config = ExampleApp._get_sensor_config(get_high_frequency_config())
28
29    client = a121.Client.open(**a121.get_client_args(args))
30    metadata = client.setup_session(sensor_config)
31
32    processor = Processor(
33        sensor_config=sensor_config,
34        metadata=metadata,
35        processor_config=ProcessorConfig(),
36        context=ProcessorContext(),
37    )
38    pg_updater = PGUpdater(sensor_config)
39    pg_process = et.PGProcess(pg_updater)
40    pg_process.start()
41
42    client.start_session()
43    interrupt_handler = et.utils.ExampleInterruptHandler()
44
45    while not interrupt_handler.got_signal:
46        result = client.get_next()
47        plot_data = processor.process(result)
48        try:
49            pg_process.put_data(plot_data)
50        except et.PGProccessDiedException:
51            break
52
53    print("Disconnecting...")
54    client.close()
55
56
57class PGUpdater:
58    def __init__(self, sensor_config: a121.SensorConfig) -> None:
59        self._sensor_config = sensor_config.subsweeps[RANGE_SUBSWEEP]
60        self._meas_dist_m = self._sensor_config.start_point * APPROX_BASE_STEP_LENGTH_M
61        self._vibration_plot = VibrationPlot()
62
63    def setup(self, win: pg.GraphicsLayoutWidget) -> None:
64        self._vibration_plot.setup_plot(win, self._meas_dist_m)
65
66    def update(self, processor_result: ProcessorResult) -> None:
67        self._vibration_plot.update_plot(
68            result=processor_result,
69            extra_result=processor_result.extra_result,
70            show_time_series_std=True,
71        )
72
73
74if __name__ == "__main__":
75    main()

View this example on GitHub: acconeer/acconeer-python-exploration