examples/a121/plot.py

examples/a121/plot.py#

  1# Copyright (c) Acconeer AB, 2022-2026
  2# All rights reserved
  3
  4from __future__ import annotations
  5
  6import numpy as np
  7
  8import acconeer.exptool as et
  9from acconeer.exptool import a121
 10
 11
 12def main():
 13    args = a121.ExampleArgumentParser().parse_args()
 14    et.utils.config_logging(args)
 15
 16    client = a121.Client.open(**a121.get_client_args(args))
 17
 18    session_config = a121.SessionConfig(
 19        [
 20            {
 21                1: a121.SensorConfig(
 22                    subsweeps=[
 23                        a121.SubsweepConfig(
 24                            start_point=25,
 25                            step_length=2,
 26                            num_points=30,
 27                            profile=a121.Profile.PROFILE_1,
 28                            hwaas=10,
 29                        ),
 30                        a121.SubsweepConfig(
 31                            start_point=75,
 32                            step_length=4,
 33                            num_points=25,
 34                            profile=a121.Profile.PROFILE_3,
 35                            hwaas=20,
 36                        ),
 37                    ],
 38                )
 39            },
 40            {
 41                1: a121.SensorConfig(
 42                    receiver_gain=20,
 43                ),
 44            },
 45        ],
 46        extended=True,
 47    )
 48
 49    extended_metadata = client.setup_session(session_config)
 50
 51    pg_updater = PGUpdater(session_config, extended_metadata)
 52    pg_process = et.PGProcess(pg_updater)
 53    pg_process.start()
 54
 55    client.start_session()
 56
 57    interrupt_handler = et.utils.ExampleInterruptHandler()
 58
 59    while not interrupt_handler.got_signal:
 60        extended_result = client.get_next()
 61
 62        try:
 63            pg_process.put_data(extended_result)
 64        except et.PGProccessDiedException:
 65            break
 66
 67    print("Disconnecting...")
 68    pg_process.close()
 69    client.close()
 70
 71
 72class PGUpdater:
 73    def __init__(
 74        self,
 75        session_config: a121.SessionConfig,
 76        extended_metadata: list[dict[int, a121.Metadata]],
 77    ) -> None:
 78        self.session_config = session_config
 79        self.extended_metadata = extended_metadata
 80
 81    def setup(self, win):
 82        self.all_plots = []
 83        self.all_curves = []
 84        self.all_smooth_maxs = []
 85
 86        for group_idx, group in enumerate(self.session_config.groups):
 87            group_plots = {}
 88            group_curves = {}
 89            group_smooth_maxs = {}
 90
 91            for sensor_id, sensor_config in group.items():
 92                title = f"Group {group_idx} / Sensor {sensor_id}"
 93                plot = win.addPlot(title=title)
 94                plot.setMenuEnabled(False)
 95                plot.setMouseEnabled(x=False, y=False)
 96                plot.hideButtons()
 97                plot.showGrid(x=True, y=True)
 98
 99                plot.setLabel("bottom", "Depth (m)")
100                plot.setLabel("left", "Amplitude")
101
102                curves = []
103                for i in range(sensor_config.num_subsweeps):
104                    curve = plot.plot(pen=et.utils.pg_pen_cycler(i))
105                    curves.append(curve)
106
107                group_plots[sensor_id] = plot
108                group_curves[sensor_id] = curves
109
110                smooth_max = et.utils.SmoothMax(self.session_config.update_rate)
111                group_smooth_maxs[sensor_id] = smooth_max
112
113            self.all_plots.append(group_plots)
114            self.all_curves.append(group_curves)
115            self.all_smooth_maxs.append(group_smooth_maxs)
116
117    def update(self, extended_result: list[dict[int, a121.Result]]) -> None:
118        for group_idx, group in enumerate(extended_result):
119            for sensor_id, result in group.items():
120                plot = self.all_plots[group_idx][sensor_id]
121                curves = self.all_curves[group_idx][sensor_id]
122
123                max_ = 0
124
125                for sub_idx, subframe in enumerate(result.subframes):
126                    x = get_distances_m(
127                        self.session_config.groups[group_idx][sensor_id].subsweeps[sub_idx]
128                    )
129                    y = np.abs(subframe).mean(axis=0)
130                    curves[sub_idx].setData(x, y)
131
132                    max_ = max(max_, np.max(y))
133
134                smooth_max = self.all_smooth_maxs[group_idx][sensor_id]
135                plot.setYRange(0, smooth_max.update(max_))
136
137
138def get_distances_m(config):
139    range_p = np.arange(config.num_points) * config.step_length + config.start_point
140    return range_p * 2.5e-3
141
142
143if __name__ == "__main__":
144    main()

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