From c25a379fbb347450e526d7f74e09bfbbb7b09574 Mon Sep 17 00:00:00 2001
From: yabu76 <yabu76@gmail.com>
Date: Sun, 7 Dec 2025 11:21:55 +0900
Subject: [PATCH] Add some socket examples of bidirectional command helpers and
 pexpect

bidir_cmd_herlper.sh: Bidirectional command helper by socat.
bidir_cmd_helper.py: Bidirectional command helper by python and netcat.

pexpect-ping.py, pexpect-pong.py: Scripts for throwing Ping-Pong between
cross-connected tios.
---
 examples/socket/bidir_cmd_helper.py | 94 +++++++++++++++++++++++++++++
 examples/socket/bidir_cmd_helper.sh | 15 +++++
 examples/socket/pexpect-ping.py     | 22 +++++++
 examples/socket/pexpect-pong.py     | 22 +++++++
 4 files changed, 153 insertions(+)
 create mode 100755 examples/socket/bidir_cmd_helper.py
 create mode 100755 examples/socket/bidir_cmd_helper.sh
 create mode 100755 examples/socket/pexpect-ping.py
 create mode 100755 examples/socket/pexpect-pong.py

diff --git a/examples/socket/bidir_cmd_helper.py b/examples/socket/bidir_cmd_helper.py
new file mode 100755
index 0000000..6344058
--- /dev/null
+++ b/examples/socket/bidir_cmd_helper.py
@@ -0,0 +1,94 @@
+#!/usr/bin/python3
+
+import sys
+import subprocess
+import shlex
+import os
+
+# netcat (openbsd)
+NC_CMD = "nc"
+NC_OPT = "-UN"
+
+def connect_and_execute_bidirectional_command(socket_path, command):
+    TIMEOUT_SEC = 600
+
+    # Sanitize user input (command stays as a string, but avoid passing raw input directly)
+    safe_cmd = " ".join(shlex.split(command))
+
+    # Launch nc process (connect to the Unix domain socket)
+    p_nc = subprocess.Popen(
+        [NC_CMD, NC_OPT, socket_path],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        shell=False,
+        text=False,  # binary transfer
+    )
+
+    # Launch the bidirectional command connected to nc
+    p_bidir_cmd = subprocess.Popen(
+        safe_cmd,                # keep string command for shell=True
+        stdin=p_nc.stdout,
+        stdout=p_nc.stdin,
+        stderr=None,
+        shell=True,
+        text=False,
+    )
+
+    # Close unused pipe ends in the parent process to propagate SIGPIPE properly
+    p_nc.stdout.close()
+    p_nc.stdin.close()
+
+    try:
+        p_bidir_cmd.communicate(timeout=TIMEOUT_SEC)
+
+    except subprocess.TimeoutExpired:
+        # Ensure the command is killed on timeout
+        p_bidir_cmd.kill()
+
+    except Exception as e:
+        # Print unexpected exceptions for debugging
+        print(type(e), file=sys.stderr)
+
+    # Check bidirectional command exit code
+    if p_bidir_cmd.returncode != 0:
+        print(f"command exited with {p_bidir_cmd.returncode}", file=sys.stderr)
+
+    # Ensure nc is terminated regardless of exceptions
+    try:
+        p_nc.terminate()
+        p_nc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        # Force kill if graceful shutdown fails
+        p_nc.kill()
+    except Exception as e:
+        print(type(e), file=sys.stderr)
+
+    # Read and report stderr of nc
+    nc_stderr = p_nc.stderr.read()
+    if nc_stderr:
+        print("nc stderr:", nc_stderr.decode(errors='ignore'), file=sys.stderr)
+
+    return p_bidir_cmd.returncode
+
+
+def main():
+    script = os.path.basename(__file__)
+    usage = f"Usage: {script} <socket-path> <bidirectional-command>"
+    example = f"Example: {script} /tmp/tio-socket0 \"sz -b -p sample.bin\""
+    note = f"Note: Please run \"tio -S <socket-path> <your-serial-device>\" beforehand."
+
+    if len(sys.argv) != 3:
+        print(usage, file=sys.stderr)
+        print(example, file=sys.stderr)
+        print(note, file=sys.stderr)
+        return 1
+
+    socket_path = sys.argv[1]
+    bidirectional_command = sys.argv[2]
+    return connect_and_execute_bidirectional_command(socket_path, bidirectional_command)
+
+
+if __name__ == '__main__':
+    exit_code = main()
+    sys.exit(exit_code)
diff --git a/examples/socket/bidir_cmd_helper.sh b/examples/socket/bidir_cmd_helper.sh
new file mode 100755
index 0000000..5459721
--- /dev/null
+++ b/examples/socket/bidir_cmd_helper.sh
@@ -0,0 +1,15 @@
+#!/bin/sh
+
+if [ $# -lt 2 ]; then
+    echo "Usage: $0 <socket-path> <bidirectional-command>"
+    echo "Example: $0 /tmp/tio-socket0 \"sz -b -p sample.bin\""
+    echo "Note: Please run \"tio -S <socket-path> <your-serial-device>\" beforehand."
+    exit 1
+fi
+
+SOCKET_PATH=$1
+BIDIR_CMD=$2
+
+socat EXEC:"$BIDIR_CMD" $SOCKET_PATH
+
+exit $?
diff --git a/examples/socket/pexpect-ping.py b/examples/socket/pexpect-ping.py
new file mode 100755
index 0000000..4323849
--- /dev/null
+++ b/examples/socket/pexpect-ping.py
@@ -0,0 +1,22 @@
+#!/usr/bin/python3
+# In MSYS2, use /mingw64/bin/python3
+#
+# Send a "Ping" and wait for a "Pong".
+# Repeat this process.
+#
+
+import pexpect
+from pexpect import popen_spawn
+
+child = pexpect.popen_spawn.PopenSpawn("nc -UN /tmp/tio-socket0")
+
+cnt = 0
+while True:
+    try:
+        child.sendline(f"Ping {cnt:d}")
+        cnt += 1
+        child.expect(r'Pong \d+[\r\n]+', timeout = 10)
+    except Exception as e:
+        print(type(e))
+        break
+
diff --git a/examples/socket/pexpect-pong.py b/examples/socket/pexpect-pong.py
new file mode 100755
index 0000000..8f93c96
--- /dev/null
+++ b/examples/socket/pexpect-pong.py
@@ -0,0 +1,22 @@
+#!/usr/bin/python3
+# In MSYS2, use /mingw64/bin/python3
+#
+# wait for a "Ping" and Send a "Pong"
+# Repeat this process.
+#
+
+import pexpect
+from pexpect import popen_spawn
+
+child = pexpect.popen_spawn.PopenSpawn("nc -UN /tmp/tio-socket1")
+
+cnt = 0
+while True:
+    try:
+        child.expect(r'Ping \d+[\r\n]+', timeout = 10)
+        child.sendline(f"Pong {cnt:d}")
+        cnt += 1
+    except Exception as e:
+        print(type(e))
+        break
+