From 2a23e18ae9608256217b71a59e4c6aac6333435b Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Wed, 8 Jul 2026 15:48:03 -0400 Subject: [PATCH 1/5] Added proxy_wheel_speed.py example app. --- python/examples/proxy_wheel_speed.py | 216 +++++++++++++++++++++++++++ 1 file changed, 216 insertions(+) create mode 100755 python/examples/proxy_wheel_speed.py diff --git a/python/examples/proxy_wheel_speed.py b/python/examples/proxy_wheel_speed.py new file mode 100755 index 00000000..8affda0a --- /dev/null +++ b/python/examples/proxy_wheel_speed.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 + +"""Proxy raw wheel/vehicle speed output from a source device to a target device as wheel/vehicle speed input. + +This is useful when a vehicle has a "primary" device with a good connection to the vehicle's wheel speed or CAN +data (the source), and one or more other devices (targets) that need that same speed data but are not wired to +the vehicle themselves. +""" + +from datetime import datetime +import os +import sys +import time + +# Add the Python root directory (fusion-engine-client/python/) to the import search path to enable FusionEngine +# imports if this application is being run directly out of the repository and is not installed as a pip package. +root_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), '..')) +sys.path.insert(0, root_dir) + +from fusion_engine_client.messages import * +from fusion_engine_client.parsers import FusionEngineDecoder, FusionEngineEncoder +from fusion_engine_client.utils import trace as logging +from fusion_engine_client.utils.argument_parser import ArgumentParser, ExtendedBooleanAction +from fusion_engine_client.utils.time_provider import TimeProvider +from fusion_engine_client.utils.transport_utils import * + +logger = logging.getLogger('point_one.fusion_engine.proxy_wheel_speed') + +# Map each raw output message we forward to the corresponding input message class sent to the target device. +_INPUT_CLASS_BY_OUTPUT_TYPE = { + MessageType.RAW_WHEEL_SPEED_OUTPUT: WheelSpeedInput, + MessageType.RAW_VEHICLE_SPEED_OUTPUT: VehicleSpeedInput, +} + +# Fields to copy directly from the source output message to the target input message. Both message types share +# the "gear" and "flags" fields; the speed field name(s) differ between wheel and vehicle speed messages. +_COPY_FIELDS_BY_OUTPUT_TYPE = { + MessageType.RAW_WHEEL_SPEED_OUTPUT: ( + 'front_left_speed_mps', 'front_right_speed_mps', 'rear_left_speed_mps', 'rear_right_speed_mps', 'gear', + 'flags'), + MessageType.RAW_VEHICLE_SPEED_OUTPUT: ('vehicle_speed_mps', 'gear', 'flags'), +} + + +def build_input_message(source_message, use_gps_time, time_provider): + """! + @brief Convert a raw wheel/vehicle speed output message from the source device into the corresponding input + message for the target device. + + Note that we intentionally do _not_ copy the source device's P1 time: P1 time is specific to each device and + is not shared between them. Instead, we either let the target device timestamp the measurement when it is + received, or, if requested, convert the source P1 time to GPS time so the target can align it more precisely with + its own P1 time. + + @param source_message The received `RawWheelSpeedOutput` or `RawVehicleSpeedOutput` message. + @param use_gps_time If `True`, timestamp the input message using GPS time derived from the source device's P1 + time. Otherwise, leave the timestamp unset so the target device timestamps it on reception. + @param time_provider A `TimeProvider` used to convert source P1 time to GPS time when `use_gps_time` is + enabled. + + @return The populated input message, or `None` if the message type is not supported. + """ + input_class = _INPUT_CLASS_BY_OUTPUT_TYPE.get(source_message.get_type()) + if input_class is None: + return None + + input_message = input_class() + for field in _COPY_FIELDS_BY_OUTPUT_TYPE[source_message.get_type()]: + setattr(input_message, field, getattr(source_message, field)) + + input_message.details.data_source = source_message.details.data_source + + gps_time = time_provider.p1_to_gps(source_message.p1_time) if use_gps_time else None + if gps_time: + input_message.details.measurement_time = gps_time + input_message.details.measurement_time_source = SystemTimeSource.GPS_TIME + else: + if use_gps_time: + logger.debug('GPS time not yet available. Falling back to timestamp on reception.') + input_message.details.measurement_time_source = SystemTimeSource.INVALID + + return input_message + + +def main(): + parser = ArgumentParser(description="""\ +Proxy raw wheel/vehicle speed output from a source device to a target device as wheel/vehicle speed input. + +Connects to a source device and a target device, enables raw wheel and vehicle speed output on the source, and +forwards each received measurement to the target as the corresponding input message. + +Messages may be timestamped either using GPS time (default) or on reception by the target device (--use-gps-time=false). + +Example: + ./proxy_wheel_speed.py tcp://192.168.1.100:30202 tcp://192.168.1.200:30201 +""") + + parser.add_argument( + 'source', type=str, + help="Source device transport (provides raw wheel/vehicle speed output).\n" + TRANSPORT_HELP_STRING) + + parser.add_argument( + 'target', type=str, + help="Target device transport (receives wheel/vehicle speed input).\n" + TRANSPORT_HELP_STRING) + + parser.add_argument( + '--use-gps-time', action=ExtendedBooleanAction, default=True, + help="""\ +Timestamp forwarded measurements using GPS time instead of letting the target timestamp them on reception.""") + + parser.add_argument( + '--summary-interval', type=float, default=5.0, + help="Print a summary of forwarded messages every N seconds.") + + parser.add_argument( + '-v', '--verbose', action='count', default=0, + help="Print verbose/trace debugging messages.") + + options = parser.parse_args() + + # Configure logging. + logging.basicConfig( + format='%(asctime)s - %(levelname)s - %(name)s:%(lineno)d - %(message)s', + stream=sys.stdout) + + if options.verbose == 0: + logger.setLevel(logging.INFO) + elif options.verbose == 1: + logger.setLevel(logging.DEBUG) + else: + logger.setLevel(logging.DEBUG) + logging.getLogger('point_one.fusion_engine.parsers').setLevel( + logging.getTraceLevel(depth=options.verbose - 1)) + + # Connect to the source and target devices. + response_timeout_sec = 3.0 + try: + source_transport = create_transport(options.source, timeout_sec=response_timeout_sec, print_func=logger.info) + except Exception as e: + logger.error(f"Failed to connect to source: {e}") + sys.exit(1) + + try: + target_transport = create_transport(options.target, timeout_sec=response_timeout_sec, print_func=logger.info) + except Exception as e: + logger.error(f"Failed to connect to target: {e}") + sys.exit(1) + + encoder = FusionEngineEncoder() + + # Ask the source device to send us raw wheel/vehicle speed output every time a new measurement is available. If + # requested, also enable pose output so we can convert the source's P1 time to GPS time. + message_types_to_enable = [MessageType.RAW_WHEEL_SPEED_OUTPUT, MessageType.RAW_VEHICLE_SPEED_OUTPUT] + if options.use_gps_time: + message_types_to_enable.append(MessageType.POSE) + + for message_type in message_types_to_enable: + rate_request = SetMessageRate(output_interface=InterfaceID(TransportType.CURRENT), + protocol=ProtocolType.FUSION_ENGINE, message_id=message_type, + rate=MessageRate.ON_CHANGE) + source_transport.send(encoder.encode_message(rate_request)) + + logger.info("Connected. Waiting for wheel/vehicle speed data from source...") + + # Main loop. + decoder = FusionEngineDecoder(return_bytes=False) + time_provider = TimeProvider() + + messages_received = 0 + messages_forwarded = 0 + start_time = datetime.now() + last_summary_time = start_time + + try: + while True: + # Need to specify a read size or read waits for end of file. This returns immediately even if 0 bytes + # are available. + received_data = recv_from_transport(source_transport, 64) + if len(received_data) == 0: + time.sleep(0.1) + else: + for _, message in decoder.on_data(received_data): + if isinstance(message, PoseMessage): + time_provider.handle_message(message) + continue + elif not isinstance(message, (RawWheelSpeedOutput, RawVehicleSpeedOutput)): + continue + + messages_received += 1 + + input_message = build_input_message(message, use_gps_time=options.use_gps_time, + time_provider=time_provider) + encoded_data = encoder.encode_message(input_message) + target_transport.send(encoded_data) + messages_forwarded += 1 + + logger.debug(f"Forwarded {message.get_type().name} -> {input_message.get_type().name}.") + + now = datetime.now() + if (now - last_summary_time).total_seconds() >= options.summary_interval: + elapsed_sec = (now - start_time).total_seconds() + logger.info(f"Elapsed: {elapsed_sec:.1f} sec, received: {messages_received}, " + f"forwarded: {messages_forwarded}") + last_summary_time = now + except KeyboardInterrupt: + pass + except (ConnectionError, OSError) as e: + logger.error(f"Connection error: {e}") + finally: + source_transport.close() + target_transport.close() + logger.info(f"Done. Forwarded {messages_forwarded} of {messages_received} received messages.") + + +if __name__ == "__main__": + main() From 527a7ab53cd6e2e1ba258790770e4df0e4b5ae04 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Wed, 8 Jul 2026 16:06:59 -0400 Subject: [PATCH 2/5] Allow multiple targets. --- python/examples/proxy_wheel_speed.py | 41 ++++++++++++++++++---------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/python/examples/proxy_wheel_speed.py b/python/examples/proxy_wheel_speed.py index 8affda0a..2f3c2f35 100755 --- a/python/examples/proxy_wheel_speed.py +++ b/python/examples/proxy_wheel_speed.py @@ -83,16 +83,22 @@ def build_input_message(source_message, use_gps_time, time_provider): def main(): - parser = ArgumentParser(description="""\ + parser = ArgumentParser( + usage="%(prog)s SOURCE TARGET [TARGET...]", + description="""\ Proxy raw wheel/vehicle speed output from a source device to a target device as wheel/vehicle speed input. -Connects to a source device and a target device, enables raw wheel and vehicle speed output on the source, and -forwards each received measurement to the target as the corresponding input message. +Connects to a source device and one or more target devices, enables raw wheel and vehicle speed output on the +source, and forwards each received measurement to every target as the corresponding input message. Messages may be timestamped either using GPS time (default) or on reception by the target device (--use-gps-time=false). -Example: +Examples: + # Proxy to a single target. ./proxy_wheel_speed.py tcp://192.168.1.100:30202 tcp://192.168.1.200:30201 + + # Proxy to multiple targets. + ./proxy_wheel_speed.py tcp://192.168.1.100:30202 tcp://192.168.1.200:30201 tcp://192.168.1.201:30201 """) parser.add_argument( @@ -100,8 +106,9 @@ def main(): help="Source device transport (provides raw wheel/vehicle speed output).\n" + TRANSPORT_HELP_STRING) parser.add_argument( - 'target', type=str, - help="Target device transport (receives wheel/vehicle speed input).\n" + TRANSPORT_HELP_STRING) + 'target', type=str, nargs='+', + help="One or more target device transports (each receives wheel/vehicle speed input).\n" + + TRANSPORT_HELP_STRING) parser.add_argument( '--use-gps-time', action=ExtendedBooleanAction, default=True, @@ -140,11 +147,14 @@ def main(): logger.error(f"Failed to connect to source: {e}") sys.exit(1) - try: - target_transport = create_transport(options.target, timeout_sec=response_timeout_sec, print_func=logger.info) - except Exception as e: - logger.error(f"Failed to connect to target: {e}") - sys.exit(1) + target_transports = [] + for target in options.target: + try: + target_transports.append( + create_transport(target, timeout_sec=response_timeout_sec, print_func=logger.info)) + except Exception as e: + logger.error(f"Failed to connect to target '{target}': {e}") + sys.exit(1) encoder = FusionEngineEncoder() @@ -191,10 +201,12 @@ def main(): input_message = build_input_message(message, use_gps_time=options.use_gps_time, time_provider=time_provider) encoded_data = encoder.encode_message(input_message) - target_transport.send(encoded_data) + for target_transport in target_transports: + target_transport.send(encoded_data) messages_forwarded += 1 - logger.debug(f"Forwarded {message.get_type().name} -> {input_message.get_type().name}.") + logger.debug(f"Forwarded {message.get_type().name} -> {input_message.get_type().name} to " + f"{len(target_transports)} target(s).") now = datetime.now() if (now - last_summary_time).total_seconds() >= options.summary_interval: @@ -208,7 +220,8 @@ def main(): logger.error(f"Connection error: {e}") finally: source_transport.close() - target_transport.close() + for target_transport in target_transports: + target_transport.close() logger.info(f"Done. Forwarded {messages_forwarded} of {messages_received} received messages.") From 364cf8b4cd920d563d43929d19071f7c04e2d644 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Wed, 8 Jul 2026 18:03:42 -0400 Subject: [PATCH 3/5] Use source GPS time verbatim if populated. --- python/examples/proxy_wheel_speed.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/python/examples/proxy_wheel_speed.py b/python/examples/proxy_wheel_speed.py index 2f3c2f35..60026281 100755 --- a/python/examples/proxy_wheel_speed.py +++ b/python/examples/proxy_wheel_speed.py @@ -70,7 +70,13 @@ def build_input_message(source_message, use_gps_time, time_provider): input_message.details.data_source = source_message.details.data_source - gps_time = time_provider.p1_to_gps(source_message.p1_time) if use_gps_time else None + gps_time = None + if use_gps_time: + if source_message.details.measurement_time_source == SystemTimeSource.GPS_TIME: + gps_time = source_message.details.measurement_time + else: + gps_time = time_provider.p1_to_gps(source_message.p1_time) + if gps_time: input_message.details.measurement_time = gps_time input_message.details.measurement_time_source = SystemTimeSource.GPS_TIME From 9764ce543c468cd5990351c5fa1b30cfddcd4d6d Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Wed, 8 Jul 2026 18:25:17 -0400 Subject: [PATCH 4/5] Added TCP server support to transport_utils.py. --- .../utils/transport_utils.py | 189 +++++++++++++++++- 1 file changed, 186 insertions(+), 3 deletions(-) diff --git a/python/fusion_engine_client/utils/transport_utils.py b/python/fusion_engine_client/utils/transport_utils.py index 6e2efc15..bb284ef2 100644 --- a/python/fusion_engine_client/utils/transport_utils.py +++ b/python/fusion_engine_client/utils/transport_utils.py @@ -2,6 +2,8 @@ import re import socket import sys +import threading +import time from typing import Any, BinaryIO, Callable, TextIO, Union # WebSocket support is optional. To use, install with: @@ -153,8 +155,8 @@ class SocketTransport: All other member or function accesses are deferred to the underlying `socket.socket` instance. """ - def __init__(self, *args, **kwargs): - self._socket = socket.socket(*args, **kwargs) + def __init__(self, *args, sock: socket.socket = None, **kwargs): + self._socket = sock if sock is not None else socket.socket(*args, **kwargs) self._closed = False @property @@ -176,6 +178,173 @@ def __exit__(self, *args): self.close() +class TCPServerTransport(SocketTransport): + """! + @brief TCP transport that accepts incoming connections on a background thread. + + Unlike connecting as a TCP client, an application acting as a TCP server does not know when (or if) a client will + connect. This class starts listening immediately and accepts connections on a background thread so construction + does not block, allowing the application to continue starting up while it waits. If the connected client + disconnects, the transport goes back to listening. + + `setblocking()`/`settimeout()`/`setsockopt()` are recorded and replayed on each newly accepted connection if + called while no client is connected. `recv()` and other calls block until a client has connected, since there is + nothing to receive until then; if the client disconnects mid-`recv()`, this waits for a new client rather than + raising. `send()` does not block; if there is no connected client (including right after a disconnect), the data + is silently dropped. + + `close()` may be called at any time, including before a client has connected, and stops the background thread + within `_POLL_INTERVAL_SEC` rather than blocking indefinitely on the OS-level `accept()` call. + """ + + _DEFERRABLE_METHODS = ('setblocking', 'settimeout', 'setsockopt') + + # How often the background thread wakes up to check whether it should stop (or whether the current client has + # disconnected and it should accept a new one). This bounds how long close() can take if no client is connected. + _POLL_INTERVAL_SEC = 0.2 + + def __init__(self, port: int, timeout_sec: float = None, print_func: Callable = None): + self._closed = False + self._connect_timeout_sec = timeout_sec + self._print_func = print_func + self._connected_event = threading.Event() + self._lock = threading.Lock() + self._pending_opts = [] + self._socket = None + + self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._listener.settimeout(self._POLL_INTERVAL_SEC) + self._listener.bind(('', port)) + self._listener.listen(1) + + self._accept_thread = threading.Thread( + target=self._accept_loop, name=f'tcp-server-accept-{port}', daemon=True) + self._accept_thread.start() + + def _accept_loop(self): + while True: + with self._lock: + if self._closed: + return + have_client = self._socket is not None + if have_client: + # A client is already connected. Wait for it to disconnect before accepting another. + time.sleep(self._POLL_INTERVAL_SEC) + continue + + try: + client_socket, client_address = self._listener.accept() + except socket.timeout: + continue + except OSError: + return # Listener was closed out from under us by close(). + + with self._lock: + if self._closed: + client_socket.close() + return + + if self._connect_timeout_sec is not None: + client_socket.settimeout(self._connect_timeout_sec) + for name, args, kwargs in self._pending_opts: + getattr(client_socket, name)(*args, **kwargs) + + self._socket = client_socket + self._connected_event.set() + + if self._print_func is not None: + self._print_func(f'Accepted connection from {client_address[0]}:{client_address[1]}.') + + def _handle_disconnect(self, sock: socket.socket): + with self._lock: + if self._socket is sock: + self._socket = None + self._connected_event.clear() + sock.close() + if self._print_func is not None: + self._print_func('Client disconnected. Waiting for a new connection.') + + def recv(self, *args, **kwargs) -> bytes: + """! + @brief Receive data from the connected client. + + Blocks until a client is connected. If the client disconnects, waits for a new client to connect instead of + raising an exception. + """ + while True: + self._connected_event.wait() + sock = self._socket + if sock is None: + continue + try: + data = sock.recv(*args, **kwargs) + except (socket.timeout, TimeoutError): + raise + except OSError: + self._handle_disconnect(sock) + continue + if len(data) == 0: + # The client performed an orderly shutdown. + self._handle_disconnect(sock) + continue + return data + + def send(self, data, *args, **kwargs) -> int: + """! + @brief Send data to the connected client. + + Unlike `recv()`, this does not block waiting for a client to connect. If there is no connected client + (including right after a disconnect), the data is silently dropped. + + @param data The data to be sent. + + @return The number of bytes sent, or 0 if no client is connected. + """ + sock = self._socket + if sock is None: + return 0 + try: + return sock.send(data, *args, **kwargs) + except OSError: + self._handle_disconnect(sock) + return 0 + + def wait_for_connection(self, timeout_sec: float = None) -> bool: + """! + @brief Block until a client has connected. + + @param timeout_sec Maximum time to wait (in seconds), or `None` to wait forever. + + @return `True` once a client is connected, or `False` if `timeout_sec` elapses first. + """ + return self._connected_event.wait(timeout_sec) + + def close(self): + with self._lock: + if self._closed: + return + self._closed = True + sock = self._socket + self._listener.close() + self._accept_thread.join(timeout=self._POLL_INTERVAL_SEC * 2) + if sock is not None: + sock.close() + + def __getattr__(self, name): + if name in self._DEFERRABLE_METHODS: + def _deferred(*args, **kwargs): + with self._lock: + if self._socket is None: + self._pending_opts.append((name, args, kwargs)) + return None + return getattr(self._socket, name)(*args, **kwargs) + return _deferred + + self._connected_event.wait() + return getattr(self._socket, name) + + class WebsocketTransport: """! @brief Websocket wrapper class, mimicking the Python socket API. @@ -237,6 +406,8 @@ def __setattr__(self, item: str, value: Any) -> None: if PATH is '-'. - tcp://HOSTNAME[:PORT] - Connect to the specified hostname (or IP address) and port over TCP (e.g., tty://192.168.0.3:30202); defaults to port 30200. +- tcp://:PORT - Listen for an incoming TCP connection on the specified port + (e.g., tcp://:30200). - udp://:PORT - Listen for incoming data on the specified UDP port (e.g., udp://:12345). Note: When using UDP, you must configure the device to send data to your @@ -292,8 +463,20 @@ def create_transport(descriptor: str, timeout_sec: float = None, print_func: Cal raise ValueError(f"Unsupported file mode '{mode}'.") return transport + # TCP server + m = re.match(r'^tcp://(?:0\.0\.0\.0)?:([0-9]+)$', descriptor) + if m: + port = int(m.group(1)) + if print_func is not None: + print_func(f'Listening for TCP connections on port {port}.') + + # Note: timeout_sec is applied to the accepted connection (once a client connects), not to + # how long we wait for that connection. We wait in the background indefinitely so the + # caller is not blocked before a client connects. + return TCPServerTransport(port, timeout_sec=timeout_sec, print_func=print_func) + # TCP client - m = re.match(r'^tcp://([a-zA-Z0-9-_.]+)?(?::([0-9]+))?$', descriptor) + m = re.match(r'^tcp://([a-zA-Z0-9-_.]+)(?::([0-9]+))?$', descriptor) if m: hostname = m.group(1) ip_address = socket.gethostbyname(hostname) From 83b645d97f0dd69f3b9b08bbf6f327c0896823fd Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Tue, 14 Jul 2026 10:57:32 -0400 Subject: [PATCH 5/5] Clarified wrapped FE summary table. - Include M_TYPE_FUSION_ENGINE_MESSAGE in table - Exclude extracted FE messages from total count/bytes to avoid double counting those + M_TYPE_FUSION_ENGINE_MESSAGE - Separate FE message table from wrapped M_TYPE table --- .../fusion_engine_client/utils/print_utils.py | 48 +++++++++++-------- 1 file changed, 28 insertions(+), 20 deletions(-) diff --git a/python/fusion_engine_client/utils/print_utils.py b/python/fusion_engine_client/utils/print_utils.py index f0cf04ae..cc711103 100644 --- a/python/fusion_engine_client/utils/print_utils.py +++ b/python/fusion_engine_client/utils/print_utils.py @@ -203,14 +203,13 @@ def update(self, header: MessageHeader, message: MessagePayload, message_types: elif header.message_type == MessageType.VERSION_INFO: self.version_info = message elif header.message_type == MessageType.INPUT_DATA_WRAPPER: - wrapped_fe_header = message.get_fe_content_header() if include_wrapped_content: - # Count all wrapped content, not including FusionEngine messages. - if wrapped_fe_header is None: - self.wrapped_non_fe_input_data_stats[message.data_type].count += 1 - self.wrapped_non_fe_input_data_stats[message.data_type].total_bytes += len(message.data) - # Count FusionEngine messages separately. - else: + # Count all wrapped content, including (framed) FusionEngine messages (M_TYPE_FUSION_ENGINE_MESSAGE). + self.wrapped_non_fe_input_data_stats[message.data_type].count += 1 + self.wrapped_non_fe_input_data_stats[message.data_type].total_bytes += len(message.data) + + # If this is M_TYPE_FUSION_ENGINE_MESSAGE, also count the FusionEngine messages themselves. + if wrapped_fe_header is not None: self.wrapped_fe_input_data_stats[wrapped_fe_header.message_type].count += 1 self.wrapped_fe_input_data_stats[wrapped_fe_header.message_type].total_bytes += len(message.data) @@ -281,19 +280,28 @@ def _print_table_header(cols: List[Dict]): format_string, dividers = _print_table_header(cols) total_messages = 0 total_bytes = 0 - have_wrapped_fe = False - for data_type in sorted(wrapped_input_data_types): - if data_type in device_summary.wrapped_non_fe_input_data_stats: - entry = device_summary.wrapped_non_fe_input_data_stats[data_type] - logger.info(format_string.format(data_type.to_string(), entry.count, entry.total_bytes)) - total_messages += entry.count - total_bytes += entry.total_bytes - - if data_type in device_summary.wrapped_fe_input_data_stats: - entry = device_summary.wrapped_fe_input_data_stats[data_type] - logger.info(format_string.format(f'{data_type.to_string()} **', entry.count, entry.total_bytes)) - have_wrapped_fe = True + have_wrapped_fe = len(device_summary.wrapped_fe_input_data_stats) > 0 + + for data_type in sorted(device_summary.wrapped_non_fe_input_data_stats.keys()): + entry = device_summary.wrapped_non_fe_input_data_stats[data_type] + logger.info(format_string.format(data_type.to_string(), entry.count, entry.total_bytes)) + total_messages += entry.count + total_bytes += entry.total_bytes logger.info(format_string.format(*dividers)) logger.info(format_string.format('Total', total_messages, total_bytes)) + + # Note: Wrapped M_TYPE_FUSION_ENGINE_MESSAGE messages are included in total_bytes above. This list displays the + # FusionEngine message payloads. We don't want to double-count them. if have_wrapped_fe: - logger.info('** FusionEngine content extracted from InputDataWrapperMessages.') + logger.info('') + logger.info('FusionEngine messages from M_TYPE_FUSION_ENGINE_MESSAGE:') + logger.info(format_string.format(*dividers)) + for data_type in sorted(device_summary.wrapped_fe_input_data_stats.keys()): + entry = device_summary.wrapped_fe_input_data_stats[data_type] + cls = message_type_to_class.get(data_type, None) + if cls is None: + type_str = data_type.to_string() + else: + type_str = f'{cls.__name__} ({int(data_type)})' + logger.info(format_string.format(type_str, entry.count, entry.total_bytes)) + logger.info(format_string.format(*dividers))