目的

特定のフォーマットでテスト用コマンドを生成し、末尾にチェックサムを付加するシリアル通信ツールの開発。

開発プロセス

初期段階ではStreamlitで一括処理を試みたが、ポートのオープン・クローズが関数内で完結するため、設定変更や状態管理が困難だった。そこで、UI部分とシリアル処理部分を分離しようとしたが、ソケット通信を導入する必要があり、構成が複雑化した。最終的に、Streamlit内でのマルチスレッド方式を採用。メインスレッドとサブスレッド間ではキューを使用してデータ交換を行い、基本的な機能は達成できた。

ただし、シリアルポートを開いた際に一時的な遅延が発生し、コマンドラインからの強制終了（Ctrl+C）が効かないという問題がある。受信処理は未実装のままになっているが、以下は動作確認可能なコード例として参考までに掲載。

# -*- coding: utf-8 -*-
import time
import queue
import threading
import streamlit as st
import serial

st.set_page_config(
    page_title="Ex-stream-ly Cool App",
    page_icon="🧊",
    layout="wide",
    initial_sidebar_state="expanded",
    menu_items={
        'Get Help': 'https://www.extremelycoolapp.com/help',
        'Report a bug': "https://www.extremelycoolapp.com/bug",
        'About': "# This is a header. This is an *extremely* cool app!"
    }
)

# メインスレッド → サブスレッドへのデータキュー
if 'maintosub_queue' not in st.session_state:
    st.session_state.maintosub_queue = queue.Queue()
# サブスレッド → メインスレッドへのデータキュー
if 'subtomain_queue' not in st.session_state:
    st.session_state.subtomain_queue = queue.Queue()
# サブスレッド停止イベント
if "stop_event" not in st.session_state:
    st.session_state.stop_event = threading.Event()
# スレッド実行状態
if "thread_running" not in st.session_state:
    st.session_state.thread_running = False
# 最新入力値保持
if "last_input_data" not in st.session_state:
    st.session_state.last_input_data = 5
# 再レンダリングフラグ
if "need_rerun" not in st.session_state:
    st.session_state.need_rerun = False

# サブスレッドタスク：ポート設定を受け取り、データ送受信を処理
def background_task(stop_event, maintosub_queue, subtomain_queue):
    try:
        config = maintosub_queue.get(timeout=1)
        ser = serial.Serial()
        ser.port = config[0]
        ser.baudrate = int(config[1])
        ser.bytesize = serial.EIGHTBITS
        ser.parity = config[2] if config[2] != 'N' else serial.PARITY_NONE
        ser.stopbits = serial.STOPBITS_ONE
        ser.timeout = 1

        ser.open()
        while not stop_event.is_set():
            time.sleep(0.1)
            # 新しい送信データのチェック
            if not maintosub_queue.empty():
                data = maintosub_queue.get()
                ser.write(data)
                print(f"Sent: {data}")

            # 受信データの読み取り
            if ser.in_waiting > 0:
                raw_data = ser.read_all()
                subtomain_queue.put(str(raw_data))

    except Exception as e:
        print(f"エラー: {e}")
    finally:
        if ser.is_open:
            ser.close()

# UI部品
st.title("スレッド間データ通信のサンプル")

# 初期化
st.session_state.need_rerun = True

col1, col2, col3 = st.columns(3)
with col1:
    port = st.selectbox("ポート", ("COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9"))
with col2:
    baudrate = st.selectbox("ボーレート", ("115200", "38400", "921600"))
with col3:
    parity = st.radio("パリティ", ("N", "O", "E"), index=1)

col4, col5, col6 = st.columns(3)
with col4:
    if st.button("ポートを開く", disabled=st.session_state.thread_running):
        st.session_state.thread_running = True
        st.session_state.stop_event.clear()
        st.session_state.maintosub_queue.queue.clear()
        st.session_state.maintosub_queue.put((port, baudrate, parity))
        st.session_state.subtomain_queue.queue.clear()
        st.session_state.last_input_data = 5
        if "result" in st.session_state:
            del st.session_state.result
        thread = threading.Thread(target=background_task,
                                  args=(st.session_state.stop_event,
                                        st.session_state.maintosub_queue,
                                        st.session_state.subtomain_queue))
        thread.start()

with col5:
    if st.button("ポートを閉じる", disabled=not st.session_state.thread_running):
        st.session_state.stop_event.set()

with col6:
    st.image("green.ico" if st.session_state.thread_running else "red.ico")

input_val = st.number_input("値入力", value=5, min_value=5, max_value=100, step=1, key='input_data')

mode = st.radio("モード選択", ("昼連続", "昼閃光", "夜連続", "夜閃光"), index=1)
mode_map = {"昼連続": 0, "昼閃光": 1, "夜連続": 2, "夜閃光": 3}
mode_code = mode_map[mode]

checksum = (0x55 + 0xAA + mode_code + input_val + 0x11) & 0xFF
command = bytes([0x55, 0xAA, mode_code, input_val, 0x11, checksum])

if st.session_state.thread_running and input_val != st.session_state.last_input_data:
    st.session_state.maintosub_queue.put(command)
    st.session_state.last_input_data = input_val

def send_shutdown():
    st.session_state.maintosub_queue.put(bytes([0x55, 0xAA, 0x00, 0x05, 0x00, 0x04]))
    print(bytes([0x55, 0xAA, 0x00, 0x05, 0x00, 0x04]))

st.button("出光停止", on_click=send_shutdown)

# サブスレッドからのデータ受信処理
while st.session_state.thread_running:
    if not st.session_state.subtomain_queue.empty():
        received = st.session_state.subtomain_queue.get()
        st.write(received)
        print(received)
    time.sleep(0.1)

if st.session_state.need_rerun:
    st.rerun()

マルチプロセス通信の試み

AI生成コードを用いてマルチプロセス間通信を実装したが、正しくデータが伝わらなかった。原因は不明だが、以下のような構成であった。

# common_queue.py
from multiprocessing import Manager
manager = Manager()
comm_queue = manager.Queue(maxsize=0)

# process1.py
from common_queue import comm_queue
import time
import random

if __name__ == "__main__":
    print("プロセス1：開始")
    count = 0
    while True:
        count += 1
        data = {"type": "send", "content": f"test{count}", "timestamp": time.time()}
        comm_queue.put(data)
        print(f"→ {data}")
        time.sleep(random.uniform(0.5, 1.5))

# process2.py
from common_queue import comm_queue
import time

if __name__ == "__main__":
    print("プロセス2：開始")
    while True:
        if not comm_queue.empty():
            data = comm_queue.get()
            print(f"← {data}")
        time.sleep(0.2)

ソケット通信の実装例

TCPソケット通信も試作。サーバーとクライアントで共通のデータ形式としてJSONを使用できる。主な関数は以下の通り：

• json.dumps()：PythonオブジェクトをJSON文字列に変換
• json.loads()：JSON文字列をPythonオブジェクトに復元

クライアント側

import socket
import threading

class TCPClient:
    def __init__(self, host='127.0.0.1', port=8888):
        self.host = host
        self.port = port
        self.client_socket = None
        self.running = False
        self.receive_thread = None

    def connect(self):
        try:
            self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.client_socket.connect((self.host, self.port))
            print(f"✅ 接続成功: {self.host}:{self.port}")
            self.running = True
            self.receive_thread = threading.Thread(target=self.receive_loop)
            self.receive_thread.daemon = True
            self.receive_thread.start()
        except Exception as e:
            print(f"❌ 接続失敗: {e}")

    def send_message(self, msg):
        if self.running and self.client_socket:
            try:
                self.client_socket.send(msg.encode('utf-8'))
            except Exception as e:
                print(f"❌ 送信エラー: {e}")

    def receive_loop(self):
        try:
            while self.running:
                data = self.client_socket.recv(1024)
                if not data:
                    break
                print(f"📥 受信: {data.decode('utf-8')}")
        except Exception as e:
            print(f"❌ 受信エラー: {e}")

    def disconnect(self):
        self.running = False
        if self.client_socket:
            self.client_socket.close()
            print("🔌 コネクション切断")

サーバー側

import socket
import threading

class TCPServer:
    def __init__(self, host='127.0.0.1', port=8888):
        self.host = host
        self.port = port
        self.server_socket = None
        self.running = False

    def start(self):
        try:
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.server_socket.bind((self.host, self.port))
            self.server_socket.listen(5)
            print(f"✅ サーバー起動: {self.host}:{self.port}")
            self.running = True

            while self.running:
                client_socket, addr = self.server_socket.accept()
                print(f"🔗 接続: {addr}")
                thread = threading.Thread(target=self.handle_client, args=(client_socket, addr))
                thread.daemon = True
                thread.start()

        except Exception as e:
            print(f"❌ 起動エラー: {e}")
        finally:
            self.stop()

    def handle_client(self, client_socket, addr):
        try:
            while True:
                data = client_socket.recv(1024)
                if not data:
                    break
                message = data.decode('utf-8').strip()
                print(f"📥 {addr} から: {message}")

                response = f"echo: {message}\n"
                client_socket.send(response.encode('utf-8'))

        except Exception as e:
            print(f"❌ 処理エラー: {e}")
        finally:
            client_socket.close()

    def stop(self):
        self.running = False
        if self.server_socket:
            self.server_socket.close()
            print("🛑 サーバー停止")