emdbg.debug.px4.utils

  1# Copyright (c) 2023, Auterion AG
  2# SPDX-License-Identifier: BSD-3-Clause
  3
  4from __future__ import annotations
  5import re
  6import math
  7
  8from datetime import datetime
  9from itertools import zip_longest
 10from pathlib import Path
 11
 12
 13# -----------------------------------------------------------------------------
 14def gdb_getfield(value: "gdb.Value", name: str, default=None):
 15    """Find the field of a struct/class by name"""
 16    for f in value.type.fields():
 17        if name == f.name:
 18            return value[name]
 19    return default
 20
 21
 22def gdb_iter(obj):
 23    """yields the values in an array or value with a range"""
 24    if hasattr(obj, "value"):
 25        obj = obj.value()
 26    if hasattr(obj.type, "range"):
 27        for ii in range(*obj.type.range()):
 28            yield obj[ii]
 29    else:
 30        return []
 31
 32def gdb_len(obj) -> int:
 33    """Computes the length of a gdb object"""
 34    if hasattr(obj.type, "range"):
 35        start, stop = obj.type.range()
 36        return stop - start
 37    else:
 38        return 1
 39
 40def gdb_backtrace(gdb) -> str:
 41    """
 42    Unfortunately the built-in gdb command `backtrace` often crashes when
 43    trying to resolve function arguments whose memory is inaccessible due to
 44    optimizations or whose type is too complex.
 45    Therefore this is a simpler implementation in Python to avoid GDB crashing.
 46
 47    ```
 48    (gdb) px4_backtrace
 49    #0  0x0800b3be in sched_unlock() at platforms/nuttx/NuttX/nuttx/sched/sched/sched_unlock.c:272
 50    #1  0x0800b59e in nxsem_post() at platforms/nuttx/NuttX/nuttx/sched/semaphore/sem_post.c:175
 51    #2  0x0800b5b6 in sem_post() at platforms/nuttx/NuttX/nuttx/sched/semaphore/sem_post.c:220
 52    #3  0x08171570 in px4::WorkQueue::SignalWorkerThread() at platforms/common/px4_work_queue/WorkQueue.cpp:151
 53    #4  0x08171570 in px4::WorkQueue::Add(px4::WorkItem*) at platforms/common/px4_work_queue/WorkQueue.cpp:143
 54    #5  0x0816f8dc in px4::WorkItem::ScheduleNow() at platforms/common/include/px4_platform_common/px4_work_queue/WorkItem.hpp:69
 55    #6  0x0816f8dc in uORB::SubscriptionCallbackWorkItem::call() at platforms/common/uORB/SubscriptionCallback.hpp:169
 56    #7  0x0816f8dc in uORB::DeviceNode::write(file*, char const*, unsigned int) at platforms/common/uORB/uORBDeviceNode.cpp:221
 57    #8  0x0816faca in uORB::DeviceNode::publish(orb_metadata const*, void*, void const*) at platforms/common/uORB/uORBDeviceNode.cpp:295
 58    #9  0x081700bc in uORB::Manager::orb_publish(orb_metadata const*, void*, void const*) at platforms/common/uORB/uORBManager.cpp:409
 59    #10 0x0816e6d8 in orb_publish(orb_metadata const*, orb_advert_t, void const*) at platforms/common/uORB/uORBManager.hpp:193
 60    #11 0x08160912 in uORB::PublicationMulti<sensor_gyro_fifo_s, (unsigned char)4>::publish(sensor_gyro_fifo_s const&) at platforms/common/uORB/PublicationMulti.hpp:92
 61    #12 0x08160912 in PX4Gyroscope::updateFIFO(sensor_gyro_fifo_s&) at src/lib/drivers/gyroscope/PX4Gyroscope.cpp:149
 62    #13 0x08040e74 in Bosch::BMI088::Gyroscope::BMI088_Gyroscope::FIFORead(unsigned long long const&, unsigned char) at src/drivers/imu/bosch/bmi088/BMI088_Gyroscope.cpp:447
 63    #14 0x08041102 in Bosch::BMI088::Gyroscope::BMI088_Gyroscope::RunImpl() at src/drivers/imu/bosch/bmi088/BMI088_Gyroscope.cpp:224
 64    #15 0x0803fb7a in I2CSPIDriver<BMI088>::Run() at platforms/common/include/px4_platform_common/i2c_spi_buses.h:343
 65    #16 0x0817164c in px4::WorkQueue::Run() at platforms/common/px4_work_queue/WorkQueue.cpp:187
 66    #17 0x08171798 in px4::WorkQueueRunner(void*) at platforms/common/px4_work_queue/WorkQueueManager.cpp:236
 67    #18 0x08014ccc in pthread_startup() at platforms/nuttx/NuttX/nuttx/libs/libc/pthread/pthread_create.c:59
 68    ```
 69
 70    :return: the selected frame's backtrace without resolving function argument
 71    """
 72    frame = gdb.selected_frame()
 73    index = 0
 74    output = []
 75    while(frame and frame.is_valid()):
 76        pc = frame.pc()
 77        if pc > 0xffff_ff00:
 78            output.append(f"#{index: <2} <signal handler called>")
 79        else:
 80            line = "??"
 81            file = "??"
 82            if sal := frame.find_sal():
 83                line = sal.line
 84                if sal.symtab:
 85                    file = sal.symtab.fullname()
 86            if func := frame.function():
 87                func = func.print_name
 88                if not func.endswith(")"): func += "()"
 89            else:
 90                func = "??"
 91            output.append(f"#{index: <2} 0x{pc:08x} in {func} at {file}:{line}")
 92        frame = frame.older()
 93        index += 1
 94    return "\n".join(output)
 95
 96def gdb_relative_location(gdb, location) -> str:
 97    """
 98    GDB can only place breakpoint on specific line number inside a file.
 99    However, if the file changes the line numbers can shift around, which makes
100    this method brittle. Therefore this function finds the function inside
101    the file and applies a offset or searches for a pattern inside that function
102    to determine the correct line number.
103
104    :param location:
105        Location `function:+offset` or `function:regex`. In case the function
106        uses static linkage you may also need to provide a unique part of the
107        filename path to arbitrate multiple identically named static functions
108        `file:function:+offset` or `file:function:regex`.
109
110    :return: absolute location string `file:line_number`
111    """
112    parts = location.split(":")
113    file_name = None
114    if len(parts) == 3:
115        file_name, function_name, line_pattern = parts
116    elif len(parts) == 2:
117        function_name, line_pattern = parts
118    else:
119        raise ValueError(f"Unknown location format '{location}'!")()
120
121    function = gdb.lookup_global_symbol(function_name, gdb.SYMBOL_VAR_DOMAIN)
122    if function is None:
123        # Multiple static symbols may exists, we use the filename to arbitrate
124        if functions := gdb.lookup_static_symbols(function_name, gdb.SYMBOL_VAR_DOMAIN):
125            file_functions = [(f.symtab.fullname(), f) for f in functions]
126            # Arbitrate using file name hint
127            if file_name is not None:
128                file_functions = [f for f in file_functions if file_name in f[0]]
129            if len(file_functions) == 1:
130                function = file_functions[0][1]
131            else:
132                raise ValueError("Multiple functions found:\n - " + "\n - ".join(functions))
133    if function is None:
134        raise ValueError(f"Cannot find function name '{function_name}'")
135    assert function.is_function
136
137    # Find source file and line numbers: how to use file_name?
138    file = function.symtab.fullname()
139    line_numbers = function.symtab.linetable().source_lines()
140    lmin, lmax = min(line_numbers), max(line_numbers)
141
142    if line_pattern.startswith("+"):
143        # line offset relative to function
144        line = int(line_pattern[1:]) + function.line
145    else:
146        # regex line pattern, read source file and find the line
147        lines = Path(file).read_text().splitlines()
148        lines = list(enumerate(lines[lmin:lmax]))
149        for ii, line in lines:
150            if re.search(line_pattern, line):
151                line = lmin + ii
152                break
153        else:
154            lines = "\n  ".join(f"{lmin+l[0]:>4}: {l[1]}" for l in lines)
155            raise ValueError(f"Cannot find source line for '{line_pattern}'!\n"
156                f"Function '{function_name}' stretches over lines {lmin}-{lmax}.\n")
157                # f"Available source lines are:\n  {lines}")
158
159    return f"{file}:{line}"
160
161
162# -----------------------------------------------------------------------------
163def _binary_search(array, value, lo: int, hi: int, direction: int):
164    middle = (lo + hi) // 2
165    if hi - lo <= 1: return middle
166
167    nslice = [(lo, middle), (middle, hi)]
168    pick_lower_half = array[middle] == value
169    if direction > 0: pick_lower_half = 1 - pick_lower_half
170
171    lo, hi = nslice[pick_lower_half]
172    return _binary_search(array, value, lo, hi, direction)
173
174def binary_search_last(array, value, lo: int = None, hi: int = None):
175    """Binary search the last occurrance of value in an array"""
176    if lo is None: lo = 0
177    if hi is None: hi = len(array)
178    return _binary_search(array, value, lo, hi, direction=-1)
179
180def binary_search_first(array, value, lo: int = None, hi: int = None):
181    """Binary search the first occurrance of value in an array"""
182    if lo is None: lo = 0
183    if hi is None: hi = len(array)
184    return _binary_search(array, value, lo, hi, direction=1)
185
186
187# -----------------------------------------------------------------------------
188def chunks(iterable, chunk_size: int, fill=None):
189    """Convert a iterable into a list of chunks and fill the rest with a value"""
190    args = [iter(iterable)] * chunk_size
191    return zip_longest(*args, fillvalue=fill)
192
193
194def add_datetime(filename: str|Path):
195    """
196    Appends a filename with the current date and time:
197    `Year_Month_Day_Hour_Minute_Second`
198
199    Example: `path/name.txt` -> `path/name_2023_04_14_15_03_24.txt`
200    """
201    filename = Path(filename)
202    return filename.with_stem(f"{filename.stem}_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}")
203
204
205def format_units(value: int | float | None, prefixes: dict[str, int] | str,
206                 unit: str = None, fmt: str = None, if_zero: str = None) -> str:
207    """
208    Format a value with the largest prefix.
209
210    The value is divided by the list of prefixes until it is smaller than the
211    next largest prefix. Trailing zeros are replaced by spaces and padding is
212    applied to align the prefixes and units. If the value is zero, the
213    `if_zero` string is returned if defined.
214
215    Predefined prefixes can be passed as a `group:input-prefix`:
216    - `t`: time prefixes from nanoseconds to days.
217    - `si`: SI prefixes from nano to Tera.
218
219    .. note:: The micro prefix is µ, not u.
220
221    Example:
222
223    ```py
224    format_units(123456, "t:µs", fmt=".1f")    # "123.5ms"
225    format_units(0, "t:s", if_zero="-")        # "-"
226    format_units(1234, "si:", "Hz", fmt=".2f") # "1.23kHz"
227    format_units(1001, "si:", "Hz", fmt=".2f") # "1   kHz"
228    format_units(2345, {"k": 1e3, "M": 1e3}, "Si", fmt=".1f") # "2.3MSi"
229    ```
230
231    :param value: An integer or floating point value. If None, an empty string is returned.
232    :param prefixes:
233        A dictionary of prefix string to ratio of the next largest prefix. The
234        dictionary must be sorted from smallest to largest prefix. The prefix of
235        the input value must be the first entry.
236    :param unit: A unit string to be appended to the formatted value.
237    :param fmt: A format specifier to be applied when formatting the value.
238    :param if_zero: A string to be returned when the value is zero.
239    """
240    if value is None: return ""
241    if if_zero is not None and value == 0: return if_zero
242
243    # Find the correct prefix from a list of predefined common prefixes
244    _found = False
245    if prefixes.startswith("t:"):
246        time_units = {"ns": 1e3, "µs": 1e3, "ms": 1e3, "s": 60, "m": 60, "h": 24, "d": 365.25/12}
247        prefixes = prefixes.split(":")[1]
248        prefixes = {k:v for k, v in time_units.items() if _found or (_found := (k == prefixes))}
249    elif prefixes.startswith("si:"):
250        prefixes = prefixes.split(":")[1]
251        prefixes = {k:1e3 for k in ["n", "µ", "m", "", "k", "M", "G", "T"]
252                    if _found or (_found := (k == prefixes))}
253
254    # Divide the value until it is smaller than the next largest prefix
255    for prefix, factor in prefixes.items():
256        if value < factor: break
257        value /= factor
258
259    # Format the value
260    value = f"{value:{fmt or ''}}"
261    value_stripped = value.rstrip("0").rstrip(".")
262    if if_zero is not None and value_stripped == "0": return if_zero
263    # pad the value to the right to align it
264    padding = max(len(p) for p in prefixes.keys()) - len(prefix)
265    padding += len(value) - len(value_stripped)
266    return f"{value_stripped}{padding * ' '}{prefix}{unit or ''}"
267
268
269# -----------------------------------------------------------------------------
270def format_table(fmtstr: str, header: list[str], rows: list[list[str]], columns: int = 1) -> str:
271    """
272    DEPRECATED: Use `rich.table.Table` instead!
273
274    Formats a list of rows into a table of multiple meta-columns based on the format string.
275    Example for formatting an array of registers into a table with three meta-columns:
276
277    ```py
278    fmtstr = "{:%d}  {:>%d}  {:>%d}"
279    header = ["NAME", "HEX VALUE", "INT VALUE"]
280    rows = [[reg, hex(value), value] for reg, value in registers.items()]
281    table = utils.format_table(fmtstr, header, rows, 3)
282    ```
283
284    :param fmtstr: A string describing the spacing between columns and their
285                   alignment. Must have exactly as many entries as the header a
286                   rows. Example: two columns, aligned right and left with a
287                   brace formatter: `"{:>%d} ({:%d})"` The `%d` is replaced by
288                   the max column width by this function.
289    :param header: A list of names for the header row. If you specify more than
290                   one column, the header will be duplicated.
291    :param rows: a list of lists of entries in the table. Each entry will be
292                 converted to `str` to count the maximal length of each column.
293    :param columns: If a table is very long, the header can be duplicated to the
294                    right side of the table to fill more screen space.
295    """
296    # duplicate and join the format string for each column
297    split_horizontal = columns > 0
298    columns = abs(columns)
299    fmtstr = "      :      ".join([fmtstr] * columns)
300    fmtcnt = fmtstr.count("%d")
301    column_width = [0] * fmtcnt
302
303    # Interleave the rows for the later chunking
304    fill = [""] * (fmtcnt // columns)
305    if split_horizontal:
306        rows = [val for tup in zip(*chunks(rows, math.ceil(len(rows) / columns), fill)) for val in tup]
307    # prepend the duplicated header before the rows
308    if header is not None:
309        rows = [header] * columns + rows
310    # Group the individual rows into multiple columns per row
311    rows = [sum(rs, []) for rs in chunks(rows, columns, fill)]
312
313    # collect each line and compute the column width
314    lines = []
315    for row in rows:
316        if len(row) != fmtcnt:
317            raise ValueError("Each row have the same number of entries as the format string")
318        line = [str(l) for l in row]
319        lines.append(line)
320        column_width = [max(w, len(l)) for w, l in zip(column_width, line)]
321
322    # Format the format string with the column width first
323    fmtstr = fmtstr % tuple(column_width)
324    # Now format the actual lines with the formatted format string
325    return "\n".join(fmtstr.format(*line).rstrip() for line in lines) + "\n"
326
327
328# -----------------------------------------------------------------------------
329class _Singleton(type):
330    _instances = {}
331    def __call__(cls, *args, **kwargs):
332        if cls not in cls._instances:
333            cls._instances[cls] = super(_Singleton, cls).__call__(*args, **kwargs)
334        return cls._instances[cls]