""" Contec CMS7000PLUS / CMS9200PLUS Discovery Tool ================================== This script listens on multiple TCP ports simultaneously to discover what data the CMS7000PLUS / CMS9200PLUS patient monitor sends over the network. Usage: python contec_discovery.py It will listen on ports: 511, 515, 516, 517, 518, 519, 520, 6060 and display all received data in hex + ASCII format for protocol analysis. IMPORTANT: Run as Administrator (required for port 511 on Windows) """ import asyncio import logging import sys import os from datetime import datetime # --- Configuration --- LISTEN_HOST = "0.0.0.0" DISCOVERY_PORTS = [511, 515, 516, 517, 518, 519, 520, 6060] LOG_DIR = "captures" # --- Setup Logging --- logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger("contec_discovery") def hex_dump(data: bytes, width: int = 16) -> str: """Format bytes as a hex dump with ASCII representation.""" lines = [] for i in range(0, len(data), width): chunk = data[i:i + width] hex_part = ' '.join(f'{b:02x}' for b in chunk) ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk) lines.append(f" {i:04x} {hex_part:<{width * 3}} |{ascii_part}|") return '\n'.join(lines) def try_decode_hl7(data: bytes) -> str | None: """Attempt to decode the data as an HL7 message.""" try: # Strip MLLP framing if present cleaned = data if cleaned.startswith(b'\x0b'): cleaned = cleaned[1:] if b'\x1c\x0d' in cleaned: cleaned = cleaned[:cleaned.find(b'\x1c\x0d')] text = cleaned.decode('utf-8', errors='ignore').strip() if 'MSH|' in text or 'MSH' in text[:20]: return text except Exception: pass return None async def handle_discovery_client( reader: asyncio.StreamReader, writer: asyncio.StreamWriter, port: int, capture_file ): """Handle a single incoming connection and dump all data.""" client_ip, client_port = writer.get_extra_info('peername') timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") print("\n" + "=" * 80) print(f" NEW CONNECTION on port {port}") print(f" From: {client_ip}:{client_port}") print(f" Time: {timestamp}") print("=" * 80) capture_file.write(f"\n{'='*80}\n") capture_file.write(f"Connection on port {port} from {client_ip}:{client_port} at {timestamp}\n") capture_file.write(f"{'='*80}\n") capture_file.flush() packet_count = 0 try: while True: data = await asyncio.wait_for(reader.read(8192), timeout=30.0) if not data: break packet_count += 1 recv_time = datetime.now().strftime("%H:%M:%S.%f") print(f"\n--- Packet #{packet_count} on port {port} [{recv_time}] ({len(data)} bytes) ---") print(hex_dump(data)) # Try to decode as HL7 hl7_text = try_decode_hl7(data) if hl7_text: print(f"\n >>> HL7 MESSAGE DETECTED <<<") segments = hl7_text.replace('\n', '\r').split('\r') for seg in segments: seg = seg.strip() if seg: print(f" {seg}") # Extract vital signs from OBX segments if seg.startswith("OBX|"): fields = seg.split('|') if len(fields) > 5: obs_id = fields[3] if len(fields) > 3 else "?" obs_value = fields[5] if len(fields) > 5 else "?" obs_unit = fields[6] if len(fields) > 6 else "" print(f" -> Vital: {obs_id} = {obs_value} {obs_unit}") # Also try plain text decode try: text = data.decode('utf-8', errors='ignore') printable_ratio = sum(1 for c in text if c.isprintable() or c in '\r\n\t') / max(len(text), 1) if printable_ratio > 0.7 and not hl7_text: print(f"\n [Text representation]:") for line in text.split('\r'): line = line.strip() if line: print(f" {line}") except Exception: pass # Write to capture file capture_file.write(f"\nPacket #{packet_count} [{recv_time}] ({len(data)} bytes)\n") capture_file.write(hex_dump(data) + "\n") if hl7_text: capture_file.write(f"HL7: {hl7_text}\n") capture_file.write(f"Raw bytes: {data.hex()}\n") capture_file.flush() except asyncio.TimeoutError: print(f" [Timeout - no data for 30s on port {port}]") except asyncio.CancelledError: pass except ConnectionResetError: print(f" [Connection reset by {client_ip} on port {port}]") except Exception as e: print(f" [Error on port {port}: {e}]") finally: print(f"\n Connection closed on port {port} from {client_ip} ({packet_count} packets received)") writer.close() try: await writer.wait_closed() except Exception: pass async def start_port_listener(port: int, capture_file): """Start a TCP server on a specific port.""" async def client_handler(reader, writer): await handle_discovery_client(reader, writer, port, capture_file) try: server = await asyncio.start_server(client_handler, LISTEN_HOST, port) logger.info(f" ✓ Listening on port {port}") return server except PermissionError: logger.error(f" ✗ Port {port} - Permission denied (try running as Administrator)") return None except OSError as e: if "already in use" in str(e).lower() or e.errno == 10048: logger.error(f" ✗ Port {port} - Already in use by another application") else: logger.error(f" ✗ Port {port} - {e}") return None async def main(): print(r""" ╔══════════════════════════════════════════════════════════════╗ ║ Contec CMS7000PLUS / CMS9200PLUS Discovery Tool ║ ║ ║ ║ This tool listens for incoming data from your patient ║ ║ monitor to discover its communication protocol. ║ ║ ║ ║ SETUP REQUIRED: ║ ║ 1. Connect monitor to your PC via Ethernet ║ ║ 2. Set your PC's Ethernet IP to: 192.168.1.50 ║ ║ (or whatever subnet your monitor uses) ║ ║ 3. On the monitor, set CMS/Server IP to YOUR PC's IP ║ ║ 4. Run this script as Administrator ║ ║ ║ ║ Press Ctrl+C to stop ║ ╚══════════════════════════════════════════════════════════════╝ """) # Create captures directory os.makedirs(LOG_DIR, exist_ok=True) capture_filename = os.path.join( LOG_DIR, f"capture_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" ) capture_file = open(capture_filename, 'w', encoding='utf-8') logger.info(f"Capture log: {capture_filename}") # Show current network config hint print("\nStarting listeners on all discovery ports...") print("(If port 511 fails, run PowerShell as Administrator)\n") servers = [] for port in DISCOVERY_PORTS: server = await start_port_listener(port, capture_file) if server: servers.append(server) if not servers: logger.error("No ports could be opened! Check permissions and existing services.") capture_file.close() return active_ports = [s.sockets[0].getsockname()[1] for s in servers if s.sockets] print(f"\n{'─' * 60}") print(f" Active on ports: {active_ports}") print(f" Waiting for data from CMS7000PLUS / CMS9200PLUS...") print(f" Make sure the monitor's CMS/Server IP points to this PC") print(f"{'─' * 60}\n") try: # Keep running until Ctrl+C await asyncio.Event().wait() except asyncio.CancelledError: pass finally: print("\nShutting down...") for server in servers: server.close() for server in servers: await server.wait_closed() capture_file.close() print(f"Capture saved to: {capture_filename}") if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n\nDiscovery stopped by user.")