diff --git a/python/fusion_engine_client/analysis/analyzer.py b/python/fusion_engine_client/analysis/analyzer.py index 8df417f7..e232968d 100755 --- a/python/fusion_engine_client/analysis/analyzer.py +++ b/python/fusion_engine_client/analysis/analyzer.py @@ -29,7 +29,7 @@ from .attitude import get_enu_rotation_matrix from .data_loader import DataLoader, MessageData, TimeRange from .reference import ReferenceData, _OWN_LOG_STATISTICS -from ..parsers.file_index import HostTimeIndexMap +from ..parsers.file_index import FileIndexEntry, HostTimeIndexMap from ..utils import trace as logging from ..utils.argument_parser import ArgumentParser, ExtendedBooleanAction, TriStateBooleanAction, CSVAction from ..utils.log import define_cli_arguments as define_log_search_arguments, locate_log @@ -313,6 +313,7 @@ def __init__(self, if self.output_dir is not None: if not os.path.exists(self.output_dir): os.makedirs(self.output_dir) + self._remove_stale_plotly_js() # Determine if this is a long log. In practice, some plots can be extremely slow to generate for long logs # because of plotly limitations when handling a lot of traces (signal status, sky plot), or some may generate @@ -1292,6 +1293,25 @@ def plot_relative_position(self): solution_type=solution_type, displacement_enu_m=displacement_enu_m, std_enu_m=std_enu_m) + @classmethod + def _warn_if_map_cannot_load(cls): + """! + @brief Warn if the installed Plotly cannot display a map opened as a local file. + + Plotly 7 bundles a MapLibre build that runs its worker as an ES module. Chrome does not allow a module + worker on a `file://` page, so the map comes up blank when the generated HTML is opened directly, which is + how these plots are normally viewed. Nothing can be done about that from inside the page, so say so rather + than leaving an empty map to puzzle over. + """ + try: + major_version = int(plotly.__version__.split('.')[0]) + except (AttributeError, IndexError, ValueError): + return + + if major_version >= 7: + _logger.warning('Plotly %s generates a map that will not load when opened as a local file. Install ' + 'plotly<7, or serve the output directory over HTTP.', plotly.__version__) + def plot_map(self, mapbox_token, reference: Optional[ReferenceData] = None): """! @brief Plot a map of the position data. @@ -1299,10 +1319,21 @@ def plot_map(self, mapbox_token, reference: Optional[ReferenceData] = None): @param reference If specified, also plot this reference/truth position, restricted to the time range covered by the pose data. """ - pose_source_ids = self._get_pose_source_ids() - if self.output_dir is None or len(pose_source_ids) == 0: + if self.output_dir is None: return + self._warn_if_map_cannot_load() + + # Get the list of available pose sources to be plotted. + pose_source_ids = self._get_pose_source_ids() + if len(pose_source_ids) == 0: + # If none are found in the designated pose source range, this may be an auxiliary sensor log (e.g., a log + # from the secondary antenna process on a dual-antenna system). Try whatever sources are available. + pose_source_ids = self.source_ids + if len(pose_source_ids) == 0: + self.logger.info('No usable data source IDs found. Skipping map.') + return + mapbox_token = self.get_mapbox_token(mapbox_token) if mapbox_token is None or mapbox_token == "": self.logger.info('*' * 80 + '\n\n' + @@ -1483,6 +1514,7 @@ def _plot_data(name, selected_idx, flags, source_id, lla_deg, customdata_all, ma hovertemplate=hovertemplate) if not have_pose_data: + self.logger.info('No pose data found for any of the available source IDs. Skipping map.') return # Add reference/truth data to the map, if available, restricted to the time range covered by the pose data. @@ -1622,7 +1654,7 @@ def _plot_gnss_skyplot_for_source(self, source_id: int, decimate=True): # Read the GNSS signal data. data = self._get_gnss_signals_data(source_id) - if len(data.messages) == 0: + if len(data.p1_time) == 0: self.logger.info(f'No GNSS signal data available for source ID {source_id}. Skipping sky plot.') return have_gnss_signals_message = not data.using_legacy_satellite_message @@ -1846,7 +1878,7 @@ def _plot_gnss_cn0_for_source(self, source_id: int): # Read the GNSS signal data. data = self._get_gnss_signals_data(source_id) - if len(data.messages) == 0: + if len(data.p1_time) == 0: self.logger.info(f'No GNSS signal data available for source ID {source_id}. Skipping C/N0 plot.') return have_gnss_signals_message = not data.using_legacy_satellite_message @@ -1922,7 +1954,7 @@ def _plot_gnss_azimuth_elevation_for_source(self, source_id: int): # Read the GNSS signal data. data = self._get_gnss_signals_data(source_id) - if len(data.messages) == 0: + if len(data.p1_time) == 0: self.logger.info(f'No GNSS signal data available for source ID {source_id}. Skipping azimuth/elevation ' 'time series plot.') return @@ -2016,7 +2048,7 @@ def _plot_gnss_signal_status_for_source(self, source_id: int): # Read the GNSS signal data. data = self._get_gnss_signals_data(source_id) - if len(data.messages) == 0: + if len(data.p1_time) == 0: self.logger.info(f'No GNSS signal data available for source ID {source_id}. Skipping signal status ' 'plot.') return @@ -2219,7 +2251,7 @@ def _count_selected(selected_p1_times, return_nonzero_time=False): for cond in conditions: idx = cond['cond'](status_flags, signal_has_corrections) if np.any(idx): - figure.add_trace(go.Scattergl(x=time[idx], y=[y_offset] * np.sum(idx), + figure.add_trace(go.Scattergl(x=time[idx], y=np.full(np.sum(idx), y_offset), customdata=np.vstack((other_time[idx], status_flags[idx], cn0_dbhz[idx], @@ -2444,7 +2476,9 @@ def _get_gnss_signals_data(self, source_id: int): self._gnss_signals_data[source_id] = data data.using_legacy_satellite_message = True - self._gnss_signals_data[source_id].to_numpy() + # Release the decoded messages once they have been converted. A long log holds millions of per-signal + # objects here, and the plots below work entirely from the Numpy arrays. + self._gnss_signals_data[source_id].to_numpy(keep_messages=False) return self._gnss_signals_data[source_id] @@ -3612,28 +3646,36 @@ class FilteredTable { self._add_page(name='event_log', html_body=body_html, title="Event Log") def extract_times_before_reset(self): - # Iterate backwards over indices to extract resets and the P1 times before them. - curr_reset_time = None - get_time_before_reset = False + # Locate the event notifications up front rather than walking the whole index. A long log can contain + # millions of entries, almost none of which are event notifications. + file_index = self.reader.get_index() + offsets = file_index.offset + types = file_index.type + event_indices = np.where(types == MessageType.EVENT_NOTIFICATION)[0] + + # For each reset, search backwards for the P1 time of the message preceding it. + def _parse(i): + entry = FileIndexEntry(time=None, type=types[i], offset=offsets[i], message_index=i) + return self.reader.reader.parse_entry_at_index(entry)[1] times_before_resets = {} - file_index = self.reader.get_index() - for entry in file_index[::-1]: - if entry.type == MessageType.EVENT_NOTIFICATION or get_time_before_reset: - # Parse entry at index for payload. - header, payload = self.reader.reader.parse_entry_at_index(entry) - # If entry at index is of a class that isn't recognized, then skip it. + for event_idx in event_indices[::-1]: + try: + payload = _parse(event_idx) + if payload.event_type != EventType.RESET: + continue + reset_time = payload.get_system_time_ns() + except Exception: + continue + + for i in range(event_idx - 1, -1, -1): try: - if get_time_before_reset and payload.get_p1_time() is not None: - times_before_resets[curr_reset_time] = float(payload.get_p1_time()) - get_time_before_reset = False - - # Check if event is a reset. - if entry.type == MessageType.EVENT_NOTIFICATION and payload.event_type == EventType.RESET: - curr_reset_time = payload.get_system_time_ns() - get_time_before_reset = True - except Exception as e: + p1_time = _parse(i).get_p1_time() + except Exception: continue + if p1_time is not None: + times_before_resets[reset_time] = float(p1_time) + break return times_before_resets @@ -3901,6 +3943,32 @@ def _add_page(self, name, html_body, title=None): self.plots[name] = {'title': title, 'path': path} + def _remove_stale_plotly_js(self): + """! + @brief Delete the shared Plotly library from the output directory if it is not the installed version. + + The figures reference one `plotly.min.js` written alongside them (see @ref _add_figure()), and Plotly only + writes that file when it is not already there. Without this, a directory written by an earlier run would go + on serving its original library no matter which version is installed now. + """ + path = os.path.join(self.output_dir, 'plotly.min.js') + if not os.path.exists(path): + return + + try: + with open(path, 'rt', encoding='utf-8') as f: + is_current = f.read() == plotly.offline.get_plotlyjs() + except OSError as e: + self.logger.warning('Unable to read "%s": %s', path, e) + return + + if not is_current: + self.logger.info('Replacing Plotly library in "%s" with the installed version.', self.output_dir) + try: + os.remove(path) + except OSError as e: + self.logger.warning('Unable to remove "%s": %s', path, e) + def _add_figure(self, name, figure=None, title=None, config=None, inject_js: str = None, inject_head: str = None, time_axis_type: Optional[str] = None, custom_hover: bool = True): """! @@ -3950,30 +4018,203 @@ def _add_figure(self, name, figure=None, title=None, config=None, inject_js: str os.makedirs(os.path.dirname(path), exist_ok=True) + compact_x_js = self.__compact_x(figure) + if inject_js is not None: - plotly.io.write_html = functools.partial(self.__write_html_and_inject_js, inject_js, time_axis_type) + shared_customdata_js = self.__hoist_shared_customdata(figure) + plotly.io.write_html = functools.partial(self.__write_html_and_inject_js, inject_js, time_axis_type, + shared_customdata_js) plotly.offline.plot( figure, output_type='file', filename=path, - include_plotlyjs=True, + # Write one copy of the Plotly library alongside the figures and reference it from each of them, + # rather than embedding several MB of Javascript in every file. The figures are generated as a set + # and linked together by `index.html`, so they are already meant to be kept together. + include_plotlyjs='directory', auto_open=False, config=config) if inject_js is not None: plotly.io.write_html = Analyzer.__original_write_html - if inject_head is not None: - with open(path, 'rt') as f: + if inject_head is not None or compact_x_js != '': + with open(path, 'rt', encoding='utf-8') as f: html = f.read() - with open(path, 'wt') as f: - f.write(html.replace('', '' + inject_head, 1)) + + if inject_head is not None: + html = html.replace('', '' + inject_head, 1) + + # The X value restoration must run before Plotly.newPlot(), so insert it at the start of the script that + # creates the plot. + if compact_x_js != '': + # The first statement of the script Plotly writes to create the plot, which runs once Plotly itself + # is loaded. + PLOT_SCRIPT_START = 'window.PLOTLYENV=window.PLOTLYENV || {};' + if PLOT_SCRIPT_START not in html: + raise RuntimeError('Unable to locate the plot creation script in "%s".' % path) + html = html.replace(PLOT_SCRIPT_START, PLOT_SCRIPT_START + compact_x_js, 1) + + with open(path, 'wt', encoding='utf-8') as f: + f.write(html) self.plots[name] = {'title': title, 'path': path if figure is not None else None} + @classmethod + def __hoist_shared_customdata(cls, figure) -> str: + """! + @brief Remove duplicate copies of customdata shared by multiple traces, so it is emitted once for the figure. + + Time series plots typically attach the same time customdata to every trace in the figure (see @ref + _time_hover_customdata()), which Plotly serializes separately for each one. On a long log, those duplicate + copies can account for most of the generated HTML file. + + The first trace using each array keeps it, so Plotly still serializes it in its own compact format. The + returned Javascript points the other traces at that trace's array once the plot is created, so the hover code + sees the same `customdata` it would have without this. + + @param figure The figure to remove duplicate customdata from. Traces sharing an array are modified in place. + + @return Javascript that reattaches the shared arrays, to be injected after `figure` is defined. + """ + # Group the traces by the contents of their customdata. Traces using Plotly's own `hovertemplate` are left + # alone, since Plotly resolves those itself while the plot is being created. + traces_by_data = defaultdict(list) + for i, trace in enumerate(figure.data): + customdata = getattr(trace, 'customdata', None) + if customdata is None or getattr(trace, 'hovertemplate', None) is not None: + continue + array = np.asarray(customdata) + traces_by_data[(array.shape, array.dtype.str, array.tobytes())].append(i) + + # Keep the array on the first trace in each group and remove it from the rest. + source_by_trace = {} + for trace_indices in traces_by_data.values(): + for i in trace_indices[1:]: + source_by_trace[i] = trace_indices[0] + figure.data[i].customdata = None + + if len(source_by_trace) == 0: + return '' + + return """\ +var shared_customdata_source_by_trace = %s; +for (const [trace_index, source_index] of Object.entries(shared_customdata_source_by_trace)) { + figure.data[trace_index].customdata = figure.data[source_index].customdata; +} +""" % json.dumps(source_by_trace) + + @classmethod + def __compact_x(cls, figure) -> str: + """! + @brief Write X values shared by multiple traces only once, and write date X values in a compact format. + + Plotly writes each date in a trace as a separate string (e.g., "2026-09-29T15:59:42.100"), which can account + for half of the generated HTML on a long log. Instead, this function writes dates as millisecond offsets, which + Plotly can encode much more efficiently in its binary format. + + Many traces in a figure also typically share the same X values, which Plotly would otherwise write separately + for each one. This function writes those values to disk just once. + + The returned Javascript restores the X values for every trace before the plot is created, so the figure + displays exactly as it would have without this. Common X arrays are shared in memory, rather than replicated, so + the loaded page uses less RAM. Converting the dates back adds a little time when the browser loads the figure, + but can reduce the size of the HTML file by hundreds of MB on a long log. + + @param figure The figure to compact. Traces with numeric or date X values are modified in place. + + @return Javascript to run before `Plotly.newPlot()`, or an empty string if no trace was modified. + """ + # Group the traces by the contents of their X values. Dates are compared at the millisecond resolution used + # in the browser. + traces_by_data = defaultdict(list) + for i, trace in enumerate(figure.data): + x = getattr(trace, 'x', None) + if not isinstance(x, np.ndarray) or x.dtype.kind not in 'Mfiub' or x.size == 0: + continue + if x.dtype.kind == 'M': + if np.all(np.isnat(x)): + continue + x = x.astype('datetime64[ms]') + traces_by_data[(x.dtype.str, x.tobytes())].append((i, x)) + + # Keep the X values on the first trace in each group and remove them from the rest. Dates are encoded as + # offsets from the earliest time. Plotly writes the offsets as 32-bit integers when they fit. Otherwise, or if + # any times are invalid, they are written as floating point values, with invalid times set to NaN. + base_ms_by_trace = {} + source_by_trace = {} + for traces in traces_by_data.values(): + i, x = traces[0] + if x.dtype.kind == 'M': + valid = ~np.isnat(x) + ms = x.astype(np.int64) + base_ms = int(np.min(ms[valid])) + offsets_ms = ms - base_ms + if not np.all(valid) or np.max(offsets_ms[valid]) > np.iinfo(np.int32).max: + offsets_ms = offsets_ms.astype(float) + offsets_ms[~valid] = np.nan + figure.data[i].x = offsets_ms + base_ms_by_trace[i] = base_ms + + for j, _ in traces[1:]: + figure.data[j].x = None + source_by_trace[j] = i + + if len(base_ms_by_trace) == 0 and len(source_by_trace) == 0: + return '' + + return """ +(function() { + const base_ms_by_trace = %s; + const source_by_trace = %s; + const array_types = {i1: Int8Array, u1: Uint8Array, i2: Int16Array, u2: Uint16Array, i4: Int32Array, + u4: Uint32Array, f4: Float32Array, f8: Float64Array}; + + function DecodeArray(value) { + if (Array.isArray(value)) { + return value; + } + const binary = atob(value.bdata); + const bytes = new Uint8Array(binary.length); + for (let i = 0; i < binary.length; ++i) { + bytes[i] = binary.charCodeAt(i); + } + return new array_types[value.dtype](bytes.buffer); + } + + // Many traces share the same times, so format each one only once. + const date_strings = new Map(); + function ToDateString(time_ms) { + let result = date_strings.get(time_ms); + if (result === undefined) { + result = new Date(time_ms).toISOString().slice(0, 23); + date_strings.set(time_ms, result); + } + return result; + } + + const original_new_plot = Plotly.newPlot; + Plotly.newPlot = function(div, data, ...args) { + Plotly.newPlot = original_new_plot; + for (const [trace_index, base_ms] of Object.entries(base_ms_by_trace)) { + const offsets_ms = DecodeArray(data[trace_index].x); + const x = new Array(offsets_ms.length); + for (let i = 0; i < offsets_ms.length; ++i) { + x[i] = Number.isNaN(offsets_ms[i]) ? null : ToDateString(base_ms + offsets_ms[i]); + } + data[trace_index].x = x; + } + for (const [trace_index, source_index] of Object.entries(source_by_trace)) { + data[trace_index].x = data[source_index].x; + } + return original_new_plot.call(this, div, data, ...args); + }; +})(); +""" % (json.dumps(base_ms_by_trace), json.dumps(source_by_trace)) + # Support for injecting custom javascript into the generated plotly HTML file. - def __write_html_and_inject_js(self, inject_js, time_axis_type, *args, **kwargs): + def __write_html_and_inject_js(self, inject_js, time_axis_type, shared_customdata_js, *args, **kwargs): post_script = kwargs.get("post_script", None) if post_script is None: post_script = "" @@ -4001,6 +4242,10 @@ def __write_html_and_inject_js(self, inject_js, time_axis_type, *args, **kwargs) with open(os.path.join(script_dir, 'plotly_data_support.js'), 'rt') as f: post_script += f.read() + # Reattach any customdata that was hoisted out of the traces. This must come after the file above, which + # defines `figure`. + post_script += shared_customdata_js + # Now inject the custom javascript. post_script += inject_js @@ -4124,8 +4369,11 @@ def _resolve_x_axis(self, p1_time: Optional[np.ndarray] = None, gps_time: Option if self.time_type == 'gps': return gps_time, axis_layout else: + # Plotly writes each of these out as a literal date string, so emit only the resolution a browser can + # represent: Javascript's Date is accurate to the millisecond, and the extra digits of a + # `datetime64[ns]` array account for a third of the X axis data in the generated file. utc = self.time_provider.gps_sec_to_datetime64_array(gps_time) - return utc, axis_layout + return utc.astype('datetime64[ms]'), axis_layout def _time_hover_customdata(self, p1_time: np.ndarray, gps_time: Optional[np.ndarray] = None, x_domain: Optional[str] = None) -> np.ndarray: diff --git a/python/fusion_engine_client/analysis/data_loader.py b/python/fusion_engine_client/analysis/data_loader.py index 164e360a..2c53379d 100644 --- a/python/fusion_engine_client/analysis/data_loader.py +++ b/python/fusion_engine_client/analysis/data_loader.py @@ -525,6 +525,7 @@ def _read(self, str(time_range))) message_count = 0 + log_each_message = logger.isEnabledFor(logging.DEBUG) while True: try: header, payload, message_bytes, message_index = \ @@ -533,18 +534,18 @@ def _read(self, except StopIteration: break - message_size_bytes = header.get_message_size() - message_offset_bytes = self.reader.get_bytes_read() - message_size_bytes + # Describe the message if per-message debug prints are enabled. Building these strings for every message + # in the log is expensive enough to dominate the read when they are not going to be printed. + if log_each_message: + message_size_bytes = header.get_message_size() + logger.debug(' %s %s message @ %d. [length=%d B]' % + ('Skipping unsupported' if payload is None else 'Parsed', header.get_type_string(), + self.reader.get_bytes_read() - message_size_bytes, message_size_bytes)) # Unsupported/unrecognized message type. if payload is None: - logger.debug(' Skipping unsupported %s message @ %d. [length=%d B]' % - (header.get_type_string(), message_offset_bytes, message_size_bytes)) continue - logger.debug(' Parsed %s message @ %d. [length=%d B]' % - (header.get_type_string(), message_offset_bytes, message_size_bytes)) - # Extract P1 and system times from this message, if applicable. p1_time = payload.get_p1_time() system_time_ns = payload.get_system_time_ns() diff --git a/python/fusion_engine_client/analysis/reference.py b/python/fusion_engine_client/analysis/reference.py index 21c5e153..9059994f 100644 --- a/python/fusion_engine_client/analysis/reference.py +++ b/python/fusion_engine_client/analysis/reference.py @@ -1,6 +1,7 @@ from typing import Optional, Union import re +import warnings import numpy as np from pymap3d import ecef2geodetic, geodetic2ecef @@ -285,7 +286,10 @@ def from_own_log(cls, loader: DataLoader, statistic: str, source_id: Optional[in else position_std_enu_m[:, 0]) else: position_ecef_m = np.median(position_ecef_m, axis=1) - with np.errstate(invalid='ignore'): + # A component that is entirely NaN (e.g., roll when only estimating yaw and pitch) yields a NaN median. + # That's ok, but nanmedian() will still warn about it. Suppress the warning. + with warnings.catch_warnings(): + warnings.filterwarnings('ignore', message='All-NaN slice encountered', category=RuntimeWarning) velocity_enu_mps = (None if np.all(np.isnan(velocity_enu_mps)) else np.nanmedian(velocity_enu_mps, axis=1)) ypr_deg = None if np.all(np.isnan(ypr_deg)) else np.nanmedian(ypr_deg, axis=1) diff --git a/python/fusion_engine_client/messages/defs.py b/python/fusion_engine_client/messages/defs.py index d68ddaa6..5accb65a 100644 --- a/python/fusion_engine_client/messages/defs.py +++ b/python/fusion_engine_client/messages/defs.py @@ -250,6 +250,9 @@ class MessageHeader: _FORMAT = ' Dict['InputDataType', Tuple[int, int]]: + """! + @brief Count the wrapped data packets of each type, without decoding the messages themselves. + + A log may contain millions of @ref InputDataWrapperMessage%s, which is far too many to deserialize + individually just to total them up. This reads the handful of bytes needed for each message directly out of + the file instead. + + @param input_path The path to the binary file containing the messages. + @param message_offsets The offset of each @ref InputDataWrapperMessage within the file (in bytes). + + @return A `dict`, keyed by @ref InputDataType, containing the number of packets of that type and the total + size of their wrapped data (in bytes). + """ + offsets = np.asarray(message_offsets, dtype=np.int64) + if len(offsets) == 0: + return {} + + # Gather the message payload sizes and wrapped data types out of the file. Both fields are at a fixed offset + # within each message, so they can be sliced out of a memory map and reinterpreted in bulk. + raw = np.memmap(input_path, dtype=np.uint8, mode='r') + payload_size_offsets = offsets + MessageHeader.PAYLOAD_SIZE_OFFSET_BYTES + payload_size_bytes = raw[(payload_size_offsets[:, np.newaxis] + np.arange(4))].copy().view(np.uint32).ravel() + data_type_offsets = offsets + MessageHeader.calcsize() + cls.DATA_TYPE_OFFSET_BYTES + data_type_ints = raw[(data_type_offsets[:, np.newaxis] + np.arange(2))].copy().view(np.uint16).ravel() + + # Total up the data sizes by type. The payload holds the wrapper's own fields followed by the wrapped data. + data_size_bytes = payload_size_bytes.astype(np.int64) - cls._STRUCT.size + result = {} + for data_type_int in np.unique(data_type_ints): + idx = data_type_ints == data_type_int + data_type = InputDataType(int(data_type_int), raise_on_unrecognized=False) + result[data_type] = (int(np.sum(idx)), int(np.sum(data_size_bytes[idx]))) + return result diff --git a/python/fusion_engine_client/messages/signal_defs.py b/python/fusion_engine_client/messages/signal_defs.py index 72eafc3e..dc310367 100644 --- a/python/fusion_engine_client/messages/signal_defs.py +++ b/python/fusion_engine_client/messages/signal_defs.py @@ -800,6 +800,32 @@ def get_satellite_type(self) -> SatelliteType: def get_prn(self) -> int: return self.prn + # Instances returned by @ref intern(), keyed by (system, PRN). + _interned = {} + + @classmethod + def intern(cls, system: SatelliteType, prn: int) -> 'SatelliteID': + """! + @brief Get a shared instance for a satellite, creating it if it has not been seen before. + + A log repeats the same handful of satellites for many epochs, so reusing one instance per satellite avoids + constructing millions of identical objects while decoding. + + @warning + The returned instance is shared with every other caller asking for the same satellite, and must not be + modified. + + @param system The GNSS constellation to which the satellite belongs. + @param prn The satellite's PRN. + + @return The shared @ref SatelliteID. + """ + key = (system, prn) + result = cls._interned.get(key) + if result is None: + result = cls._interned[key] = cls(system=system, prn=prn) + return result + def decode_hash(self, sv_hash: int): self.system, self.prn, _ = decode_signal_hash(sv_hash) self._hash = int(sv_hash) @@ -917,7 +943,7 @@ def get_satellite_id(self) -> SatelliteID: @return The @ref SatelliteID for the satellite transmitting this signal. """ - return SatelliteID(system=self.signal_type.get_satellite_type(), prn=self.prn) + return SatelliteID.intern(self.signal_type.get_satellite_type(), self.prn) def get_satellite_type(self) -> SatelliteType: return self.signal_type.get_satellite_type() @@ -937,6 +963,32 @@ def get_gnss_component(self) -> GNSSComponent: def get_prn(self) -> int: return self.prn + # Instances returned by @ref intern(), keyed by (signal type, PRN). + _interned = {} + + @classmethod + def intern(cls, signal_type: GNSSSignalType, prn: int) -> 'SignalID': + """! + @brief Get a shared instance for a signal, creating it if it has not been seen before. + + A log repeats the same handful of signals for many epochs, so reusing one instance per signal avoids + constructing millions of identical objects while decoding. + + @warning + The returned instance is shared with every other caller asking for the same signal, and must not be + modified. + + @param signal_type The type of the signal. + @param prn The signal's PRN. + + @return The shared @ref SignalID. + """ + key = (signal_type, prn) + result = cls._interned.get(key) + if result is None: + result = cls._interned[key] = cls(signal_type=signal_type, prn=prn) + return result + def decode_hash(self, signal_hash: int): _, self.prn, self.signal_type = decode_signal_hash(signal_hash) if self.signal_type is None: diff --git a/python/fusion_engine_client/messages/solution.py b/python/fusion_engine_client/messages/solution.py index 20244bbb..abb27a3d 100644 --- a/python/fusion_engine_client/messages/solution.py +++ b/python/fusion_engine_client/messages/solution.py @@ -6,7 +6,7 @@ import numpy as np from ..utils.construct_utils import AutoEnum, construct_message_to_string -from ..utils.enum_utils import IntEnum +from ..utils.enum_utils import IntEnum, lookup_value from .defs import * @@ -441,7 +441,7 @@ def __init__(self): self.cn0_dbhz = np.nan def get_satellite_id(self) -> SatelliteID: - return SatelliteID(system=self.system, prn=self.prn) + return SatelliteID.intern(self.system, self.prn) def pack(self, buffer: bytes = None, offset: int = 0, return_buffer: bool = True) -> (bytes, int): if buffer is None: @@ -730,7 +730,7 @@ class GNSSSatelliteInfo(_GNSSSatelliteInfo): _STRUCT = struct.Struct(' SatelliteID: - return SatelliteID(system=self.system, prn=self.prn) + return SatelliteID.intern(self.system, self.prn) def pack(self, buffer: bytes = None, offset: int = 0, return_buffer: bool = True) -> (bytes, int): if buffer is None: @@ -764,7 +764,7 @@ def unpack(self, buffer: bytes, offset: int = 0, version: int = MessagePayload._ self._STRUCT.unpack_from(buffer=buffer, offset=offset) offset += self._STRUCT.size - self.system = SatelliteType(system, raise_on_unrecognized=False) + self.system = lookup_value(SatelliteType, system) self.elevation_deg = np.nan if elev_int == self._INVALID_ELEVATION else (elev_int * 0.01) self.azimuth_deg = np.nan if azim_int == self._INVALID_AZIMUTH else (azim_int * 0.01) @@ -804,10 +804,10 @@ class GNSSSignalInfo(_GNSSSignalInfo): _STRUCT = struct.Struct(' SignalID: - return SignalID(signal_type=self.signal_type, prn=self.prn) + return SignalID.intern(self.signal_type, self.prn) def get_satellite_id(self) -> SatelliteID: - return SatelliteID(system=self.signal_type.get_satellite_type(), prn=self.prn) + return SatelliteID.intern(self.signal_type.get_satellite_type(), self.prn) def pack(self, buffer: bytes = None, offset: int = 0, return_buffer: bool = True) -> (bytes, int): if buffer is None: @@ -839,7 +839,7 @@ def unpack(self, buffer: bytes, offset: int = 0, version: int = MessagePayload._ self._STRUCT.unpack_from(buffer=buffer, offset=offset) offset += self._STRUCT.size - self.signal_type = GNSSSignalType(signal_type, raise_on_unrecognized=False) + self.signal_type = lookup_value(GNSSSignalType, signal_type) self.cn0_dbhz = np.nan if cn0_int == self._INVALID_CN0 else cn0_int * 0.25 return offset - initial_offset diff --git a/python/fusion_engine_client/parsers/fast_indexer.py b/python/fusion_engine_client/parsers/fast_indexer.py index db984ee0..d5a57ff5 100644 --- a/python/fusion_engine_client/parsers/fast_indexer.py +++ b/python/fusion_engine_client/parsers/fast_indexer.py @@ -38,13 +38,15 @@ def _search_blocks_for_fe(input_path: str, thread_idx: int, block_starts: List[i @return The raw index data corresponding to this thread's data blocks. """ + # Resolved once here rather than inside the search below, where it would run for every message found. + trace3_enabled = _logger.isEnabledFor(logging.getTraceLevel(depth=3)) + if len(block_starts) == 0: - _logger.trace(f'Skipping search thread {thread_idx}. [num_blocks={len(block_starts)}]', depth=2) + _logger.trace('Skipping search thread %d. [num_blocks=0]', thread_idx, depth=2) return np.array([], dtype=_RAW_DTYPE_WITH_SIZE) else: - _logger.trace(f'Starting search thread {thread_idx}. ' - f'[num_blocks={len(block_starts)}, first={block_starts[0]} B, last={block_starts[-1]} B]', - depth=2) + _logger.trace('Starting search thread %d. [num_blocks=%d, first=%d B, last=%d B]', + thread_idx, len(block_starts), block_starts[0], block_starts[-1], depth=2) header = MessageHeader() message_end = 0 num_syncs = 0 @@ -83,7 +85,7 @@ def _search_blocks_for_fe(input_path: str, thread_idx: int, block_starts: List[i # This is lot faster then doing this check in raw Python due to numpy optimizations. sync_matches = np.where(np_data == _PREAMBLE)[0] - _logger.trace(f'Thread {thread_idx}, block {i}: {len(sync_matches)} matches', depth=2) + _logger.trace('Thread %d, block %d: %d matches', thread_idx, i, len(sync_matches), depth=2) num_syncs += len(sync_matches) # To do the CRC check and find a p1_time the full message needs to be parsed. This @@ -118,14 +120,14 @@ def _search_blocks_for_fe(input_path: str, thread_idx: int, block_starts: List[i # Convert the Timestamp to an integer. p1_time_raw = Timestamp._INVALID if math.isnan(p1_time.seconds) else int(p1_time.seconds) message_end = absolute_offset + header.get_message_size() - if _logger.isEnabledFor(logging.getTraceLevel(depth=3)): - _logger.trace(f'Thread {thread_idx}, block {i}: message={header.message_type.to_string()}, ' - f'file_offset={absolute_offset} B, p1_time={p1_time}', + if trace3_enabled: + _logger.trace('Thread %d, block %d: message=%s, file_offset=%d B, p1_time=%s', + thread_idx, i, header.message_type.to_string(), absolute_offset, p1_time, depth=3) raw_list.append((p1_time_raw, int(header.message_type), absolute_offset, header.get_message_size())) except Exception: pass - _logger.trace(f'Thread {thread_idx}: {num_syncs} sync with {len(raw_list)} valid FE.') + _logger.trace('Thread %d: %d sync with %d valid FE.', thread_idx, num_syncs, len(raw_list)) # Return the index data for this section of the file. return np.array(raw_list, dtype=_RAW_DTYPE_WITH_SIZE) diff --git a/python/fusion_engine_client/parsers/mixed_log_reader.py b/python/fusion_engine_client/parsers/mixed_log_reader.py index 1fe737b8..2c038cc1 100644 --- a/python/fusion_engine_client/parsers/mixed_log_reader.py +++ b/python/fusion_engine_client/parsers/mixed_log_reader.py @@ -13,6 +13,9 @@ from ..utils import trace as logging from ..utils.time_range import TimeRange +# The amount of data that must be read between progress prints. +_PROGRESS_INTERVAL_BYTES = 10e6 + class MixedLogReader(object): """! @@ -126,8 +129,8 @@ def __init__(self, else: self.input_file = open(input_file, 'rb') - input_path = self.input_file.name - self.file_size_bytes = os.stat(input_path).st_size + self.input_path = self.input_file.name + self.file_size_bytes = os.stat(self.input_path).st_size if max_bytes is None: self.max_bytes = sys.maxsize @@ -135,7 +138,7 @@ def __init__(self, self.max_bytes = max_bytes # Open the companion index file if one exists, otherwise index the file. - self._original_index = fast_indexer.fast_generate_index(input_path, force_reindex=ignore_index, + self._original_index = fast_indexer.fast_generate_index(self.input_path, force_reindex=ignore_index, save_index=save_index, max_bytes=max_bytes, num_threads=num_threads) self.next_index_elem = 0 @@ -216,6 +219,10 @@ def read_next(self, require_p1_time=False, require_system_time=False): return self._read_next(require_p1_time=require_p1_time, require_system_time=require_system_time) def _read_next(self, require_p1_time=False, require_system_time=False, force_eof=False): + debug_enabled = self.logger.isEnabledFor(logging.DEBUG) + trace1_enabled = self.logger.isEnabledFor(logging.getTraceLevel(depth=1)) + trace2_enabled = self.logger.isEnabledFor(logging.getTraceLevel(depth=2)) + if force_eof: if not self.reached_eof(): self.logger.debug('Forcibly seeking to EOF.') @@ -245,14 +252,18 @@ def _read_next(self, require_p1_time=False, require_system_time=False, force_eof break start_offset_bytes = self.total_bytes_read - self._print_progress() + # Only call into the progress print when enough data has been read for it to have anything to say. The + # call itself is cheap, but not relative to the rest of the loop when it runs once per message. + if (self.total_bytes_read - self.last_print_bytes > _PROGRESS_INTERVAL_BYTES or + self.total_bytes_read < self.last_print_bytes): + self._print_progress() if start_offset_bytes + MessageHeader.calcsize() > self.max_bytes: - self.logger.debug('Max read length exceeded (%d B).' % self.max_bytes) + self.logger.debug('Max read length exceeded (%d B).', self.max_bytes) break - if self.logger.isEnabledFor(logging.getTraceLevel(depth=2)): - self.logger.trace('Reading candidate message @ %d (0x%x).' % (start_offset_bytes, start_offset_bytes), + if trace2_enabled: + self.logger.trace('Reading candidate message @ %d (0x%x).', start_offset_bytes, start_offset_bytes, depth=2) # Read the next message header. @@ -266,7 +277,8 @@ def _read_next(self, require_p1_time=False, require_system_time=False, force_eof try: header = MessageHeader() - header.unpack(data, warn_on_unrecognized=False) + header.unpack(data, warn_on_unrecognized=False, return_sync_bytes=True) + message_length_bytes = MessageHeader.calcsize() + header.payload_size_bytes # Check if the payload is too big. If so, we most likely found an invalid header -- message sync bytes # occurring randomly in non-FusionEngine binary data in the file. @@ -302,15 +314,15 @@ def _read_next(self, require_p1_time=False, require_system_time=False, force_eof header.source_identifier not in self.requested_source_ids): continue - message_length_bytes = MessageHeader.calcsize() + header.payload_size_bytes - if self.logger.isEnabledFor(logging.getTraceLevel(depth=1)): - self.logger.trace('Read %s message @ %d (0x%x). [length=%d B, sequence=%d, # messages=%d]' % - (header.get_type_string(), start_offset_bytes, start_offset_bytes, - message_length_bytes, header.sequence_number, self.valid_count + 1), + + if trace1_enabled: + self.logger.trace('Read %s message @ %d (0x%x). [length=%d B, sequence=%d, # messages=%d]', + header.get_type_string(), start_offset_bytes, start_offset_bytes, + message_length_bytes, header.sequence_number, self.valid_count + 1, depth=1) if start_offset_bytes + message_length_bytes > self.max_bytes: - self.logger.debug('Max read length exceeded (%d B).' % self.max_bytes) + self.logger.debug('Max read length exceeded (%d B).', self.max_bytes) break current_message_index = self.current_message_index @@ -326,10 +338,11 @@ def _read_next(self, require_p1_time=False, require_system_time=False, force_eof self.prev_sequence_number is not None and \ (header.sequence_number - self.prev_sequence_number) != 1 and \ not (header.sequence_number == 0 and self.prev_sequence_number == 0xFFFFFFFF): - func = self.logger.warning if self.warn_on_gaps else self.logger.debug - func('Data gap detected @ %d (0x%x). [sequence=%d, gap_size=%d, total_messages=%d]' % - (start_offset_bytes, start_offset_bytes, header.sequence_number, - header.sequence_number - self.prev_sequence_number, self.valid_count + 1)) + if self.warn_on_gaps or debug_enabled: + func = self.logger.warning if self.warn_on_gaps else self.logger.debug + func('Data gap detected @ %d (0x%x). [sequence=%d, gap_size=%d, total_messages=%d]', + start_offset_bytes, start_offset_bytes, header.sequence_number, + header.sequence_number - self.prev_sequence_number, self.valid_count + 1) self.prev_sequence_number = header.sequence_number # Deserialize the payload if we need it. @@ -344,19 +357,22 @@ def _read_next(self, require_p1_time=False, require_system_time=False, force_eof payload = cls() payload.unpack(buffer=payload_bytes, offset=0, message_version=header.message_version) except NotImplementedError as e: - self.logger.debug("Error %s message: %s" % (header.get_type_string(), str(e))) + if debug_enabled: + self.logger.debug("Error %s message: %s", header.get_type_string(), str(e)) payload = None except Exception as e: - self.logger.error("Error parsing %s message: %s" % (header.get_type_string(), str(e))) + self.logger.error("Error parsing %s message: %s", header.get_type_string(), str(e)) payload = None else: payload = None if require_p1_time and (payload is None or payload.get_p1_time() is None): - self.logger.trace("Skipping %s message. P1 time requested." % header.get_type_string()) + if trace1_enabled: + self.logger.trace("Skipping %s message. P1 time requested.", header.get_type_string()) continue elif require_system_time and (payload is None or payload.get_system_time_ns() is None): - self.logger.trace("Skipping %s message. System time requested." % header.get_type_string()) + if trace1_enabled: + self.logger.trace("Skipping %s message. System time requested.", header.get_type_string()) continue # Extract P1 time if available. @@ -400,16 +416,16 @@ def _read_next(self, require_p1_time=False, require_system_time=False, force_eof return result except ValueError as e: start_offset_bytes += 1 - if self.logger.isEnabledFor(logging.getTraceLevel(depth=2)): - self.logger.trace('%s Rewinding to offset %d (0x%x).' % - (str(e), start_offset_bytes, start_offset_bytes), + if trace2_enabled: + self.logger.trace('%s Rewinding to offset %d (0x%x).', + str(e), start_offset_bytes, start_offset_bytes, depth=2) self.input_file.seek(start_offset_bytes, os.SEEK_SET) self.total_bytes_read = start_offset_bytes # Out of the loop - EOF reached. self._print_progress(self.total_bytes_read) - self.logger.debug("Read %d bytes total." % self.total_bytes_read) + self.logger.debug("Read %d bytes total.", self.total_bytes_read) # Finished iterating. if force_eof: @@ -441,14 +457,17 @@ def _print_progress(self, file_size=None): file_size = min(self.file_size_bytes, self.max_bytes) if self.total_bytes_read < self.last_print_bytes or \ - self.total_bytes_read - self.last_print_bytes > 10e6 or \ + self.total_bytes_read - self.last_print_bytes > _PROGRESS_INTERVAL_BYTES or \ self.total_bytes_read == file_size: elapsed_sec = (datetime.now() - self.start_time).total_seconds() - self.logger.log(logging.INFO if show_progress else logging.DEBUG, - 'Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % - (self.total_bytes_read, file_size, - 100.0 if file_size == 0 else 100.0 * float(self.total_bytes_read) / file_size, - elapsed_sec, (self.total_bytes_read / elapsed_sec / 1e6) if elapsed_sec > 0 else np.nan)) + + level = logging.INFO if show_progress else logging.DEBUG + if self.logger.isEnabledFor(level): + self.logger.log(level, + 'Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]', + self.total_bytes_read, file_size, + 100.0 if file_size == 0 else 100.0 * float(self.total_bytes_read) / file_size, + elapsed_sec, (self.total_bytes_read / elapsed_sec / 1e6) if elapsed_sec > 0 else np.nan) self.last_print_bytes = self.total_bytes_read def parse_entry_at_index(self, index: file_index.FileIndexEntry): @@ -547,14 +566,14 @@ def filter_in_place(self, key, clear_existing: Union[bool, str] = False, invert_ unavailable_source_ids = list(source_ids.difference(self.available_source_ids)) if len(unavailable_source_ids) > 0: self.logger.debug('Not all source IDs requested are available. Cannot extract the following ' - 'source IDs: {}'.format(unavailable_source_ids)) + 'source IDs: %s', unavailable_source_ids) source_ids = list(source_ids.intersection(self.available_source_ids)) if len(source_ids) == 0: self.logger.debug('Requested source IDs unavailable. Cannot extract data.') self.filter_in_place(None, clear_existing='source_id') if len(unavailable_source_ids) > 0: - self.logger.info('Extracting the following available requested source IDs: {}'.format(source_ids)) + self.logger.info('Extracting the following available requested source IDs: %s', source_ids) self.requested_source_ids = source_ids # No key specified (convenience case). diff --git a/python/fusion_engine_client/utils/enum_utils.py b/python/fusion_engine_client/utils/enum_utils.py index e002502c..8cb644d6 100644 --- a/python/fusion_engine_client/utils/enum_utils.py +++ b/python/fusion_engine_client/utils/enum_utils.py @@ -7,6 +7,25 @@ from aenum import extend_enum +def lookup_value(enum_class, value: int): + """! + @brief Resolve an integer to its enum member, without raising for an unrecognized value. + + Returns the same result as `enum_class(value, raise_on_unrecognized=False)`, but skips the enum's call + machinery for a value it has resolved before. This saves CPU usage and time in cases where a value is decoded very + frequently. + + @param enum_class The enum to resolve `value` against. + @param value The integer value to resolve. + + @return The corresponding enum member. + """ + result = enum_class._recognized_value_map_.get(value) + if result is None: + result = enum_class(value, raise_on_unrecognized=False) + return result + + class DynamicEnumMeta(EnumMeta): UNRECOGNIZED_PREFIX = '_U' diff --git a/python/fusion_engine_client/utils/trace.py b/python/fusion_engine_client/utils/trace.py index 0fdef738..df46d737 100644 --- a/python/fusion_engine_client/utils/trace.py +++ b/python/fusion_engine_client/utils/trace.py @@ -180,7 +180,8 @@ def getTraceLevel(depth=1): if not hasattr(logging.Logger, 'trace'): - def _trace_member(self, msg, depth=1, *args, **kwargs): + # Note that `depth` is keyword-only so that it does not capture the first of the `%`-style arguments to `msg`. + def _trace_member(self, msg, *args, depth=1, **kwargs): # The stacklevel value (1) tells findCaller() to get the line number one function call up from log() in # logging/__init__.py. Since we have an function call (this one) in between log() and the caller, we need to # pop up the stack one extra call. @@ -193,7 +194,7 @@ def _trace_member(self, msg, depth=1, *args, **kwargs): logging.Logger.trace = _trace_member -def trace(msg, depth=1, *args, **kwargs): +def trace(msg, *args, depth=1, **kwargs): """ Log a message with severity 'TRACE' on the root logger. If the logger has no handlers, call basicConfig() to add a console handler with a pre-defined @@ -201,4 +202,4 @@ def trace(msg, depth=1, *args, **kwargs): """ if len(logging.root.handlers) == 0: logging.basicConfig() - logging.root.trace(msg, depth=depth, *args, **kwargs) + logging.root.trace(msg, *args, depth=depth, **kwargs) diff --git a/python/requirements.txt b/python/requirements.txt index 51afd243..d9672e33 100644 --- a/python/requirements.txt +++ b/python/requirements.txt @@ -9,8 +9,13 @@ construct>=2.10.0 # Required for analysis and example applications only. Not used by the `messages` package. argparse-formatter>=1.4 colorama>=0.4.4 +# Note: orjson is optional for plotly, but it serializes the figure data several times faster than the +# standard Python json library. +orjson>=3.9.0 palettable>=3.3.0 -plotly>=6.9.0 +# Note: Plotly 7 bundles a MapLibre build whose web worker is an ES module, which Chrome refuses to start +# from a `file://` page. That breaks the generated map, which is normally opened as a local file. +plotly>=6.9.0,<7 pymap3d>=2.4.3 scipy>=1.5.0 diff --git a/python/setup.py b/python/setup.py index 7d83b5e6..4f94f8b8 100644 --- a/python/setup.py +++ b/python/setup.py @@ -32,8 +32,12 @@ def find_version(*file_paths): display_requirements = set([ 'colorama>=0.4.4', + 'orjson>=3.9.0', 'palettable>=3.3.0', - 'plotly>=6.9.0', + # Note: Plotly 7 bundles a MapLibre build whose web worker is an ES module, which Chrome refuses + # to start from a `file://` page. That breaks the generated map, which is normally opened as a + # local file. + 'plotly>=6.9.0,<7', 'pymap3d>=2.4.3', ]) | tools_requirements