【PythonでつくるTUI(Text User Interface)】シリアル通信をリアルタイムで表示するアプリの作り方(CURSES、JSON)

ATOM Matrix/Lite
けんろう
けんろう

Python、Cursesを使用すれば、シリアル通信データをリアルタイムに表示するアプリを簡単に作れます。社内の評価用など、ちょっと動かしてみたいときに便利です。

この記事に載っているサンプルコードをコピーして、Pythonで実行すれば、簡単に動きを確認できますので、是非試してみて下さい。

この記事では、初心者にもわかりやすいように、各処理の内容を、サンプルコード内にコメントとして載せています。

スポンサーリンク

今回作成するアプリ

今回紹介するアプリは、

・センサ(M5ATOM)から送られてくるシリアル通信データを、PCのコンソール上に、リアルタイムで表示する
・なにかキーを押下すると、アプリが終了

シリアル通信に流すJSONデータ

{
 ”TIME”:[XXXX],
 ”ID”:[X],
 ”DLC”:[X],
 ”DATA”:[XXX,XXX,XXX,XXX,XXX,XXX,XXX,XXX]
}

{“TIME”:[3155],”ID”:[1],”DLC”:[8],”DATA”:[233,212,38,131,76,6,197,234]}
{“TIME”:[4155],”ID”:[2],”DLC”:[4],”DATA”:[158,47,242,168]}
{“TIME”:[5155],”ID”:[3],”DLC”:[6],”DATA”:[150,238,44,71,10,116]}

環境準備

システム構成

Windows PCに、M5ATOMをUSBケーブルで繋ぎます。

Windows PC側の準備

本アプリを動かすために、以下のライブラリが必要です。

curses:テキストユーザインターフェースのためのライブラリ
pySerial:シリアル通信を扱うためのライブラリ

curses

公式サイトは、以下です。

インストール方法は、以下です。

pip install windows-curses

pySerial

インストール方法は、以下です。

pip install pyserial 

M5ATOM側のサンプルコード

M5ATOMでは、PCへ送るためのJSONテストデータを生成します。

#include <M5Atom.h>

#include "M5Atom.h"
#include "time.h"

#include <ArduinoJson.h>

void setup() {

  M5.begin(true, false, true);
  delay(50);

  // シリアル通信機能の設定
  Serial.begin(115200);

  Serial2.begin(4800, SERIAL_8O1, 26, 32);
  Serial2.setTimeout(50);

  // 起動時LED点灯
  M5.dis.drawpix(0,0xBBFFFF);  // 紫
}

void loop() {

  M5.update();      // update button state

  // フレーム1

  StaticJsonDocument<500> doc;
   
  // JSONメッセージの作成
  JsonArray timeValues = doc.createNestedArray("TIME");
  JsonArray idValues = doc.createNestedArray("ID");
  JsonArray dlcValues = doc.createNestedArray("DLC");
  JsonArray dataValues = doc.createNestedArray("DATA");

  // フレーム1の作成
  timeValues.add(millis());
  idValues.add(0x01);
  dlcValues.add(0x08);
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  dataValues.add(random(0,255));
  
  serializeJson(doc, Serial);
  Serial.println("");

  delay(1000);

  // フレーム2
  
  StaticJsonDocument<500> doc2;
   
  // JSONメッセージの作成
  JsonArray timeValues2 = doc2.createNestedArray("TIME");
  JsonArray idValues2 = doc2.createNestedArray("ID");
  JsonArray dlcValues2 = doc2.createNestedArray("DLC");
  JsonArray dataValues2 = doc2.createNestedArray("DATA");

  // フレーム2の作成
  timeValues2.add(millis());
  idValues2.add(0x02);
  dlcValues2.add(0x04);
  dataValues2.add(random(0,255));
  dataValues2.add(random(0,255));
  dataValues2.add(random(0,255));
  dataValues2.add(random(0,255));
  
  serializeJson(doc2, Serial);
  Serial.println("");

  delay(1000);

  // フレーム3
  StaticJsonDocument<500> doc3;
   
  // JSONメッセージの作成
  JsonArray timeValues3 = doc3.createNestedArray("TIME");
  JsonArray idValues3 = doc3.createNestedArray("ID");
  JsonArray dlcValues3 = doc3.createNestedArray("DLC");
  JsonArray dataValues3 = doc3.createNestedArray("DATA");

  // フレーム3の作成
  timeValues3.add(millis());
  idValues3.add(0x03);
  dlcValues3.add(0x06);
  dataValues3.add(random(0,255));
  dataValues3.add(random(0,255));
  dataValues3.add(random(0,255));
  dataValues3.add(random(0,255));
  dataValues3.add(random(0,255));
  dataValues3.add(random(0,255));
  
  serializeJson(doc3, Serial);
  Serial.println("");

  delay(1000);

}

サンプルコード(PC側)

#!usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import curses
import serial
import json
import time
import msvcrt

# 起動時に、コンソールの画面サイズ設定とCOMポート番号を入力される
print("コンソールのサイズ設定を、幅:170、高さ:45に設定してください。")
print("タイトルバーで右クリックで、プロパティを選択して、ウィンドウのサイズを変更。")
var2 = input("設定できたら、なにかキー+「ENTER」を押してください : ")
var = input("シリアル通信に使用するCOMポート番号を入力してください(例:COM3) : ")
print(" {} が入力されました。".format(var))


# 個別テーブルデータの描画
# (描画するテーブルデータ、描画位置(X軸)、描画位置(Y軸)、コンソール、見出し、テーブルデータ数)
def func_drawblock(table,y_data,x_data,stdscr,title,switchdata):

    # 見出しの表示設定
    stdscr.attron(curses.color_pair(2))     # 文字色をカラー2に切り替えON
    stdscr.addstr(y_data, x_data, title)    # タイトルを描画
    stdscr.attroff(curses.color_pair(2))     # 文字色をカラー2に切り替えOFF

    # テーブルを描画
    i = 1
    for name in table:

        if switchdata == 2:
            temp1 = name[0]
            temp2 = name[1]

            draw_data = " {} : {} ".format(temp1,temp2)
            stdscr.addstr(y_data + i, x_data, draw_data)

        elif switchdata == 3:
            temp1 = name[0]
            temp2 = name[1]
            temp3 = name[2]

            draw_data = " {} : {} , {} ".format(temp1,temp2,temp3)
            stdscr.addstr(y_data + i, x_data, draw_data)

        elif switchdata == 5:
            temp1 = name[0]
            temp2 = name[1]
            temp3 = name[2]
            temp4 = name[3]
            temp5 = name[4]

            draw_data = " {} : {} , {} , {} , {} ".format(temp1,temp2,temp3,temp4,temp5)
            stdscr.addstr(y_data + i, x_data, draw_data)

        elif switchdata == 6:
            temp1 = name[0]
            temp2 = name[1]
            temp3 = name[2]
            temp4 = name[3]
            temp5 = name[4]
            temp6 = name[5]

            draw_data = " {} : {} , {} , {} , {} , {} ".format(temp1,temp2,temp3,temp4,temp5,temp6)
            stdscr.addstr(y_data + i, x_data, draw_data)

        i=i+1

# 画面の描画
def func_draw(stdscr):

    # シリアル通信の設定(COM3は、各自のパソコンで違います)
    ser = serial.Serial(var,115200,timeout=50)

    # 初期化
    temp_data = []
    temp_list = ['NONE','NONE','NONE']
    temp_Rcvlist = ['NONE','NONE','NONE','NONE','NONE']
 
    # 各テーブルデータの配置を決める
    # 1列目
    y_frame1 = int(5)
    x_frame1 = int(0)

    y_frame2 = int(20)
    x_frame2 = int(0)

    # 2列目
    y_frame3 = int(5)
    x_frame3 = int(25)

    # 3列目
    y_RcvData = int(5)
    x_RcvData = int(80)


    # フレームデータ1の表示
    title_frame1 = "フレームデータ1 "
    table_frame1 = [
                                ['データ1 ','OFF'],
                                ['データ2 ','OFF'],
                                ['データ3 ','OFF'],
                                ['データ4 ','OFF'],
                                ['データ5 ','OFF'],
                                ['データ6 ','OFF'],
                                ['データ7 ','OFF'],
                                ['データ8 ','OFF']
                    ]

    # フレームデータ2の表示
    title_frame2 = "フレームデータ2 "
    table_frame2 = [
                            ['データ1 ','OFF'],
                            ['データ2 ','OFF'],
                            ['データ3 ','OFF'],
                            ['データ4 ','OFF']
                        ]

    # フレームデータ3の表示
    title_frame3 = " フレームデータ3 "
    table_frame3 = [
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE'],
                    ['NONE','NONE','NONE']
                ]

    # 受信データの表示
    title_RcvData = "受 信 データ (TIME,ID,DLC,DATA) "
    table_RcvData = [
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE'],
                        ['NONE','NONE','NONE','NONE','NONE']
                    ]

    # 画面の初期化
    stdscr.clear()
    stdscr.refresh()

    # カラーの定義
    curses.start_color()
    curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLACK)     #文字色=白、背景=黒
    curses.init_pair(2, curses.COLOR_YELLOW, curses.COLOR_BLACK)     #文字色=黄、背景=黒
    curses.init_pair(3, curses.COLOR_RED, curses.COLOR_BLACK)     #文字色=赤、背景=黒
    curses.init_pair(4, curses.COLOR_BLACK, curses.COLOR_WHITE)     #文字色=黒、背景=白

    # ループ
    while True:

        # キーが押下されたら、終了
        if msvcrt.kbhit(): # キーが押されているか
            sys.exit()

        # シリアル通信の受信
        byte_data = ser.readline()
        str_data = byte_data.decode().split()[0]

        try:
            # 受信データをJSON形式に変換
            json_data = json.loads(str_data)
            #print(json_data)

            temp_data = []

            # jsonデータから各データを取得 -----------------------------
            temp_id = json_data["ID"][0]
            temp_dlc = json_data["DLC"][0]

            for i in range(temp_dlc):
                temp_data.append(json_data["DATA"][i])

            # 受信データにセット ---------------------------
            temp_Rcvlist[0] = '-'
            temp_Rcvlist[1] = json_data["TIME"][0]
            temp_Rcvlist[2] = hex(temp_id)
            temp_Rcvlist[3] = temp_dlc
            temp_Rcvlist[4] = temp_data

            # 受信データを、テーブルデータに、FIFOでセット
            table_RcvData.append(list(temp_Rcvlist))    # テーブルデータの最後尾に追加
            table_RcvData.pop(0)    # テーブルデータの最初のデータを削除

            # フレーム1解析 ---------------------------
            if temp_id == 0x1:

                table_frame1[0][1] =temp_data[0]
                table_frame1[1][1] =temp_data[1]
                table_frame1[2][1] =temp_data[2]
                table_frame1[3][1] =temp_data[3]
                table_frame1[4][1] =temp_data[4]
                table_frame1[5][1] =temp_data[5]
                table_frame1[6][1] =temp_data[6]
                table_frame1[7][1] =temp_data[7]

            else:
                pass

            # フレーム2解析 ---------------------------
            if temp_id == 0x2:

                table_frame2[0][1] =temp_data[0]
                table_frame2[1][1] =temp_data[1]
                table_frame2[2][1] =temp_data[2]
                table_frame2[3][1] =temp_data[3]

            else:
                pass

            # フレーム3解析 ---------------------------
            if temp_id == 0x3:

               # 受信データリストを作成
                temp_list[0] = json_data["TIME"][0]
                temp_list[1] = hex(json_data["ID"][0])
                temp_list[2] = json_data["DATA"]
                # RCコマンドリストをRCリストの最後尾に追加
                table_frame3.append(list(temp_list))
                # RCリストの先頭を削除
                table_frame3.pop(0)

        except json.JSONDecodeError:
            continue
            #sys.exit()


        # 画面の初期化
        stdscr.clear()

        # 画面の最大値を取得(高さ、幅)
        height, width = stdscr.getmaxyx()

        # ーー文字列の表示ーー
        # タイトルの表示設定
        y_title = int(1)
        x_title = int(0)
        title = "シリアルデータモニター "[:width-1]
        stdscr.attron(curses.color_pair(3))     # 文字色を、カラー3に切り替えON
        stdscr.attron(curses.A_BOLD)     # 太字をONに切り替え
        stdscr.addstr(y_title, x_title, title)      # タイトルを描画
        stdscr.attroff(curses.color_pair(3))     # 文字色を、カラー3に切り替えOFF
        stdscr.attroff(curses.A_BOLD)     # 太字をOFFに切り替え

        # 現在時刻の表示設定
        y_time = int(1)
        x_time = int(40)
        stdscr.attron(curses.color_pair(3))    # 文字色を、カラー3に切り替えON
        stdscr.attron(curses.A_BOLD)     # 太字をONに切り替え
        draw_time = " UNIX Time : {} ".format(time.time())
        stdscr.addstr(y_time, x_time, draw_time)    # 現在時刻を描画
        stdscr.attroff(curses.color_pair(3))    # 文字色を、カラー3に切り替えOFF
        stdscr.attroff(curses.A_BOLD)     # 太字をOFFに切り替え

        # 受信時刻の表示設定
        y_time = int(3)
        x_time = int(0)
        stdscr.attron(curses.color_pair(3))    # 文字色を、カラー3に切り替えON
        stdscr.attron(curses.A_BOLD)     # 太字をONに切り替え
        draw_time = " Received Time : {} ".format(json.dumps(json_data["TIME"]))
        stdscr.addstr(y_time, x_time, draw_time)    # 受信時刻の描画
        stdscr.attroff(curses.color_pair(3))    # 文字色を、カラー3に切り替えOFF
        stdscr.attroff(curses.A_BOLD)     # 太字をOFFに切り替え

        # ステータスバーの表示設定
        statusbarstr = " な に か の キ ー を 押 下 す る と 終 了"
        stdscr.attron(curses.color_pair(4))    # 文字色を、カラー4に切り替えON
        stdscr.addstr(height-1, 0, statusbarstr)    # ステータスバーを描画
        stdscr.addstr(height-1, len(statusbarstr), " " * (width - len(statusbarstr) - 1))   # ステータスバーの空きをスペースで埋める
        stdscr.attroff(curses.color_pair(4))    # 文字色を、カラー4に切り替えOFF

        # フレーム1の表示
        func_drawblock(table_frame1,y_frame1,x_frame1,stdscr,title_frame1,2)

        # フレーム2の表示
        func_drawblock(table_frame2,y_frame2,x_frame2,stdscr,title_frame2,2)

        # フレーム3の表示
        func_drawblock(table_frame3,y_frame3,x_frame3,stdscr,title_frame3,3)

        # 受信データの表示
        func_drawblock(table_RcvData,y_RcvData,x_RcvData,stdscr,title_RcvData,5)

        # 画面をリフレッシュする
        stdscr.refresh()

# メイン関数
def main():
    curses.wrapper(func_draw)

if __name__ == "__main__":
    main()

このまま実行してもいいですが、よりアプリケーションっぽくするために、実行形式に変換します。

pyinstaller sample.py --onefile

以下のコマンドを実行すると、exeファイルが生成されます。

そのexeファイルをダブルクリックすると、以下の画面を持つアプリが起動します。

コンソールの大きさがデフォルトのままだと、アプリが強制停止してしまうので、
コンソールのサイズを変更します。

①ウィンドウのタイトルバーのところで、右クリックする
②「プロパティ」を選択
③「ウィンドウのサイズ」の欄を、幅:170、高さ:45に変更

なにかキーを押して、「ENTRY」を押すと、下図の画面になります。

M5ATOMとつながるCOMポート番号を入力して、「ENTRY」を押してください。

多分、COM3になります。ご自分のPCの状況を確認してください。

そうすると、以下のように、シリアル通信データをリアルタイムで表示する画面が出ます。

\ Pythonで遊ぶなら、ラズパイがお薦め。特にこのキーボード付きがオススメです /

まとめ

今回は、Python、cursesを使用したTUI(テキストユーザインターフェース)アプリとして、シリアル通信をリアルタイムに表示するアプリの作り方を紹介しました。

コメント

タイトルとURLをコピーしました