【python-can】ISO-TPのリクエストとレスポンスをシミュレーションするぞ！

みなさま、故障診断していますか？

本記事はクルマの故障診断通信に使われるISO-TPについて理解しよう！pythonを使ってCAN通信のシミュレーションをしよう！という試みの記事。第5回目です。

CAN素人の筆者が書いたCAN関連記事は以下です。気になる方は見てみて下さい。

0回目：「CAN通信って何？」という人向けにCAN通信の解説をしてみる

1回目：【Vector Virtual CAN BUS & Python-can】CAN通信のシミュレーション環境構築

2回目：【python-can】CANを定期的に送信する！！

3回目：【Python-can】受信したら返答するCANプログラム

4回目：【python-can】ISO-TPの勉強！シミュレーションで動作確認するぞ！！

5回目：【python-can】ISO-TPのリクエストとレスポンスをシミュレーションするぞ！(本記事)

6回目：【python-can】ISO-TP通信でFunctional Addressで送信する

7回目：【python-can】CANの送受信をするclassを作った

前回はとりあえずISO-TPを使った通信シミュレーションを流してみました。

今回はリクエストとレスポンスをシミュレーションしてみようと思います。

あ？リクエスト？レスポンス？なにいってんだこいつ？

と、わからないことにイラつく方は同じようにわからない人間が書いた記事をまぁ読んでいってみて下さいな。

リクエストとレスポンス
おっしゃ！シミュレーションすっぞ！！
1. 実際のコード
2. 実行結果
まとめ

リクエストとレスポンス

ISO-TPを使うときは基本リクエストとレスポンスがセットになってます。

例えば、クルマの故障診断です。

クルマの故障診断をするときはDLCコネクタにテスターやらをぶすっと差してクルマの故障がないかを調べます。そのときにテスターが車内の部品たちに

テスター

おーい！！おめぇら異常があるか？あるならどんな異常があるか教えろや！！

これがリクエストです。このリクエストに対して聞かれた部品たちは

部品A

異常がありません。

部品B

異常あります。壊れてます。

と答えます。これがレスポンスです。

このリクエストとかレスポンスをISO-TPで通信するので、そこそこ長いメッセージをやり取りできるって感じです。

なので、今日はISO-TPを使ってリクエストしたらレスポンスする。ってのをシミュレーションしてみたいと思います。

おっしゃ！シミュレーションすっぞ！！

とりあえずシミュレーションをやってみましょう。今回のコードはほぼこちらを参照にしています。

Examples — isotp 1.8 documentation

ここの”Threaded reception with python-can”を試してみたって感じです。

実際のコード

送信側(リクエストする側) can_send.py

import isotp
import logging
import time
import threading
import can

#ISO-TP通信をスレッドで実行するclassを作る。
class ThreadedApp:
    def __init__(self):#classが呼び出されたときにBUSに接続したりIDを設定したりする。
        self.ext_requested = False
        #バス接続
        self.bus = can.interface.Bus(bustype='vector',
                                     channel=0,
                                     bitrate=500000,
                                     app_name='python-can')
        #ID,frame 受信ID：123, 送信ID：456
        self.addr =  isotp.Address(isotp.AddressingMode.Normal_11bits,
                                   rxid=0x123,txid=0x456,)
        self.stack = isotp.CanStack(self.bus, address=self.addr,
                                    error_handler=self.my_error_handler)

    def start(self):#スレッド処理を実行する関数
        self.exit_requested = False
        self.thread = threading.Thread(target = self.thread_task)
        self.thread.start()
    
    def stop(self):
        self.exit_requested = True
        if self.thread.isAlive():
            self.thread.join()

    def my_error_handler(self, error):
      logging.warning('IsoTp error happened : %s - %s' % (error.__class__.__name__, str(error)))

    def thread_task(self):#わからん
      while self.exit_requested == False:
         self.stack.process()                # Non-blocking
         time.sleep(self.stack.sleep_time()) # Variable sleep time based on state machine state

    def shutdown(self):
      self.stop()
      self.bus.shutdown()

if __name__ == '__main__':
    app = ThreadedApp()  #appをThreadedApp class化 BUS接続とID設定などを行う。
    app.start()          #app classでstart関数を実行。iso-tpのスレッド処理を実行。

    print('Waiting for payload - maximum 5 sec')

    #request messageの送信設定
    #msg = b'\x01\x02\x03\x04'#Signal Frame
    msg = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b'#Multi Frame
    app.stack.send(msg)

    #シミュレーション実行時間は5秒間。
    t1 = time.time()
    while time.time() - t1 < 5:
        if app.stack.available():
            payload = app.stack.recv()
            break
    time.sleep(0.2)

    #5秒経過したら終わり。
    print("Exiting")
    app.shutdown()

受信側(リクエストを受けてレスポンスする側)can_recv.py

import isotp
import logging
import time
import threading
import can

class ThreadedApp:
    def __init__(self):
        self.ext_requested = False#これ何？
        #バス接続
        self.bus = can.interface.Bus(bustype='vector',
                                     channel=0,
                                     bitrate=500000,
                                     app_name='python-can')
        #ID,frame 受信ID：456, 送信ID：123
        self.addr =  isotp.Address(isotp.AddressingMode.Normal_11bits,
                                   rxid=0x456,txid=0x123,)
        #
        self.stack = isotp.CanStack(self.bus, address=self.addr,
                                    error_handler=self.my_error_handler)

    def start(self):
        self.exit_requested = False
        self.thread = threading.Thread(target = self.thread_task)
        self.thread.start()
    
    def stop(self):
        self.exit_requested = True
        if self.thread.isAlive():
            self.thread.join()

    def my_error_handler(self, error):
      logging.warning('IsoTp error happened : %s - %s' % (error.__class__.__name__, str(error)))

    def thread_task(self):
      while self.exit_requested == False:
         self.stack.process()                # Non-blocking
         time.sleep(self.stack.sleep_time()) # Variable sleep time based on state machine state

    def shutdown(self):
      self.stop()
      self.bus.shutdown()

if __name__ == '__main__':
    app = ThreadedApp()  #appをThreadedApp class化
    app.start()          #app classでstart関数を実行

    print('Waiting for payload - maximum 5 sec')

    t1 = time.time()
    while time.time() - t1 < 5:
        if app.stack.available():
            payload = app.stack.recv()
			#response message
            msg = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b'
            app.stack.send(msg)
            break
    time.sleep(0.2)

    print("Exiting")
    app.shutdown()