Mastering Serial Communication on Ubuntu and VMware: Complete Setup & Bidirectional Communication Guide

sudo usermod -aG dialout $USER

newgrp dialout

# Log out of the current session
exit
# or use Ctrl+D

sudo reboot

groups

ls /dev/ttyUSB* /dev/ttyS*

sudo dmesg | grep tty

# List connected USB devices
lsusb

# Detailed information about a specific device
udevadm info -q all -n /dev/ttyUSB0
# or
sudo udevadm info /dev/ttyUSB0

ls -l /dev/ttyS0

# Check whether the device was recognized
dmesg | grep -i usb
sudo dmesg | grep tty

sudo dmesg | grep tty

#!/usr/bin/env python3
"""
VMware Port Occupier
Forces VMware to choose COM2 and COM4 via automatic detection
"""

import serial
import time
from datetime import datetime

class PortOccupier:
    def __init__(self):
        self.ports = {}
        self.running = False

    def occupy_ports(self, port_list):
        """Occupy specified ports"""
        print("🔧 Port Occupier started")
        print(f"📅 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 50)

        self.running = True

        for port in port_list:
            try:
                ser = serial.Serial(port, 9600, timeout=1)
                self.ports[port] = ser
                print(f"✅ Occupied {port}")
            except serial.SerialException as e:
                print(f"❌ Failed to occupy {port}: {e}")

        if self.ports:
            print("\n🎯 VMware setup steps:")
            print("1. Open VMware virtual machine settings")
            print("2. Run automatic detection in the serial port settings")
            print("3. VMware selects ports that are not occupied")
            print("4. When done, type 'q' + Enter to exit")

            print(f"\n📋 Ports in use: {list(self.ports.keys())}")

            # Wait for user input
            while self.running:
                try:
                    user_input = input("Type 'q' + Enter to exit: ").strip().lower()
                    if user_input == 'q':
                        break
                except KeyboardInterrupt:
                    break

    def release_ports(self):
        """Release all ports"""
        print("\n🔌 Releasing ports...")
        for port, ser in self.ports.items():
            try:
                if ser.is_open:
                    ser.close()
                print(f"✅ Released {port}")
            except Exception as e:
                print(f"❌ Error releasing {port}: {e}")

        self.ports.clear()
        self.running = False
        print("✅ All ports released")

def main():
    """Main routine"""
    print("🔧 VMware Port Occupier")
    print("🎯 Goal: Make VMware automatically choose COM2 and COM4")

    # Occupy COM1 and COM3 so that VMware selects COM2 and COM4
    target_ports = ['COM1', 'COM3']

    occupier = PortOccupier()

    try:
        occupier.occupy_ports(target_ports)
    except KeyboardInterrupt:
        print("\n⚠️ Terminated")
    finally:
        occupier.release_ports()

if __name__ == "__main__":
    main()

# Check serial devices
sudo dmesg | grep tty
# Expected lines:
# ttyS0 at I/O 0x3f8 (irq = 4) - pipe 1
# ttyS1 at I/O 0x2f8 (irq = 3) - pipe 2

# Set permissions
sudo chmod 666 /dev/ttyS0 /dev/ttyS1
# or
sudo usermod -a -G dialout $USER  # requires relogin

#!/usr/bin/env python3
"""
VMware Named Pipe Bidirectional Communication
Windows-side implementation
"""

import win32pipe
import win32file
import threading
import time
import json
from datetime import datetime

class VMwarePipeComm:
    def __init__(self):
        self.running = False

    def connect_to_pipe(self, pipe_name, timeout=5000):
        """Connect to a named pipe"""
        try:
            handle = win32file.CreateFile(
                pipe_name,
                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                0, None,
                win32file.OPEN_EXISTING,
                0, None
            )
            return handle
        except Exception as e:
            print(f"Pipe connection error {pipe_name}: {e}")
            return None

    def bidirectional_communication(self):
        """Run bidirectional communication"""
        print("=== VMware Named Pipe Bidirectional Communication ===")

        # Connect to pipes
        tx_pipe = self.connect_to_pipe(r"\\.\pipe\vmware_tx")  # sending
        rx_pipe = self.connect_to_pipe(r"\\.\pipe\vmware_rx")  # receiving

        if not tx_pipe or not rx_pipe:
            print("Pipe connection failed")
            return

        print("✅ Both pipes connected successfully")
        self.running = True

        # Start receive thread
        rx_thread = threading.Thread(target=self.receive_data, args=(rx_pipe,))
        rx_thread.daemon = True
        rx_thread.start()

        # Sending loop
        self.send_data(tx_pipe)

        # Clean up
        self.running = False
        win32file.CloseHandle(tx_pipe)
        win32file.CloseHandle(rx_pipe)

    def send_data(self, pipe_handle):
        """Send data"""
        counter = 1
        try:
            while self.running:
                message = {
                    "id": counter,
                    "timestamp": datetime.now().isoformat(),
                    "source": "Windows_Host",
                    "data": f"Message_{counter:03d}"
                }

                data = (json.dumps(message) + "\n").encode("utf-8")
                win32file.WriteFile(pipe_handle, data)

                print(f"[TX {counter:03d}] → Linux: {message['data']}")
                counter += 1
                time.sleep(2)

        except KeyboardInterrupt:
            print("Stopped sending")

    def receive_data(self, pipe_handle):
        """Receive data"""
        buffer = b""
        try:
            while self.running:
                try:
                    _, data = win32file.ReadFile(pipe_handle, 1024)
                    if data:
                        buffer += data
                        while b"\n" in buffer:
                            line, buffer = buffer.split(b"\n", 1)
                            if line:
                                msg = line.decode("utf-8", errors="ignore")
                                print(f"[RX] ← Linux: {msg}")
                except:
                    time.sleep(0.1)
        except Exception as e:
            print(f"Receive error: {e}")

if __name__ == "__main__":
    comm = VMwarePipeComm()
    comm.bidirectional_communication()

#!/usr/bin/env python3
"""
VMware Dual-Pipe Bidirectional Communication - Windows Side
Achieves full duplex messaging
"""

import win32pipe
import win32file
import win32event
import win32api
import threading
import time
import json
import signal
import sys
import atexit
from datetime import datetime
import pywintypes

class DualPipeComm:
    def __init__(self):
        self.running = False
        # Windows → Linux (send)
        self.tx_pipe = None
        # Linux → Windows (receive)
        self.rx_pipe = None
        self.threads = []

        # Set up safe exit
        atexit.register(self.cleanup)
        signal.signal(signal.SIGINT, self.signal_handler)

    def signal_handler(self, signum, frame):
        """Handle termination signal"""
        print("\n⚠️ Signal received")
        self.safe_shutdown()
        sys.exit(0)

    def safe_shutdown(self):
        """Graceful shutdown"""
        print("🛑 Shutting down...")
        self.running = False

        # Wait for threads to end
        for thread in self.threads:
            if thread.is_alive():
                thread.join(timeout=1.0)

        self.cleanup()
        print("✅ Shutdown complete")

    def cleanup(self):
        """Clean up resources"""
        try:
            if self.tx_pipe:
                win32file.CloseHandle(self.tx_pipe)
                self.tx_pipe = None
                print("📤 Closed TX pipe")
        except:
            pass

        try:
            if self.rx_pipe:
                win32file.CloseHandle(self.rx_pipe)
                self.rx_pipe = None
                print("📥 Closed RX pipe")
        except:
            pass

    def connect_pipe_safe(self, pipe_name, timeout=5000):
        """Safely connect to a named pipe"""
        print(f"🔌 Attempting to connect: {pipe_name}")

        try:
            # Wait for the pipe to become available
            win32pipe.WaitNamedPipe(pipe_name, timeout)

            # Connect in overlapped mode
            handle = win32file.CreateFile(
                pipe_name,
                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                0,
                None,
                win32file.OPEN_EXISTING,
                win32file.FILE_FLAG_OVERLAPPED,
                None,
            )

            print(f"✅ Connected: {pipe_name}")
            return handle

        except Exception as e:
            print(f"❌ Connection failed {pipe_name}: {e}")
            return None

    def dual_pipe_communication(self):
        """Bidirectional communication using two pipes"""
        print("=== VMware Dual-Pipe Communication ===")
        print("📤 TX pipe: \\\\.\\pipe\\win_to_linux")
        print("📥 RX pipe: \\\\.\\pipe\\linux_to_win")
        print("💡 Press Ctrl+C to exit safely")
        print("=" * 50)

        # Connect to pipes
        self.tx_pipe = self.connect_pipe_safe(r"\\.\pipe\win_to_linux")
        self.rx_pipe = self.connect_pipe_safe(r"\\.\pipe\linux_to_win")

        if not self.tx_pipe:
            print("❌ Failed to connect TX pipe")
            print("💡 Check VMware settings:")
            print("   Serial Port 1: \\\\.\\pipe\\win_to_linux")
            return False

        if not self.rx_pipe:
            print("❌ Failed to connect RX pipe")
            print("💡 Check VMware settings:")
            print("   Serial Port 2: \\\\.\\pipe\\linux_to_win")
            return False

        print("✅ Both pipes connected")
        print("🚀 Starting bidirectional session...")

        self.running = True

        # Start receiver thread
        rx_thread = threading.Thread(
            target=self.receive_from_linux,
            name="Linux_RX_Thread",
        )
        rx_thread.daemon = True
        rx_thread.start()
        self.threads.append(rx_thread)

        # Main send loop
        self.send_to_linux()

        return True

    def send_to_linux(self):
        """Send data Windows → Linux"""
        counter = 1
        last_send = time.time()

        print("📤 Sending from Windows to Linux")

        try:
            while self.running:
                current_time = time.time()

                # Send every 2 seconds
                if current_time - last_send >= 2.0:
                    message = {
                        "id": counter,
                        "timestamp": datetime.now().isoformat(),
                        "source": "Windows_Host",
                        "direction": "Win_to_Linux",
                        "data": f"Windows_Message_{counter:03d}",
                        "system_info": {
                            "platform": "Windows",
                            "counter": counter,
                        },
                    }

                    try:
                        data = (json.dumps(message) + "\n").encode("utf-8")

                        # Asynchronous send
                        overlapped = pywintypes.OVERLAPPED()
                        overlapped.hEvent = win32event.CreateEvent(
                            None, True, False, None
                        )

                        win32file.WriteFile(self.tx_pipe, data, overlapped)
                        result = win32event.WaitForSingleObject(
                            overlapped.hEvent, 1000
                        )

                        if result == win32event.WAIT_OBJECT_0:
                            print(
                                f"[TX {counter:03d}] Windows → Linux: {message['data']}"
                            )
                            counter += 1
                            last_send = current_time
                        else:
                            print(f"⚠️ Send timeout: {counter}")

                        win32api.CloseHandle(overlapped.hEvent)

                    except Exception as e:
                        print(f"❌ Send error: {e}")
                        break

                time.sleep(0.1)

        except KeyboardInterrupt:
            pass
        finally:
            print("📤 Windows sender stopped")

    def receive_from_linux(self):
        """Receive data Linux → Windows"""
        buffer = b""

        print("📥 Waiting for data from Linux")

        try:
            while self.running:
                try:
                    overlapped = pywintypes.OVERLAPPED()
                    overlapped.hEvent = win32event.CreateEvent(
                        None, True, False, None
                    )

                    # Asynchronous read
                    win32file.ReadFile(self.rx_pipe, 1024, overlapped)
                    result = win32event.WaitForSingleObject(
                        overlapped.hEvent, 100
                    )

                    if result == win32event.WAIT_OBJECT_0:
                        bytes_read = win32file.GetOverlappedResult(
                            self.rx_pipe, overlapped, False
                        )
                        if bytes_read > 0:
                            # Get the actual data
                            try:
                                _, data = win32file.ReadFile(
                                    self.rx_pipe, bytes_read
                                )
                                buffer += data

                                # Process line by line
                                while b"\n" in buffer:
                                    line, buffer = buffer.split(b"\n", 1)
                                    if line:
                                        msg = line.decode(
                                            "utf-8", errors="ignore"
                                        ).strip()
                                        if msg:
                                            timestamp = datetime.now().strftime(
                                                "%H:%M:%S.%f"
                                            )[:-3]
                                            print(
                                                f"[RX {timestamp}] Linux → Windows: {msg}"
                                            )
                            except:
                                # Handle overlapping read
                                pass

                    win32api.CloseHandle(overlapped.hEvent)

                except pywintypes.error as e:
                    if e.args[0] == 109:  # ERROR_BROKEN_PIPE
                        print("⚠️ Pipe closed on Linux side")
                        break
                    elif e.args[0] == 232:  # ERROR_NO_DATA
                        pass  # No data (normal)

                time.sleep(0.01)

        except Exception as e:
            print(f"❌ Receive error: {e}")
        finally:
            print("📥 Linux receiver stopped")

    def test_pipes(self):
        """Check pipe connectivity"""
        print("=== Dual-Pipe Connection Test ===")

        pipes = [
            (r"\\.\pipe\win_to_linux", "Windows → Linux"),
            (r"\\.\pipe\linux_to_win", "Linux → Windows"),
        ]

        for pipe_name, description in pipes:
            print(f"\n🔍 Testing: {description}")
            print(f"   Pipe: {pipe_name}")

            try:
                win32pipe.WaitNamedPipe(pipe_name, 1000)
                print("   ✅ Pipe detected")

                handle = self.connect_pipe_safe(pipe_name, 2000)
                if handle:
                    print("   ✅ Connection successful")
                    win32file.CloseHandle(handle)
                else:
                    print("   ❌ Connection failed")

            except Exception as e:
                print(f"   ❌ Pipe not found: {e}")
                print("   💡 Check VMware settings")

def main():
    """Main execution"""
    comm = DualPipeComm()

    print("🔧 VMware Dual-Pipe Communication Tool")
    print("=" * 40)
    print("1. Test pipe connections")
    print("2. Start bidirectional session")
    print("3. VMware setup guide")
    print("4. Exit")

    while True:
        try:
            choice = input("\nChoose an option (1-4): ").strip()

            if choice == "1":
                comm.test_pipes()

            elif choice == "2":
                if comm.dual_pipe_communication():
                    print("✅ Session ended")
                else:
                    print("❌ Failed to start communication")

            elif choice == "3":
                print("\n=== VMware Setup Guide ===")
                print("【Serial Port 1】")
                print("  Connection type: Use named pipe")
                print("  Pipe name: \\\\.\\pipe\\win_to_linux")
                print("  Pipe endpoint: Server")
                print("  I/O mode: Application")
                print()
                print("【Serial Port 2】")
                print("  Connection type: Use named pipe")
                print("  Pipe name: \\\\.\\pipe\\linux_to_win")
                print("  Pipe endpoint: Server")
                print("  I/O mode: Application")
                print()
                print("【Linux Side】")
                print("  /dev/ttyS0 ← Receive from Windows")
                print("  /dev/ttyS1 → Send to Windows")

            elif choice == "4":
                print("👋 Exiting")
                break

            else:
                print("❌ Choose 1–4")

        except KeyboardInterrupt:
            print("\n👋 Exiting")
            break

    comm.cleanup()

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Enhanced Linux Serial Communication Tool
VMware Dual Pipe Support – Full Duplex
"""

import serial
import time
import random
import json
import os
import sys
import threading
from datetime import datetime

class LinuxSerialComm:
    def __init__(self):
        self.running = False
        self.threads = []

    def check_serial_devices(self):
        """Check serial devices"""
        devices = {
            '/dev/ttyS0': 'Receive from Windows',
            '/dev/ttyS1': 'Send to Windows'
        }

        print("🔍 Checking serial devices:")
        status = {}

        for device, purpose in devices.items():
            if os.path.exists(device):
                readable = os.access(device, os.R_OK)
                writable = os.access(device, os.W_OK)
                status[device] = readable and writable

                status_icon = "✅" if status[device] else "❌"
                print(f"  {status_icon} {device}: {purpose}")
                print(f"      Readable={readable}, Writable={writable}")

                if not status[device]:
                    print(f"      💡 Fix permissions: sudo chmod 666 {device}")
            else:
                status[device] = False
                print(f"  ❌ {device}: device does not exist")
                print(f"      💡 Check your VMware settings")

        all_ready = all(status.values())

        if not all_ready:
            print("\n⚠️ How to fix permissions:")
            print("  sudo usermod -a -G dialout $USER")
            print("  logout && login  # re-login required")
            print("or")
            print("  sudo chmod 666 /dev/ttyS*")

        return all_ready

    def send_to_windows(self, device='/dev/ttyS1', test_mode=True):
        """Send Linux → Windows"""
        description = f"Sending Linux → Windows ({device})"

        try:
            print(f"📤 {description} starting...")
            if test_mode:
                print("🧪 Test mode: Ctrl+C to stop")

            with serial.Serial(device, 9600, timeout=1) as ser:
                counter = 1

                while self.running:
                    if test_mode:
                        test_patterns = [
                            f"LINUX_TO_WIN,{counter},{random.randint(20, 30)}.{random.randint(0, 99):02d},CPU_TEMP",
                            f"UBUNTU_STATUS,{counter},RUNNING,{datetime.now().strftime('%H:%M:%S')}",
                            f"SYSTEM_DATA,{counter},{random.randint(0, 100)},MEMORY_USAGE_PERCENT",
                            f"LINUX_HEARTBEAT,{counter},ALIVE_FROM_LINUX",
                            f"PROCESS_COUNT,{counter},{random.randint(100, 300)},TOTAL_PROCESSES",
                            f"NETWORK_STAT,{counter},{random.randint(1000, 9999)},PACKETS_PER_SEC",
                            json.dumps({
                                "id": counter,
                                "timestamp": datetime.now().isoformat(),
                                "source": "Linux_Guest",
                                "direction": "Linux_to_Windows",
                                "system": {
                                    "hostname": "ubuntu-vm",
                                    "uptime": random.randint(3600, 86400),
                                    "load_avg": round(random.uniform(0.1, 2.0), 2),
                                    "free_memory": random.randint(1000, 4000)
                                },
                                "sensors": {
                                    "cpu_temp": random.randint(40, 70),
                                    "fan_speed": random.randint(1000, 3000),
                                    "voltage": round(random.uniform(11.8, 12.2), 1),
                                }
                            })
                        ]
                        message = random.choice(test_patterns) + "\r\n"
                    else:
                        message = input(f"[{counter:03d}] Message: ").strip()
                        if message.lower() == 'quit':
                            break
                        message = f"MANUAL,{counter},{message},{datetime.now().strftime('%H:%M:%S')}\r\n"

                    ser.write(message.encode('utf-8'))
                    print(f"[TX {counter:03d}] Linux → Windows: {message.strip()}")
                    counter += 1

                    if test_mode:
                        time.sleep(random.uniform(1.5, 3.0))

        except KeyboardInterrupt:
            print(f"\n✅ Stopped {description}")
        except serial.SerialException as e:
            print(f"❌ Serial error ({device}): {e}")
            print("💡 Check device permissions and VMware settings")
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
        finally:
            self.running = False

    def receive_from_windows(self, device='/dev/ttyS0'):
        """Monitor Windows → Linux reception"""
        try:
            print(f"📥 Starting Windows → Linux monitor ({device})")

            with serial.Serial(device, 9600, timeout=1) as ser:
                buffer = ""

                while self.running:
                    if ser.in_waiting > 0:
                        data = ser.read(ser.in_waiting).decode('utf-8', errors='ignore')
                        buffer += data

                        while '\n' in buffer:
                            line, buffer = buffer.split('\n', 1)
                            line = line.strip()
                            if line:
                                timestamp = datetime.now().strftime('%H:%M:%S.%f')[:-3]
                                print(f"[RX {timestamp}] Windows → Linux: {line}")

                    time.sleep(0.01)

        except serial.SerialException as e:
            print(f"❌ Receive error ({device}): {e}")
        except Exception as e:
            print(f"❌ Unexpected receive error: {e}")
        finally:
            print(f"📥 Monitor stopped ({device})")

    def bidirectional_test(self):
        """Bidirectional communication test"""
        print("🔄 Starting bidirectional communication test")
        print("📤 /dev/ttyS1 → Windows")
        print("📥 /dev/ttyS0 ← Windows")
        print("Press Ctrl+C to stop")
        print("=" * 40)

        self.running = True

        rx_thread = threading.Thread(
            target=self.receive_from_windows,
            args=('/dev/ttyS0',),
            name="Windows_RX"
        )
        rx_thread.daemon = True
        rx_thread.start()
        self.threads.append(rx_thread)

        self.send_to_windows('/dev/ttyS1', test_mode=True)

    def manual_communication(self):
        """Manual communication mode"""
        print("💬 Manual communication mode")
        print("📤 Send: /dev/ttyS1 → Windows")
        print("📥 Receive: /dev/ttyS0 ← Windows")
        print("Type a message to send, 'quit' to exit")
        print("=" * 40)

        self.running = True

        rx_thread = threading.Thread(
            target=self.receive_from_windows,
            args=('/dev/ttyS0',),
            name="Manual_RX"
        )
        rx_thread.daemon = True
        rx_thread.start()
        self.threads.append(rx_thread)

        self.send_to_windows('/dev/ttyS1', test_mode=False)

    def performance_test(self):
        """Performance test"""
        print("⚡ Starting performance test")
        device = '/dev/ttyS1'
        duration = 30

        try:
            with serial.Serial(device, 9600, timeout=1) as ser:
                start_time = time.time()
                bytes_sent = 0
                packets_sent = 0

                print(f"📊 Measuring for {duration} seconds...")

                while time.time() - start_time < duration:
                    test_data = f"PERF_{packets_sent:06d}_" + "L" * 80 + "\r\n"
                    ser.write(test_data.encode('utf-8'))

                    bytes_sent += len(test_data)
                    packets_sent += 1

                    if packets_sent % 50 == 0:
                        elapsed = time.time() - start_time
                        bps = bytes_sent / elapsed if elapsed > 0 else 0
                        print(f"📈 Progress: {packets_sent} packets, {bps:.1f} bytes/sec")

                    time.sleep(0.02)  # 50Hz

                elapsed = time.time() - start_time
                print("\n📊 Performance test results:")
                print(f"  Packets sent: {packets_sent:,}")
                print(f"  Bytes sent: {bytes_sent:,}")
                print(f"  Elapsed time: {elapsed:.2f} sec")
                print(f"  Average throughput: {bytes_sent/elapsed:.1f} bytes/sec")
                print(f"  Theoretical (9600 baud): {9600/10:.1f} bytes/sec")
                print(f"  Efficiency: {(bytes_sent/elapsed)/(9600/10)*100:.1f}%")

        except Exception as e:
            print(f"❌ Performance test error: {e}")

    def stop_all(self):
        """Stop all communication"""
        self.running = False

        for thread in self.threads:
            if thread.is_alive():
                thread.join(timeout=1.0)

        print("✅ All communication stopped")

def main():
    """Main execution"""
    comm = LinuxSerialComm()

    print("=== Linux Serial Communication Tool (Enhanced) ===")
    print("🐧 VMware Guest - Dual Pipe Support")
    print(f"📅 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 55)

    if not comm.check_serial_devices():
        print("\n❌ Serial devices not ready")
        print("Please address the issues above and try again")
        return

    try:
        while True:
            print("\n🎛️ Menu:")
            print("1. Check device status")
            print("2. Send test from Linux to Windows")
            print("3. Monitor Windows → Linux")
            print("4. Bidirectional communication test")
            print("5. Manual communication mode")
            print("6. Performance test")
            print("7. VMware setup guide")
            print("8. Exit")

            choice = input("\nChoose an option (1-8): ").strip()

            if choice == "1":
                comm.check_serial_devices()
            elif choice == "2":
                comm.running = True
                comm.send_to_windows('/dev/ttyS1', test_mode=True)
            elif choice == "3":
                comm.running = True
                comm.receive_from_windows('/dev/ttyS0')
            elif choice == "4":
                comm.bidirectional_test()
            elif choice == "5":
                comm.manual_communication()
            elif choice == "6":
                comm.performance_test()
            elif choice == "7":
                print("\n=== VMware Setup Guide ===")
                print("[Required Configuration]")
                print("Serial Port 1:")
                print("  Pipe name: \\\\.\\pipe\\win_to_linux")
                print("  Use: Windows → Linux (/dev/ttyS0)")
                print()
                print("Serial Port 2:")
                print("  Pipe name: \\\\.\\pipe\\linux_to_win")
                print("  Use: Linux → Windows (/dev/ttyS1)")
                print()
                print("[Linux Permissions]")
                print("sudo usermod -a -G dialout $USER && logout")
                print("or")
                print("sudo chmod 666 /dev/ttyS*")
            elif choice == "8":
                print("👋 Exiting")
                break
            else:
                print("❌ Please choose 1-8")

    except KeyboardInterrupt:
        print("\n👋 Exiting")
    finally:
        comm.stop_all()

if __name__ == "__main__":
    main()

Windows App
  ↓ Send     ↑ Receive
\\.\pipe\host_to_guest  \\.\pipe\guest_to_host
  ↓ VMware   ↑ VMware
/dev/ttyS0    /dev/ttyS1
  ↓ Receive  ↑ Send
Linux App

#!/usr/bin/env python3
"""Windows side - using a physical COM port"""
import serial
import threading
import time
import json
from datetime import datetime

def windows_communication():
    # Port exposed by VMware
    port = "COM3"  # or COM4

    def send_data():
        with serial.Serial(port, 9600, timeout=1) as ser:
            counter = 1
            while True:
                message = {
                    "id": counter,
                    "source": "Windows",
                    "data": f"Win_Message_{counter:03d}",
                    "timestamp": datetime.now().isoformat()
                }

                data = (json.dumps(message) + "\n").encode('utf-8')
                ser.write(data)
                print(f"[TX] Windows({port}) → Linux: {message['data']}")
                counter += 1
                time.sleep(2)

    def receive_data():
        with serial.Serial(port, 9600, timeout=1) as ser:
            while True:
                if ser.in_waiting > 0:
                    data = ser.readline().decode('utf-8', errors='ignore').strip()
                    if data:
                        print(f"[RX] Linux → Windows({port}): {data}")
                time.sleep(0.01)

    try:
        choice = input("1: Send only, 2: Receive only, 3: Bidirectional: ")
        if choice == "1":
            send_data()
        elif choice == "2":
            receive_data()
        elif choice == "3":
            rx_thread = threading.Thread(target=receive_data, daemon=True)
            rx_thread.start()
            send_data()
    except KeyboardInterrupt:
        print("Communication ended")

if __name__ == "__main__":
    windows_communication()

#!/usr/bin/env python3
"""Linux side - VMware virtual serial port"""
import serial
import threading
import time
import json
from datetime import datetime

def linux_communication():
    port = "/dev/ttyS0"  # Detected by VMware

    def receive_data():
        with serial.Serial(port, 9600, timeout=1) as ser:
            while True:
                if ser.in_waiting > 0:
                    data = ser.readline().decode('utf-8', errors='ignore').strip()
                    if data:
                        print(f"[RX] Windows → Linux({port}): {data}")
                time.sleep(0.01)

    def send_data():
        with serial.Serial(port, 9600, timeout=1) as ser:
            counter = 1
            while True:
                message = {
                    "id": counter,
                    "source": "Linux",
                    "data": f"Linux_Message_{counter:03d}",
                    "timestamp": datetime.now().isoformat()
                }

                data = (json.dumps(message) + "\n").encode('utf-8')
                ser.write(data)
                print(f"[TX] Linux({port}) → Windows: {message['data']}")
                counter += 1
                time.sleep(2)

    try:
        choice = input("1: Send only, 2: Receive only, 3: Bidirectional: ")
        if choice == "1":
            send_data()
        elif choice == "2":
            receive_data()
        elif choice == "3":
            rx_thread = threading.Thread(target=receive_data, daemon=True)
            rx_thread.start()
            send_data()
    except KeyboardInterrupt:
        print("Communication ended")

if __name__ == "__main__":
    linux_communication()

python network_serial_bridge.py  # Start the receiver server

python enhanced_linux_sender.py  # Send via TCP

python network_serial_bridge.py
# Select Menu 1 when prompted

#!/usr/bin/env python3
"""
Bridge to send serial data over a network.
Supports transferring serial data from Windows to Linux.
"""

import socket
import serial
import threading
import time
import json
from datetime import datetime

class NetworkSerialBridge:
    def __init__(self):
        self.running = False

    def serial_to_network_server(self, serial_port, baudrate, network_port=12345):
        """Transfer data from the serial port to the network (server side)."""
        print("Serial → Network server started")
        print(f"Serial Port: {serial_port} ({baudrate} baud)")
        print(f"Network Port: {network_port}")
        print("Waiting for client connection...")

        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_socket.bind(('0.0.0.0', network_port))
        server_socket.listen(1)

        try:
            with serial.Serial(serial_port, baudrate, timeout=1) as ser:
                while True:
                    client_socket, addr = server_socket.accept()
                    print(f"Client connected: {addr}")

                    try:
                        while True:
                            if ser.in_waiting > 0:
                                data = ser.read(ser.in_waiting)
                                client_socket.send(data)
                                print(f"Transferred: {len(data)} bytes → {addr}")

                            time.sleep(0.01)

                    except (ConnectionResetError, BrokenPipeError):
                        print(f"Client disconnected: {addr}")
                        client_socket.close()

        except KeyboardInterrupt:
            print("Server stopped")
        finally:
            server_socket.close()

    def network_to_serial_client(self, target_host, network_port, serial_port, baudrate):
        """Transfer data from the network to the serial port (client side)."""
        print("Network → Serial client started")
        print(f"Destination: {target_host}:{network_port}")
        print(f"Serial Port: {serial_port} ({baudrate} baud)")

        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.connect((target_host, network_port))
                print("Connected to server")

                with serial.Serial(serial_port, baudrate, timeout=1) as ser:
                    while True:
                        data = sock.recv(1024)
                        if not data:
                            break

                        ser.write(data)
                        print(f"Received: {len(data)} bytes → {serial_port}")

        except KeyboardInterrupt:
            print("Client stopped")
        except Exception as e:
            print(f"Error: {e}")

    def send_test_data_to_network(self, target_host, network_port):
        """Send test data over the network."""
        print(f"Sending test data → {target_host}:{network_port}")

        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.connect((target_host, network_port))
                print("Connection successful")

                counter = 1
                while True:
                    test_data = {
                        "id": counter,
                        "timestamp": datetime.now().isoformat(),
                        "data": f"TEST_MESSAGE_{counter:03d}",
                        "source": "Windows_Client"
                    }

                    message = json.dumps(test_data) + "\n"
                    sock.send(message.encode("utf-8"))

                    print(f"[{counter:03d}] Sent: {message.strip()}")
                    counter += 1

                    time.sleep(2)

        except KeyboardInterrupt:
            print("Transmission stopped")
        except Exception as e:
            print(f"Error: {e}")

    def receive_network_data(self, network_port=12345):
        """Receive data from the network (server)."""
        print(f"Data receive server started (port: {network_port})")
        print("Waiting for connection...")

        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_socket.bind(('0.0.0.0', network_port))
        server_socket.listen(1)

        try:
            while True:
                client_socket, addr = server_socket.accept()
                print(f"Client connected: {addr}")

                try:
                    while True:
                        data = client_socket.recv(1024)
                        if not data:
                            break

                        timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
                        decoded = data.decode("utf-8", errors="ignore").strip()
                        print(f"[{timestamp}] Received: {decoded}")

                except ConnectionResetError:
                    print(f"Client disconnected: {addr}")
                finally:
                    client_socket.close()

        except KeyboardInterrupt:
            print("Receive server stopped")
        finally:
            server_socket.close()

def main():
    bridge = NetworkSerialBridge()

    while True:
        print("\n=== Serial Communication Bridge via Network ===")
        print("1. Serial → Network (Server)")
        print("2. Network → Serial (Client)")
        print("3. Send Test Data (Network)")
        print("4. Receive Data Server")
        print("5. Usage Guide")
        print("6. Exit")

        choice = input("\nSelect (1-6): ").strip()

        if choice == "1":
            # Select serial port
            import serial.tools.list_ports
            ports = serial.tools.list_ports.comports()

            if not ports:
                print("No serial ports found")
                continue

            print("\nAvailable ports:")
            for i, port in enumerate(ports, 1):
                print(f"{i}. {port.device}")

            try:
                port_choice = int(input("Choose port: ")) - 1
                if 0 <= port_choice < len(ports):
                    serial_port = ports[port_choice].device
                    baudrate = int(input("Baud rate (9600): ") or "9600")
                    network_port = int(input("Network port (12345): ") or "12345")

                    bridge.serial_to_network_server(serial_port, baudrate, network_port)

            except (ValueError, IndexError):
                print("Invalid selection")

        elif choice == "2":
            target_host = input("Destination IP address: ").strip()
            network_port = int(input("Network port (12345): ") or "12345")

            # Select serial port
            import serial.tools.list_ports
            ports = serial.tools.list_ports.comports()

            if not ports:
                print("No serial ports found")
                continue

            print("\nAvailable ports:")
            for i, port in enumerate(ports, 1):
                print(f"{i}. {port.device}")

            try:
                port_choice = int(input("Choose port: ")) - 1
                if 0 <= port_choice < len(ports):
                    serial_port = ports[port_choice].device
                    baudrate = int(input("Baud rate (9600): ") or "9600")

                    bridge.network_to_serial_client(target_host, network_port, serial_port, baudrate)

            except (ValueError, IndexError):
                print("Invalid selection")

        elif choice == "3":
            target_host = input("Destination IP address: ").strip()
            network_port = int(input("Network port (12345): ") or "12345")

            bridge.send_test_data_to_network(target_host, network_port)

        elif choice == "4":
            network_port = int(input("Receive port (12345): ") or "12345")
            bridge.receive_network_data(network_port)

        elif choice == "5":
            print("\n=== Usage Guide ===")
            print("\n[Sending Data from Windows to Linux]")
            print("1. Prepare Linux to receive:")
            print("   nc -l -p 12345 | tee received_data.txt")
            print("   or")
            print("   python3 thisscript.py (choose Menu 4)")
            print()
            print("2. Send data from Windows:")
            print("   python thisscript.py (choose Menu 3)")
            print("   Enter the IP address of the Linux machine")
            print()
            print("[Network Transfer Between Serial Ports]")
            print("1. On the sending side:")
            print("   Menu 1: Serial → Network")
            print("2. On the receiving side:")
            print("   Menu 2: Network → Serial")
            print()
            print("[Checking IP Address]")
            print("- Windows: ipconfig")
            print("- Linux: ip addr show or ifconfig")
            print("- Make sure both systems are on the same network")

        elif choice == "6":
            print("Exiting")
            break

        else:
            print("Please select 1-6")

if __name__ == "__main__":
    main()

python3 network_serial_bridge.py
# Select Menu 2: Network → Serial