#!/usr/bin/env python3 """ Test script for dictionary transport testing - Receiver Tests receiving dictionary messages via NATS using nats_bridge.py smartreceive """ import sys import os import json # Add src to path for import sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) from nats_bridge import smartreceive, log_trace import nats import asyncio # Configuration SUBJECT = "/NATSBridge_dict_test" NATS_URL = "nats://nats.yiem.cc:4222" async def main(): log_trace("", f"Starting dictionary transport receiver test...") log_trace("", f"Note: This receiver will wait for messages from the sender.") log_trace("", f"Run test_micropython_dict_sender.py first to send test data.") # Connect to NATS nc = await nats.connect(NATS_URL) log_trace("", f"Connected to NATS at {NATS_URL}") # Subscribe to the subject async def message_handler(msg): log_trace("", f"Received message on {msg.subject}") # Use smartreceive to handle the data result = smartreceive(msg.data) # Result is an envelope dictionary with payloads field containing list of (dataname, data, data_type) tuples for dataname, data, data_type in result["payloads"]: if isinstance(data, dict): log_trace(result.get("correlationId", ""), f"Received dictionary '{dataname}' of type {data_type}") log_trace(result.get("correlationId", ""), f" Keys: {list(data.keys())}") # Display first few items for small dicts if isinstance(data, dict) and len(data) <= 10: log_trace(result.get("correlationId", ""), f" Content: {json.dumps(data, indent=2)}") else: # For large dicts, show summary log_trace(result.get("correlationId", ""), f" Summary: {json.dumps(data, default=str)[:200]}...") # Save to file output_path = f"./received_{dataname}.json" with open(output_path, 'w') as f: json.dump(data, f, indent=2) log_trace(result.get("correlationId", ""), f"Saved dictionary to {output_path}") else: log_trace(result.get("correlationId", ""), f"Received unexpected data type for '{dataname}': {type(data)}") sid = await nc.subscribe(SUBJECT, cb=message_handler) log_trace("", f"Subscribed to {SUBJECT} with subscription ID: {sid}") # Keep listening for 120 seconds await asyncio.sleep(120) await nc.close() log_trace("", "Test completed.") if __name__ == "__main__": asyncio.run(main())