以前、Windowsでシリアル通信の記事を投稿しました。しかし、LinuxであるUbuntuでは、Windowsとは大きく勝手が異なります。特に権限の概念やデバイスファイルの扱いが根本的に違うため、まずはこれらの違いを理解してから実際の通信に入る必要があります。

WindowsとUbuntuの主な違い

重要ポイント： Ubuntuでは一般ユーザーはデフォルトでシリアルポートにアクセスできません。これが初心者が最初に躓く大きな要因です。

ステップ1：権限設定（最重要）

dialoutグループへの追加

シリアルポートにアクセスするには、現在のユーザーをdialoutグループに追加する必要があります：

sudo usermod -aG dialout $USER

⚠️ 重要な注意点

コマンド実行後は必ずログアウト＆再ログイン（または再起動）してください。グループ設定の変更はセッションを再開するまで反映されません。

一時的な回避方法： 再ログインせずに現在のセッションで試したい場合：

newgrp dialout

ただし、これは一時的な措置で、新しいターミナルでは再び実行が必要です。

注意： 自動ログインを有効にしている場合は、ログアウト&再ログインではなく、再起動が必要な場合があります。

ログアウト・ログインの方法

GUI環境の場合：

1. 画面右上の電源ボタンまたはユーザー名をクリック
2. 「ログアウト」を選択
3. ログイン画面でパスワードを入力して再ログイン

コマンドライン環境の場合：

# 現在のセッションからログアウト
exit

# またはCtrl+Dでも可能

確実な方法（再起動）：

sudo reboot

設定確認

ログイン後、以下のコマンドでdialoutグループに所属しているか確認：

groups

出力例：

your_username adm dialout cdrom sudo dip plugdev lpadmin lxd sambashare

dialoutが含まれていれば設定完了です。

ステップ2：シリアルポートの確認

基本的な確認方法

方法1：デバイスファイル一覧

ls /dev/ttyUSB* /dev/ttyS*

• ttyUSB0 → USB接続のシリアルアダプター（PL2303、FTDIなど）
• ttyS0 → 内蔵COMポート（または仮想COMポート）

方法2：デバイス認識ログの確認 USBシリアルアダプターが正しく認識されているかログで確認：

sudo dmesg | grep tty

出力例：

[    0.597201] printk: legacy console [tty0] enabled
[    1.298837] 00:05: ttyS0 at I/O 0x3f8 (irq = 4, base_baud = 115200) is a 16550A

※このコマンドは確認用です。何かを「有効にする」コマンドではありません。

詳細情報の取得

より詳しい情報が必要な場合：

# USB接続デバイスの一覧
lsusb

# デバイスの詳細情報
udevadm info -q all -n /dev/ttyUSB0

# または
sudo udevadm info /dev/ttyUSB0

ステップ3：環境別の注意点

物理マシンの場合

通常は上記の手順で問題ありません。USB-シリアルアダプターを挿入すると、自動的に/dev/ttyUSB0として認識されます。

仮想マシン（VMware/VirtualBox）の場合

上記はWindowsホストのCOMポートの状態です。仮想マシンでは追加の設定が必要な場合があります：

VMwareの場合：

1. 「仮想マシン設定」→「シリアルポート」を追加
2. 「物理COMに接続」または「ソケットへ」を選択
3. USB-シリアルアダプターの場合：「VM」→「Removable Devices」→「USB Device」→「Connect」

VirtualBoxの場合：

1. 「設定」→「シリアルポート」を有効化
2. USB-シリアルアダプターの場合：USBフィルターを設定

ステップ4：アクセステスト

権限が正しく設定されているか確認：

ls -l /dev/ttyS0

出力例：

crw-rw---- 1 root dialout 4, 64 Jun  5 17:40 /dev/ttyS0

• dialoutグループに読み書き権限（rw-）があることを確認
• 自分がdialoutグループに所属していることを確認

トラブルシューティング

よくある問題と解決方法

問題1：「Permission denied」エラー

解決策：dialoutグループへの追加＋ログアウト＆再ログイン

問題2：デバイスが見つからない

# デバイス認識の確認
dmesg | grep -i usb
sudo dmesg | grep tty

問題3：仮想マシンでポートが表示されない

解決策：VM設定でシリアルポートまたはUSBデバイスの接続を確認

まとめ

Ubuntuでシリアル通信を始める前に必要な準備：

1. dialoutグループへの追加（最重要）
2. ログアウト＆再ログイン
3. シリアルポートの確認
4. 権限テスト

これらの準備が完了すれば、Pythonのpyserialライブラリやその他のツールを使ってシリアル通信を開始できます。

実際にPythonを使ったシリアル通信のサンプルコードを紹介しています。

参考リンク：

• Windowsでのシリアル通信（前回記事）
• GitHubページ

🔄 VMware仮想環境での真の双方向通信実現方法

重要な発見: VMwareの制限を巧妙に回避して、無料で真の双方向通信を実現する方法

🎯 問題の背景

従来の限界

• VMwareの仮想シリアルポートは通常1つのみ
• Host-Guest間で真の双方向通信が困難
• 有料のシリアルポートツールが必要

解決のアイデア

VMwareの「自動検出」機能を逆手に取り、意図的にポートを占有することで複数ポートを確保する画期的な方法を発見しました。それは最初にWindows側でcom1とcom3を使用状態にしておくことです。その後VMwareで自動検出にするのです。前提として仮想シリアルポートのペアを2つ作成しておくことです。実際に確認すると・・・

sudo dmesg | grep tty
[sudo] mamu のパスワード: 
[    0.624149] printk: legacy console [tty0] enabled
[    1.336023] 00:05: ttyS0 at I/O 0x3f8 (irq = 4, base_baud = 115200) is a 16550A
[    1.373873] 00:06: ttyS1 at I/O 0x2f8 (irq = 3, base_baud = 115200) is a 16550A

VMwareで2つのシリアルポートが正常に認識されている

• /dev/ttyS0 (I/O 0x3f8, IRQ 4)
• /dev/ttyS1 (I/O 0x2f8, IRQ 3)

これは確実にWindows側でCOM2とCOM4（またはCOM1とCOM3）が設定されていることを示しています！

もしくは以下のスクリプトを使用します。

🔧 実装方法

ステップ1: ポート占有ツールの作成

#!/usr/bin/env python3
"""
VMware用ポート占有ツール
自動検出を操作してCOM2とCOM4を選択させる
"""

import serial
import time
from datetime import datetime

class PortOccupier:
    def __init__(self):
        self.ports = {}
        self.running = False
    
    def occupy_ports(self, port_list):
        """指定されたポートを占有"""
        print(f"🔧 ポート占有ツール開始")
        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"✅ {port} を占有しました")
            except serial.SerialException as e:
                print(f"❌ {port} の占有に失敗: {e}")
        
        if self.ports:
            print("\n🎯 VMware設定手順:")
            print("1. VMwareの仮想マシン設定を開く")
            print("2. シリアルポート設定で「自動検出」を実行")
            print("3. 占有されていないポートが自動選択される")
            print("4. 設定完了後、'q' + Enter でツールを終了")
            
            print(f"\n📋 占有中のポート: {list(self.ports.keys())}")
            
            # 入力待ちループ
            while self.running:
                try:
                    user_input = input("終了するには 'q' + Enter: ").strip().lower()
                    if user_input == 'q':
                        break
                except KeyboardInterrupt:
                    break
    
    def release_ports(self):
        """全ポートを解放"""
        print("\n🔌 ポート解放中...")
        for port, ser in self.ports.items():
            try:
                if ser.is_open:
                    ser.close()
                print(f"✅ {port} を解放しました")
            except Exception as e:
                print(f"❌ {port} の解放エラー: {e}")
        
        self.ports.clear()
        self.running = False
        print("✅ 全ポート解放完了")

def main():
    """メイン実行"""
    print("🔧 VMware用ポート占有ツール")
    print("🎯 目的: VMwareの自動検出でCOM2とCOM4を選択させる")
    
    # COM1とCOM3を占有してCOM2とCOM4を自動選択させる
    target_ports = ['COM1', 'COM3']
    
    occupier = PortOccupier()
    
    try:
        occupier.occupy_ports(target_ports)
    except KeyboardInterrupt:
        print("\n⚠️ 強制終了されました")
    finally:
        occupier.release_ports()

if __name__ == "__main__":
    main()

ステップ2: VMware設定の操作

1. ポート占有ツール実行
   python port_occupier.py
2. VMware仮想マシン設定
   • 仮想マシン設定を開く
   • シリアルポート → 「物理シリアルポートを使用する」
   • 自動検出を実行
   • 重要: COM1, COM3が占有されているため、VMwareは自動的にCOM2を選択
3. 2つ目のシリアルポート追加
   • 「ハードウェアの追加」→「シリアルポート」
   • 再度自動検出を実行
   • COM4が自動選択される

ステップ3: 結果の確認

この手法により以下の構成が実現：

Windows Host側:
- COM2 ↔ /dev/ttyS0 (Linux)
- COM4 ↔ /dev/ttyS1 (Linux)

Linux Guest側:
- /dev/ttyS0 ↔ COM2 (Windows)
- /dev/ttyS1 ↔ COM4 (Windows)

🎮 双方向通信の設定例

Windows側ダッシュボード設定

📥 RX: COM2  (Linux側からのメッセージを受信)
📤 TX: COM4  (Linux側にメッセージを送信)
送受信分離: ✅ チェック

Linux側ダッシュボード設定

📥 RX: /dev/ttyS1  (Windows側からのメッセージを受信)
📤 TX: /dev/ttyS0  (Windows側にメッセージを送信)
送受信分離: ✅ チェック

🔄 通信フロー

真の双方向通信:
Windows → COM4 → VMware → /dev/ttyS0 → Linux
Linux → /dev/ttyS1 → VMware → COM2 → Windows

💡 この方法の利点

✅ 完全無料

• 有料のシリアルポートツール不要
• VMwareの標準機能のみ使用

✅ 安定性

• VMwareネイティブ機能による安定した通信
• com0comのような不安定さがない

✅ 汎用性

• 任意のシリアル通信アプリケーションで使用可能
• プログラミング言語に依存しない

⚠️ 注意点

トラブルシューティング

1. ポート占有失敗
   # ポート使用状況確認
   netstat -an | findstr :COM
2. VMware自動検出が期待通りにならない
   • ポート占有ツールが正常に動作しているか確認
   • VMware設定をリセットして再試行
3. 通信が片方向のみ
   • 送受信ポートの設定を再確認
   • VMwareのシリアルポート設定を見直し

💡 この発見により、シリアル通信開発の敷居が大幅に下がり、より多くの開発者が気軽にシリアル通信を活用できるようになります。

🔄 VMware仮想環境での双方向通信実現方法【改訂版】

重要な修正: 従来のポート占有方法よりも、名前付きパイプを使用する方法が最も安定して推奨されることが判明しました。

🎯 問題の背景と解決策の比較

従来の課題

• VMwareの仮想シリアルポートは設定が複雑
• 物理シリアルポートの自動検出は不安定
• Host-Guest間での真の双方向通信が困難

📊 解決策の比較表

🏆 推奨方法1: 名前付きパイプ（最安定）

ステップ1: VMware設定

VM設定 → ハードウェア追加 → シリアルポート

【シリアルポート1設定】
✅ 接続方法: 名前付きパイプを使用
✅ パイプ名: \\.\pipe\vmware_tx
✅ パイプの端: サーバー
✅ I/Oモード: アプリケーション
✅ 仮想マシンの電源投入時に接続: チェック

【シリアルポート2設定】
✅ 接続方法: 名前付きパイプを使用  
✅ パイプ名: \\.\pipe\vmware_rx
✅ パイプの端: サーバー
✅ I/Oモード: アプリケーション
✅ 仮想マシンの電源投入時に接続: チェック

ステップ2: Linux側確認

# シリアルデバイス確認
sudo dmesg | grep tty
# 期待結果:
# ttyS0 at I/O 0x3f8 (irq = 4) - パイプ1
# ttyS1 at I/O 0x2f8 (irq = 3) - パイプ2

# 権限設定
sudo chmod 666 /dev/ttyS0 /dev/ttyS1
# または
sudo usermod -a -G dialout $USER  # 再ログイン必要

ステップ3: Windows側でパイプアクセス

#!/usr/bin/env python3
"""
VMware名前付きパイプ双方向通信
Windows側実装
"""

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):
        """名前付きパイプに接続"""
        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_name}: {e}")
            return None
    
    def bidirectional_communication(self):
        """双方向通信実行"""
        print("=== VMware名前付きパイプ双方向通信 ===")
        
        # パイプ接続
        tx_pipe = self.connect_to_pipe(r"\\.\pipe\vmware_tx")  # 送信用
        rx_pipe = self.connect_to_pipe(r"\\.\pipe\vmware_rx")  # 受信用
        
        if not tx_pipe or not rx_pipe:
            print("パイプ接続失敗")
            return
        
        print("✅ 両方向パイプ接続成功")
        
        self.running = True
        
        # 受信スレッド開始
        rx_thread = threading.Thread(target=self.receive_data, args=(rx_pipe,))
        rx_thread.daemon = True
        rx_thread.start()
        
        # 送信ループ
        self.send_data(tx_pipe)
        
        # クリーンアップ
        self.running = False
        win32file.CloseHandle(tx_pipe)
        win32file.CloseHandle(rx_pipe)
    
    def send_data(self, pipe_handle):
        """データ送信"""
        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("送信停止")
    
    def receive_data(self, pipe_handle):
        """データ受信"""
        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"受信エラー: {e}")

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

1. VMware記事の主な修正点

❌ 従来の問題

• ポート占有という技巧的手法を「画期的」として推奨
• 名前付きパイプの優位性に言及なし
• 環境依存性や不安定要因を軽視

✅ 修正版の改善

• 名前付きパイプを最推奨として明記
• 各手法の安定性・実用性を客観評価
• 環境別の具体的な推奨方法を提示

2. 実用性重視の改善

✅ 修正版の改善（続き）

• トラブルシューティングガイドを充実
• 実機・VirtualBox・VMwareそれぞれに最適化

3. 新しい汎用テストツールの特徴

🎯 インテリジェント機能

• 実行環境の自動検出（VMware/VirtualBox/物理環境）
• ポートの種類別分類（USB/仮想/物理）
• 環境に応じた推奨設定の自動提案

🔧 実用的機能

• 双方向テストウィザード
• 性能ベンチマーク
• リアルタイム受信監視

環境別の実際の推奨度

実際の使用シナリオ

開発・学習用途

# VirtualBox環境（推奨）
1. HostとGuest間で仮想シリアル設定
2. 汎用テストツールで双方向確認
3. 実際のアプリケーション開発

プロダクション環境

# 実機同士（最安定）
1. USBシリアル変換器×2使用
2. クロスケーブルで直接接続
3. 冗長化設定でフェイルセーフ

テスト・検証用途

# TCP/IPブリッジ（最柔軟）
1. ネットワーク経由で複数環境接続
2. ログ記録・監視機能付き
3. 自動テストスクリプト連携

しかし以下の問題が浮上します。

名前付きパイプは本質的に一方向通信のため、単一パイプでは双方向通信ができません。その場合は、「旧VMware仮想環境での真の双方向通信実現方法」が役立つことになります。もしくは以下の方法で解決します。

VMware双方向通信完全解決版

🎯 問題の整理

現在の状況

• 名前付きパイプ: Windows→Linuxは送信可能
• Linux→Windows: 名前付きパイプでは送信不可
• 必要: 真の双方向通信

🔧 解決策1: デュアルパイプ設定（推奨）

VMware設定方法

シリアルポート1（Windows→Linux用）

VM設定 → シリアルポート1
✅ 接続方法: 名前付きパイプを使用
✅ パイプ名: \\.\pipe\win_to_linux
✅ パイプの端: サーバー
✅ I/Oモード: アプリケーション

シリアルポート2（Linux→Windows用）

VM設定 → シリアルポート2  
✅ 接続方法: 名前付きパイプを使用
✅ パイプ名: \\.\pipe\linux_to_win
✅ パイプの端: サーバー
✅ I/Oモード: アプリケーション

Linux側対応表

🔧 解決策2: TCP/IPハイブリッド（代替推奨）

物理シリアルポート + TCP/IPの組み合わせ

構成

• Windows→Linux: 名前付きパイプ
• Linux→Windows: TCP/IP通信

💻 実装コード

Windows側（デュアルパイプ対応）

#!/usr/bin/env python3
"""
VMware デュアルパイプ双方向通信 - Windows側
真の双方向通信を実現
"""

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
        self.tx_pipe = None  # Windows→Linux（送信）
        self.rx_pipe = None  # Linux→Windows（受信）
        self.threads = []
        
        # 安全終了設定
        atexit.register(self.cleanup)
        signal.signal(signal.SIGINT, self.signal_handler)
    
    def signal_handler(self, signum, frame):
        """安全終了処理"""
        print(f"\n⚠️ 終了シグナル受信")
        self.safe_shutdown()
        sys.exit(0)
    
    def safe_shutdown(self):
        """安全な終了"""
        print("🛑 安全終了処理...")
        self.running = False
        
        # スレッド終了待ち
        for thread in self.threads:
            if thread.is_alive():
                thread.join(timeout=1.0)
        
        self.cleanup()
        print("✅ 終了完了")
    
    def cleanup(self):
        """リソースクリーンアップ"""
        try:
            if self.tx_pipe:
                win32file.CloseHandle(self.tx_pipe)
                self.tx_pipe = None
                print("📤 送信パイプクローズ")
        except:
            pass
        
        try:
            if self.rx_pipe:
                win32file.CloseHandle(self.rx_pipe)
                self.rx_pipe = None
                print("📥 受信パイプクローズ")
        except:
            pass
    
    def connect_pipe_safe(self, pipe_name, timeout=5000):
        """安全なパイプ接続"""
        print(f"🔌 接続試行: {pipe_name}")
        
        try:
            # パイプ待機
            win32pipe.WaitNamedPipe(pipe_name, timeout)
            
            # 非同期モードで接続
            handle = win32file.CreateFile(
                pipe_name,
                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                0, None,
                win32file.OPEN_EXISTING,
                win32file.FILE_FLAG_OVERLAPPED,
                None
            )
            
            print(f"✅ 接続成功: {pipe_name}")
            return handle
            
        except Exception as e:
            print(f"❌ 接続失敗 {pipe_name}: {e}")
            return None
    
    def dual_pipe_communication(self):
        """デュアルパイプ双方向通信"""
        print("=== VMware デュアルパイプ双方向通信 ===")
        print("📤 送信用: \\\\.\\\pipe\\\\win_to_linux")
        print("📥 受信用: \\\\.\\\pipe\\\\linux_to_win")
        print("💡 Ctrl+Cで安全終了")
        print("=" * 50)
        
        # パイプ接続
        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("❌ 送信パイプ接続失敗")
            print("💡 VMware設定確認:")
            print("   シリアルポート1: \\\\.\\\pipe\\\\win_to_linux")
            return False
        
        if not self.rx_pipe:
            print("❌ 受信パイプ接続失敗") 
            print("💡 VMware設定確認:")
            print("   シリアルポート2: \\\\.\\\pipe\\\\linux_to_win")
            return False
        
        print("✅ デュアルパイプ接続成功")
        print("🚀 双方向通信開始...")
        
        self.running = True
        
        # 受信スレッド起動
        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)
        
        # 送信メインループ
        self.send_to_linux()
        
        return True
    
    def send_to_linux(self):
        """Windows→Linux送信"""
        counter = 1
        last_send = time.time()
        
        print("📤 Windows→Linux送信開始")
        
        try:
            while self.running:
                current_time = time.time()
                
                # 2秒間隔で送信
                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')
                        
                        # 非同期送信
                        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"⚠️ 送信タイムアウト: {counter}")
                        
                        win32api.CloseHandle(overlapped.hEvent)
                        
                    except Exception as e:
                        print(f"❌ 送信エラー: {e}")
                        break
                
                time.sleep(0.1)
                
        except KeyboardInterrupt:
            pass
        finally:
            print("📤 Windows送信終了")
    
    def receive_from_linux(self):
        """Linux→Windows受信"""
        buffer = b""
        
        print("📥 Linux→Windows受信待機")
        
        try:
            while self.running:
                try:
                    overlapped = pywintypes.OVERLAPPED()
                    overlapped.hEvent = win32event.CreateEvent(None, True, False, None)
                    
                    # 非同期読み取り
                    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:
                            # データ取得（実際の読み取り）
                            try:
                                # overlappedから実際のデータを取得
                                _, data = win32file.ReadFile(self.rx_pipe, bytes_read)
                                buffer += data
                                
                                # 行単位処理
                                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:
                                # ReadFileが重複する場合の対処
                                pass
                    
                    win32api.CloseHandle(overlapped.hEvent)
                    
                except pywintypes.error as e:
                    if e.args[0] == 109:  # ERROR_BROKEN_PIPE
                        print("⚠️ Linux側パイプ切断")
                        break
                    elif e.args[0] == 232:  # ERROR_NO_DATA
                        pass  # データなし（正常）
                
                time.sleep(0.01)
                
        except Exception as e:
            print(f"❌ 受信エラー: {e}")
        finally:
            print("📥 Linux受信終了")
    
    def test_pipes(self):
        """パイプ接続テスト"""
        print("=== デュアルパイプ接続テスト ===")
        
        pipes = [
            (r"\\.\pipe\win_to_linux", "Windows→Linux"),
            (r"\\.\pipe\linux_to_win", "Linux→Windows")
        ]
        
        for pipe_name, description in pipes:
            print(f"\n🔍 テスト: {description}")
            print(f"   パイプ: {pipe_name}")
            
            try:
                win32pipe.WaitNamedPipe(pipe_name, 1000)
                print(f"   ✅ パイプ検出成功")
                
                handle = self.connect_pipe_safe(pipe_name, 2000)
                if handle:
                    print(f"   ✅ 接続テスト成功")
                    win32file.CloseHandle(handle)
                else:
                    print(f"   ❌ 接続テスト失敗")
                    
            except Exception as e:
                print(f"   ❌ パイプ未検出: {e}")
                print(f"   💡 VMware設定確認が必要")

def main():
    """メイン実行"""
    comm = DualPipeComm()
    
    print("🔧 VMware デュアルパイプ通信ツール")
    print("=" * 40)
    print("1. パイプ接続テスト")
    print("2. 双方向通信開始")  
    print("3. VMware設定ガイド")
    print("4. 終了")
    
    while True:
        try:
            choice = input("\n選択してください (1-4): ").strip()
            
            if choice == "1":
                comm.test_pipes()
            
            elif choice == "2":
                if comm.dual_pipe_communication():
                    print("✅ 通信セッション終了")
                else:
                    print("❌ 通信開始失敗")
            
            elif choice == "3":
                print("\n=== VMware設定ガイド ===")
                print("【シリアルポート1設定】")
                print("  接続方法: 名前付きパイプを使用")
                print("  パイプ名: \\\\.\\\pipe\\\\win_to_linux")
                print("  パイプの端: サーバー")
                print("  I/Oモード: アプリケーション")
                print()
                print("【シリアルポート2設定】")
                print("  接続方法: 名前付きパイプを使用")
                print("  パイプ名: \\\\.\\\pipe\\\\linux_to_win")
                print("  パイプの端: サーバー")
                print("  I/Oモード: アプリケーション")
                print()
                print("【Linux側対応】")
                print("  /dev/ttyS0 ← Windows送信を受信")
                print("  /dev/ttyS1 → Windowsへ送信")
            
            elif choice == "4":
                print("👋 終了します")
                break
            
            else:
                print("❌ 1-4を選択してください")
                
        except KeyboardInterrupt:
            print("\n👋 終了します")
            break
    
    comm.cleanup()

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
改良版Linux用シリアル通信ツール
VMwareデュアルパイプ対応・双方向通信
"""

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):
        """シリアルデバイス確認"""
        devices = {
            '/dev/ttyS0': 'Windows→Linux受信用',
            '/dev/ttyS1': 'Linux→Windows送信用'
        }
        
        print("🔍 シリアルデバイス確認:")
        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}, 書き込み={writable}")
                
                if not status[device]:
                    print(f"      💡 権限修正: sudo chmod 666 {device}")
            else:
                status[device] = False
                print(f"  ❌ {device}: デバイスが存在しません")
                print(f"      💡 VMware設定を確認してください")
        
        all_ready = all(status.values())
        
        if not all_ready:
            print("\n⚠️ 権限修正方法:")
            print("  sudo usermod -a -G dialout $USER")
            print("  logout && login  # 再ログイン必要")
            print("または")
            print("  sudo chmod 666 /dev/ttyS*")
        
        return all_ready
    
    def send_to_windows(self, device='/dev/ttyS1', test_mode=True):
        """Linux→Windows送信"""
        description = f"Linux→Windows送信 ({device})"
        
        try:
            print(f"📤 {description}開始...")
            if test_mode:
                print("🧪 テストモード: Ctrl+Cで停止")
            
            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}] メッセージ: ").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✅ {description}を停止")
        except serial.SerialException as e:
            print(f"❌ シリアルエラー ({device}): {e}")
            print("💡 デバイス権限・VMware設定を確認してください")
        except Exception as e:
            print(f"❌ 予期しないエラー: {e}")
        finally:
            self.running = False
    
    def receive_from_windows(self, device='/dev/ttyS0'):
        """Windows→Linux受信監視"""
        try:
            print(f"📥 Windows→Linux受信監視開始 ({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"❌ 受信エラー ({device}): {e}")
        except Exception as e:
            print(f"❌ 予期しない受信エラー: {e}")
        finally:
            print(f"📥 受信監視終了 ({device})")
    
    def bidirectional_test(self):
        """双方向通信テスト"""
        print("🔄 双方向通信テスト開始")
        print("📤 /dev/ttyS1 → Windows送信")
        print("📥 /dev/ttyS0 ← Windows受信")
        print("Ctrl+Cで停止")
        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):
        """手動通信モード"""
        print("💬 手動通信モード")
        print("📤 送信: /dev/ttyS1 → Windows")
        print("📥 受信: /dev/ttyS0 ← Windows")
        print("メッセージ入力で送信、'quit'で終了")
        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):
        """性能テスト"""
        print("⚡ 性能テスト開始")
        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"📊 {duration}秒間の性能測定...")
                
                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"📈 進行: {packets_sent} packets, {bps:.1f} bytes/sec")
                    
                    time.sleep(0.02)  # 50Hz
                
                # 結果
                elapsed = time.time() - start_time
                print(f"\n📊 性能テスト結果:")
                print(f"  送信パケット: {packets_sent:,}")
                print(f"  送信バイト: {bytes_sent:,}")
                print(f"  実測時間: {elapsed:.2f}秒")
                print(f"  平均スループット: {bytes_sent/elapsed:.1f} bytes/sec")
                print(f"  理論値 (9600baud): {9600/10:.1f} bytes/sec")
                print(f"  効率: {(bytes_sent/elapsed)/(9600/10)*100:.1f}%")
                
        except Exception as e:
            print(f"❌ 性能テストエラー: {e}")
    
    def stop_all(self):
        """全通信停止"""
        self.running = False
        
        for thread in self.threads:
            if thread.is_alive():
                thread.join(timeout=1.0)
        
        print("✅ 全通信停止完了")

def main():
    """メイン実行"""
    comm = LinuxSerialComm()
    
    print("=== Linux Serial Communication Tool (Enhanced) ===")
    print("🐧 VMware Guest - デュアルパイプ対応")
    print(f"📅 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 55)
    
    # 初期チェック
    if not comm.check_serial_devices():
        print("\n❌ シリアルデバイス準備未完了")
        print("上記の対処方法を実行してから再度お試しください")
        return
    
    try:
        while True:
            print("\n🎛️ メニュー:")
            print("1. デバイス状態確認")
            print("2. Linux→Windows送信テスト")
            print("3. Windows→Linux受信監視")
            print("4. 双方向通信テスト")
            print("5. 手動通信モード")
            print("6. 性能テスト")
            print("7. VMware設定ガイド")
            print("8. 終了")
            
            choice = input("\n選択してください (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設定ガイド ===")
                print("【必要な設定】")
                print("シリアルポート1:")
                print("  パイプ名: \\\\.\\\pipe\\\\win_to_linux")
                print("  用途: Windows→Linux (/dev/ttyS0)")
                print()
                print("シリアルポート2:")
                print("  パイプ名: \\\\.\\\pipe\\\\linux_to_win")
                print("  用途: Linux→Windows (/dev/ttyS1)")
                print()
                print("【Linux側権限設定】")
                print("sudo usermod -a -G dialout $USER && logout")
                print("または")
                print("sudo chmod 666 /dev/ttyS*")
            
            elif choice == "8":
                print("👋 終了します")
                break
            
            else:
                print("❌ 1-8を選択してください")
                
    except KeyboardInterrupt:
        print("\n👋 終了します")
    finally:
        comm.stop_all()

if __name__ == "__main__":
    main()

🎯 VMware双方向通信の完全解決法

問題の根本原因

名前付きパイプは本質的に一方向通信のため、単一パイプでは双方向通信ができません。

📋 解決策：デュアルパイプ設定

VMware設定（2つのシリアルポート必要）

シリアルポート1（Windows→Linux）

接続方法: 名前付きパイプを使用
パイプ名: \\.\pipe\win_to_linux
パイプの端: サーバー
I/Oモード: アプリケーション

シリアルポート2（Linux→Windows）

接続方法: 名前付きパイプを使用
パイプ名: \\.\pipe\linux_to_win
パイプの端: サーバー
I/Oモード: アプリケーション

🔄 通信フローの仕組み

Windows側アプリケーション
    ↓ 送信          ↑ 受信
\\.\pipe\win_to_linux  \\.\pipe\linux_to_win
    ↓ VMware        ↑ VMware
/dev/ttyS0          /dev/ttyS1
    ↓ 受信          ↑ 送信
Linux側アプリケーション

📊 実際の通信例

Windows側ログ

[TX 001] Windows → Linux: Windows_Message_001
[RX 14:30:15.123] Linux → Windows: LINUX_TO_WIN,001,45.23,CPU_TEMP

Linux側ログ

[RX 14:30:15.100] Windows → Linux: {"source":"Windows_Host","data":"Windows_Message_001"}
[TX 001] Linux → Windows: LINUX_TO_WIN,001,45.23,CPU_TEMP

🛠️ 設定手順

1. VMware設定

1. VM設定を開く
2. 「ハードウェア追加」→「シリアルポート」を2回実行
3. 各ポートを上記設定に従って構成

2. Windows側実行

python dual_pipe_windows.py
# メニュー2: 双方向通信開始

3. Linux側実行

python3 dual_pipe_linux.py
# メニュー4: 双方向通信テスト

簡化版VMware双方向通信設定ガイド

🎯 新発見による真の理解

重要な発見: VMwareで名前付きパイプを一度設定してVM再起動すると、Windows側に既存のCOM3/COM4をVMwareが認識・表示するようになり、物理シリアルポートとして利用可能になる。

🔍 VMware動作メカニズムの解明

実際に起きていること

1. Windows環境: COM1-COM4が元々存在（デバイスマネージャー確認済み）
2. VMware初期状態: COM1, COM2のみ選択肢に表示
3. 名前付きパイプ設定: VMwareの内部ポート検索を更新
4. VM再起動後: COM3, COM4が選択肢に追加表示
5. 物理ポート利用: COM3, COM4を直接利用可能

名前付きパイプの役割

• 新しいポート作成ではない
• 既存ポートの認識更新: VMwareの内部ポートスキャンをトリガー
• 設定後は不要: 物理ポート設定に変更可能

🔧 設定手順（実証済み手順）

ステップ1: VMware初期設定

名前付きパイプ設定（一時的）

VM設定 → ハードウェア追加 → シリアルポート

✅ 接続方法: 名前付きパイプを使用
✅ パイプ名: \\.\pipe\host_to_guest
✅ パイプの端: サーバー  
✅ I/Oモード: アプリケーション
✅ 仮想マシンの電源投入時に接続: チェック

ステップ2: VMware再起動による認識更新

重要な手順

1. 名前付きパイプ設定後、VM起動
2. VM正常シャットダウン  
3. VMware設定を再度確認
4. 🎯 発見: 物理シリアルポート選択肢にCOM3, COM4が追加表示される

ステップ3: 物理ポート設定への変更

実際の設定変更

VM設定 → シリアルポート設定変更

✅ 接続方法: 物理シリアルポートを使用する(U)
✅ ポート選択: COM3 または COM4 を選択
   （名前付きパイプ設定により表示されるようになったポート）

ステップ4: 確認作業

Windows側での状況確認

■ 設定前の状態:
  - Windowsデバイスマネージャー: COM1-COM4表示
  - VMware選択肢: COM1, COM2のみ表示

■ 名前付きパイプ設定→VM再起動後:
  - Windowsデバイスマネージャー: 変化なし（COM1-COM4）
  - VMware選択肢: COM1, COM2, COM3, COM4すべて表示

■ 結論:
  - Windows側に元々存在していたCOM3, COM4
  - VMwareが認識・表示するようになった
  - 名前付きパイプ設定がトリガーとなった

💻 実装例（物理ポート使用）

Windows側送受信

#!/usr/bin/env python3
"""
Windows側 - 物理COMポート利用
"""
import serial
import threading
import time
import json
from datetime import datetime

def windows_communication():
    # VMware設定により利用可能になったポート
    port = "COM3"  # またはCOM4（VM設定に応じて）
    
    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: 送信のみ, 2: 受信のみ, 3: 双方向: ")
        
        if choice == "1":
            send_data()
        elif choice == "2":
            receive_data()
        elif choice == "3":
            # 受信スレッド開始
            rx_thread = threading.Thread(target=receive_data)
            rx_thread.daemon = True
            rx_thread.start()
            # 送信開始
            send_data()
    except KeyboardInterrupt:
        print("通信終了")

if __name__ == "__main__":
    windows_communication()

Linux側送受信

#!/usr/bin/env python3
"""
Linux側 - VMware仮想シリアルポート利用
"""
import serial
import threading
import time
import json
from datetime import datetime

def linux_communication():
    port = "/dev/ttyS0"  # 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: 送信のみ, 2: 受信のみ, 3: 双方向: ")
        
        if choice == "1":
            send_data()
        elif choice == "2":
            receive_data()
        elif choice == "3":
            # 受信スレッド開始
            rx_thread = threading.Thread(target=receive_data)
            rx_thread.daemon = True
            rx_thread.start()
            # 送信開始
            send_data()
    except KeyboardInterrupt:
        print("通信終了")

if __name__ == "__main__":
    linux_communication()

🎯 ポート対応（実証済み）

⚡ 利点

1. 設定の単純化

• 1つの名前付きパイプ設定で済む
• 複雑なWin32 API不要
• 標準的なpyserialで完結

2. 安定性

• VMware標準機能活用
• OS標準のシリアルポートとして認識
• エラーハンドリングが簡単

3. 開発効率

• 従来のシリアル通信知識で対応可能
• デバッグツール（Teratermなど）も使用可能
• クロスプラットフォーム対応

🔍 トラブルシューティング

よくある問題

1. COMポートが認識されない

→ VM再起動後に確認
→ VMware Toolsが最新か確認
→ デバイスマネージャーで更新

2. Linux側デバイス未検出

# VMwareシリアルポート確認
sudo dmesg | grep ttyS

# 手動権限設定
sudo chmod 666 /dev/ttyS*

3. 通信が片方向のみ

→ 2つ目のシリアルポートを追加
→ COM3/COM4両方が認識されているか確認
→ ポート番号の割り当て確認

⚡ 代替解決策（より簡単）

VMware設定が複雑な場合の代替案：

TCP/IPハイブリッド方式

# Windows→Linux: 名前付きパイプ（1つだけ）
# Linux→Windows: TCP/IP経由

# Windows側
python network_serial_bridge.py  # 受信サーバー起動

# Linux側  
python enhanced_linux_sender.py  # TCP送信

🔍 トラブルシューティング

よくある問題と解決法

1. パイプ接続失敗

❌ パイプ接続エラー: システムが指定されたファイルが見つけられません
→ VM起動順序を確認（VM起動後にWindowsツール実行）
→ パイプ名の綴りを再確認

2. Linux側デバイス未検出

❌ /dev/ttyS1 が存在しません
→ VMwareで2つ目のシリアルポートが追加されているか確認
→ VM再起動後に再度確認

3. 権限エラー

❌ /dev/ttyS0: 権限がありません
→ sudo chmod 666 /dev/ttyS*
→ sudo usermod -a -G dialout $USER && logout

💡 実用的な活用例

開発・テスト用途

# Windows側でダッシュボード表示
# Linux側からセンサーデータ送信
# リアルタイム双方向データ交換

IoTプロトタイピング

# Windows: 制御コマンド送信
# Linux: ステータス・センサー値返送
# ハードウェア連携シミュレーション

🥈 推奨方法2: TCP/IPブリッジ（代替推奨）

VMware設定が困難な場合の代替案：

Windows側サーバー

# network_serial_bridgeを使用
python network_serial_bridge.py
# メニュー1: シリアル→ネットワーク

#!/usr/bin/env python3
"""
ネットワーク経由でシリアルデータを送信するブリッジ
Windowsから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):
        """シリアルポートからネットワークへデータ転送（サーバー側）"""
        print(f"シリアル→ネットワーク サーバー開始")
        print(f"シリアルポート: {serial_port} ({baudrate} baud)")
        print(f"ネットワークポート: {network_port}")
        print("クライアントの接続を待機中...")
        
        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"クライアント接続: {addr}")
                    
                    try:
                        while True:
                            if ser.in_waiting > 0:
                                data = ser.read(ser.in_waiting)
                                client_socket.send(data)
                                print(f"転送: {len(data)} bytes → {addr}")
                            
                            time.sleep(0.01)
                            
                    except (ConnectionResetError, BrokenPipeError):
                        print(f"クライアント切断: {addr}")
                        client_socket.close()
                        
        except KeyboardInterrupt:
            print("サーバー停止")
        finally:
            server_socket.close()
    
    def network_to_serial_client(self, target_host, network_port, serial_port, baudrate):
        """ネットワークからシリアルポートへデータ転送（クライアント側）"""
        print(f"ネットワーク→シリアル クライアント開始")
        print(f"接続先: {target_host}:{network_port}")
        print(f"シリアルポート: {serial_port} ({baudrate} baud)")
        
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.connect((target_host, network_port))
                print("サーバーに接続しました")
                
                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"受信: {len(data)} bytes → {serial_port}")
                        
        except KeyboardInterrupt:
            print("クライアント停止")
        except Exception as e:
            print(f"エラー: {e}")
    
    def send_test_data_to_network(self, target_host, network_port):
        """ネットワーク経由でテストデータ送信"""
        print(f"テストデータ送信開始 → {target_host}:{network_port}")
        
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.connect((target_host, network_port))
                print("接続成功")
                
                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}] 送信: {message.strip()}")
                    counter += 1
                    
                    time.sleep(2)
                    
        except KeyboardInterrupt:
            print("送信停止")
        except Exception as e:
            print(f"エラー: {e}")
    
    def receive_network_data(self, network_port=12345):
        """ネットワークデータ受信サーバー"""
        print(f"データ受信サーバー開始 (port: {network_port})")
        print("接続待機中...")
        
        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"クライアント接続: {addr}")
                
                try:
                    while True:
                        data = client_socket.recv(1024)
                        if not data:
                            break
                        
                        timestamp = datetime.now().strftime('%H:%M:%S.%f')[:-3]
                        decoded_data = data.decode('utf-8', errors='ignore').strip()
                        print(f"[{timestamp}] 受信: {decoded_data}")
                        
                except ConnectionResetError:
                    print(f"クライアント切断: {addr}")
                finally:
                    client_socket.close()
                    
        except KeyboardInterrupt:
            print("受信サーバー停止")
        finally:
            server_socket.close()

def main():
    bridge = NetworkSerialBridge()
    
    while True:
        print("\n=== ネットワーク経由シリアル通信ブリッジ ===")
        print("1. シリアル→ネットワーク (サーバー)")
        print("2. ネットワーク→シリアル (クライアント)")
        print("3. テストデータ送信 (ネットワーク)")
        print("4. データ受信サーバー")
        print("5. 使用方法ガイド")
        print("6. 終了")
        
        choice = input("\n選択してください (1-6): ").strip()
        
        if choice == "1":
            # シリアルポート選択
            import serial.tools.list_ports
            ports = serial.tools.list_ports.comports()
            
            if not ports:
                print("シリアルポートが見つかりません")
                continue
            
            print("\n利用可能なポート:")
            for i, port in enumerate(ports, 1):
                print(f"{i}. {port.device}")
            
            try:
                port_choice = int(input("ポート選択: ")) - 1
                if 0 <= port_choice < len(ports):
                    serial_port = ports[port_choice].device
                    baudrate = int(input("ボーレート (9600): ") or "9600")
                    network_port = int(input("ネットワークポート (12345): ") or "12345")
                    
                    bridge.serial_to_network_server(serial_port, baudrate, network_port)
                
            except (ValueError, IndexError):
                print("無効な選択です")
        
        elif choice == "2":
            target_host = input("接続先IPアドレス: ").strip()
            network_port = int(input("ネットワークポート (12345): ") or "12345")
            
            # シリアルポート選択
            import serial.tools.list_ports
            ports = serial.tools.list_ports.comports()
            
            if not ports:
                print("シリアルポートが見つかりません")
                continue
            
            print("\n利用可能なポート:")
            for i, port in enumerate(ports, 1):
                print(f"{i}. {port.device}")
            
            try:
                port_choice = int(input("ポート選択: ")) - 1
                if 0 <= port_choice < len(ports):
                    serial_port = ports[port_choice].device
                    baudrate = int(input("ボーレート (9600): ") or "9600")
                    
                    bridge.network_to_serial_client(target_host, network_port, serial_port, baudrate)
                
            except (ValueError, IndexError):
                print("無効な選択です")
        
        elif choice == "3":
            target_host = input("送信先IPアドレス: ").strip()
            network_port = int(input("ネットワークポート (12345): ") or "12345")
            
            bridge.send_test_data_to_network(target_host, network_port)
        
        elif choice == "4":
            network_port = int(input("受信ポート (12345): ") or "12345")
            bridge.receive_network_data(network_port)
        
        elif choice == "5":
            print("\n=== 使用方法ガイド ===")
            print("\n【WindowsからLinuxへのデータ送信】")
            print("1. Linux側で受信準備:")
            print("   nc -l -p 12345 | tee received_data.txt")
            print("   または")
            print("   python3 thisscript.py → メニュー4選択")
            print()
            print("2. Windows側からデータ送信:")
            print("   python thisscript.py → メニュー3選択")
            print("   LinuxのIPアドレスを入力")
            print()
            print("【シリアルポート間でのネットワーク経由転送】")
            print("1. サーバー側 (データ送信元):")
            print("   メニュー1: シリアル→ネットワーク")
            print("2. クライアント側 (データ受信先):")
            print("   メニュー2: ネットワーク→シリアル")
            print()
            print("【IPアドレス確認方法】")
            print("- Windows: ipconfig")
            print("- Linux: ip addr show または ifconfig")
            print("- 同一ネットワーク内で使用してください")
        
        elif choice == "6":
            print("終了します")
            break
        
        else:
            print("1-6を選択してください")

if __name__ == "__main__":
    main()

Linux側クライアント

# VMware仮想ネットワーク経由で接続
python3 network_serial_bridge.py  
# メニュー2: ネットワーク→シリアル

⚠️ 従来方法の問題点

ポート占有方法の課題

# 元記事の方法 - 不安定要因が多い
target_ports = ['COM1', 'COM3']  # 占有対象
# 問題:
# - VMwareの自動検出は環境依存
# - ポート競合リスクが高い
# - Windows Updateで設定が変わる可能性

🎮 実用的な双方向通信設定

Windows側ダッシュボード（推奨設定）

📥 RX設定:
  - 方法1: パイプ \\.\pipe\guest_to_host
  - 方法2: TCP Socket (port 9999)

📤 TX設定:  
  - 方法1: パイプ \\.\pipe\host_to_guest
  - 方法2: TCP Socket (port 9998)

✅ 送受信分離: チェック

Linux側ダッシュボード（推奨設定）

📥 RX設定:
  - 方法1: /dev/ttyS1
  - 方法2: TCP Client (Windows_IP:9998)

📤 TX設定:
  - 方法1: /dev/ttyS0  
  - 方法2: TCP Client (Windows_IP:9999)

✅ 送受信分離: チェック

💡 環境別の推奨度

🔧 トラブルシューティング

よくある問題と解決法

1. パイプ接続失敗

# パイプ存在確認
dir \\.\pipe\ | findstr vmware

# VMware Tools再インストール
# VM設定でシリアルポート削除→再追加

2. Linux側でデバイス未検出

# VMware Tools確認
vmware-toolbox-cmd --version

# カーネルログ確認  
sudo dmesg | grep -E "(tty|serial)"

# 手動デバイス作成（最終手段）
sudo mknod /dev/ttyS0 c 4 64

3. 権限エラー

# 一時的解決
sudo chmod 666 /dev/ttyS*

# 永続的解決
sudo usermod -a -G dialout $USER
# 再ログイン必要

🎉 本当のまとめ

この発見（名前付きパイプ設定を一時的に作成）により、VMware環境での双方向シリアル通信が大幅に簡素化されました。

従来: 複雑なデュアルパイプ + Win32 API 新方式: 単一名前付きパイプ + 標準pyserial

これで開発・学習・テスト環境での活用が格段に容易になります！

実用性ランキング

1. 🏆 名前付きパイプ – VMware環境での最適解
2. 🥈 TCP/IPブリッジ – 汎用性が高い代替案
3. 🥉 物理シリアル – 実機環境でのみ有効
4. 🔧 ポート占有 – 技巧的だが不安定

開発用途別推奨

• 学習・プロトタイピング: TCP/IPブリッジ
• 本格開発・テスト: 名前付きパイプ
• プロダクション: 実機 + 物理シリアル
• IoT開発: VirtualBox または 実機

この改訂版により、より実用的で安定した双方向通信が実現できます。特に名前付きパイプを使用することで、VMwareの制限を最も効果的に回避できるでしょう。