バックナンバーはこちら。
https://www.simulationroom999.com/blog/diagnostic-communication-backnumber/
はじめに
AUTOSAR-Dcmのシミュレーションの話。
TesterPresentsのシミュレーションのPythonコードについて。
登場人物
博識フクロウのフクさん
イラストACにて公開の「kino_k」さんのイラストを使用しています。
https://www.ac-illust.com/main/profile.php?id=iKciwKA9&area=1
エンジニア歴8年の太郎くん
イラストACにて公開の「しのみ」さんのイラストを使用しています。
https://www.ac-illust.com/main/profile.php?id=uCKphAW2&area=1
TesterPresentsのシミュレーション用のPythonコード
じゃ、さっくりPythonコードを。
前回よりかはシンプルだよ。
import isotp
import logging
import time
import threading
from can.interfaces.vector import VectorBus
class ThreadedApp:
def __init__(self):
isotp_params = {
'stmin' : 0,
'blocksize' : 4,
'wftmax' : 0,
'll_data_length' : 8,
'tx_padding' : 0xCC,
'rx_flowcontrol_timeout' : 1000,
'rx_consecutive_frame_timeout' : 1000,
'squash_stmin_requirement' : False,
'can_fd' : False,
'tx_data_min_length' : 8
}
self.exit_requested = False
#self.bus = VectorBus(channel='0', bitrate=500000)
self.bus = VectorBus(channel='0', bitrate=500000, fd=True)
addr = isotp.Address(isotp.AddressingMode.NormalFixed_29bits, source_address=0xF1, target_address=0x10)
self.stack = isotp.CanStack(self.bus, address=addr, params=isotp_params, 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 send(self, msg):
self.stack.send(msg)
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
time.sleep(0.001) # Variable sleep time based on state machine state
def shutdown(self):
self.stop()
self.bus.shutdown()
def sendrecv( app, msg ):
if msg[0] == 0x00:
tsleep = msg[1]*0x100 + msg[2]
tsleep = tsleep / 1000
print("sleep : %f [ms]" % tsleep)
time.sleep(tsleep)
else:
print("Send msg : %s" % (msg.hex()))
app.send(msg)
t1 = time.time()
while time.time() - t1 < 0.05:
if app.stack.available():
payload = app.stack.recv()
print("Recv msg : %s" % (payload.hex()))
break
time.sleep(0.001)
if __name__ == '__main__':
app = ThreadedApp()
app.start()
datas=[
bytes([0x3E, 0x00]),
bytes([0x3E, 0x00, 0x00]),
bytes([0x3E, 0x01]),
bytes([0x3E, 0x80]),
bytes([0x3E, 0x80, 0x00]),
bytes([0x3E, 0x81]),
bytes([0x10, 0x03]),
bytes([0x27, 0x13]),
bytes([0x27, 0x14, 0xDE, 0xAD, 0xBE, 0xEF]),
bytes([0x00, 0x13, 0x24]),
bytes([0x3E, 0x80]),
bytes([0x00, 0x13, 0x24]),
bytes([0x3E, 0x80]),
bytes([0x00, 0x13, 0x24]),
bytes([0x27, 0x13]),
bytes([0x00, 0x13, 0xEC]),
bytes([0x27, 0x13]),
]
#while True:
for i in range(len(datas)):
sendrecv(app, datas[i])
print("Exiting")
app.shutdown()
TesterPresentsのシミュレーション用のPythonコード解説
んー。
今回はTesterPresentsのはずだけど、
メッセージ的にはSessionControl、SecurityAccessもいるねー。
TesterPresentsの利用シーンがセッションの維持だからねー。
よって、セッションとセキュリティも混ぜて、5秒以内なら維持、5秒以上かかるとdefaultSessionに戻るよー。
って流れだねー。
年のため詳細な流れの説明よろしく。
はいよ。
①普通にTesterPresents
②TesterPresentsのメッセージ長間違い
③TesterPresentsの存在しないsub-function
④TesterPresents(suppressPosRspMsgIndicationBitあり)
⑤TesterPresentsのメッセージ長間違い(suppressPosRspMsgIndicationBitあり)
⑥TesterPresentsの存在しないsub-function(suppressPosRspMsgIndicationBitあり)
⑦extendDiagnosticSessionへ遷移
⑧Seed要求
⑨Key送信(Key=0xdeadbeef)
⑩4.900[ms] Wait
⑪TesterPresents(suppressPosRspMsgIndicationBitあり)
⑫4.900[ms] Wait
⑬TesterPresents(suppressPosRspMsgIndicationBitあり)
⑭4.900[ms] Wait
⑮Seed要求
⑯5.100[ms] Wait
⑰Seed要求
最初の方はsuppressPosRspMsgIndicationBitありなしの両方で通常パターン、エラーパターンを試してる感じだね。
途中からセキュリティアンロックして、S3タイムアウト直前にTesterPresentsを投げてセッション維持。
セッションが維持されているかは試しにSeed要求してみてSeedが0値だったら維持されているってことかな?
大正解!
今回はバッチリじゃない?!
前回と比べべたらこっちはチョロいよ!
まとめ
まとめだよ。
- TesterPresentsのシミュレーション用のPythonコード書いた。
- TesterPresentsの主目的から考えて、SessionControl、SecurityAccessも実施する。
- S3タイムアウト抑制が主目的。
バックナンバーはこちら。
コメント