概要

本記事の想定対象者
- 車載ネットワークであるCAN通信をWindowsから安価なデバイス(CANalyze等)かつPythonで行いたい方
- libusb/WinUSBドライバー経由でデバイスをPythonから制御してみたい方(PyUSB使用)
- Linux上でのUSB通信の解析方法の例を知りたい方(usbmon使用)
CAN通信をPythonから行う場合、LinuxでSocketCANやSerial can(slcan)対応のデバイス(PCAN,CANalyze,USBCAN,...)を使用すると安価で便利
しかし、WindowsにはSocketCANという概念が無いため、slcan相当のデバイスを使用するしかなかった(他にも下記方法があるようだが、それ以外は単に知らない)
- 他の方法としては、CANALドライバー(usb2can.dll(32bit))を使用してusb 8dev対応のPCANなどに接続する方法があるが、手元のCANalyzeではデバイスが見つからないとエラーが出てうまくいかなかった
そこで、linuxのusb 8devドライバー対応のCAN通信デバイス(CANalyze)のファームウェアソースコードを解析し、Windowsでlibusb/WinUSBドライバー経由でSocketCAN対応のデバイスにも接続できるようにした
- CANalyzeに対するlibusb(32bit)/WinUSB(64bit)ドライバーの適用にはzadigを使用した
- Pythonを使用しているが、C言語等でもlibusb経由で同様のコマンドを送受信すればCAN通信が行えるはず

デバイス準備

usb 8dev対応デバイス(ここではCANalyze)を調達する
- 引用元 https://kkuchera.github.io/canalyze/

USB通信の解析

まずLinux上で利用した際のUSB通信を解析する。以下の手順で送受信を行う

# apt-get install can-utils
# ip link show
can0が存在することを確認
# ip link set can0 type can bitrate 500000
# ip link set can0 up
# cansend can0 123#1122334455667788
# candump can0

Ctrl^C
# ip link set can0 down

usbmonドライバーのインストールとデバイスの確認

# modprobe usbmon
# cat /sys/kernel/debug/usb/devices

T:  Bus=02 Lev=02 Prnt=03 Port=00 Cnt=01 Dev#=  5 Spd=12   MxCh= 0
D:  Ver= 2.00 Cls=00(>ifc ) Sub=00 Prot=00 MxPS=64 #Cfgs=  1
P:  Vendor=0483 ProdID=1234 Rev= 1.00
S:  Manufacturer=STMicroelectronics
S:  Product=USB2CAN converter
S:  SerialNumber=205236******
C:* #Ifs= 1 Cfg#= 1 Atr=00 MxPwr=100mA
I:* If#= 0 Alt= 0 #EPs= 4 Cls=ff(vend.) Sub=ff Prot=ff Driver=usb_8dev
E:  Ad=81(I) Atr=02(Bulk) MxPS=  64 Ivl=0ms
E:  Ad=02(O) Atr=02(Bulk) MxPS=  64 Ivl=0ms
E:  Ad=83(I) Atr=02(Bulk) MxPS=  64 Ivl=0ms
E:  Ad=04(O) Atr=02(Bulk) MxPS=  64 Ivl=0ms

USB2CAN converter(CANalyze)のUSB論理構成
└ Interface:1 (usb_8dev)
　├ Endopint:1 Bulk Input (USB調査よりData入力用)
　├ Endopint:2 Bulk Output (USB調査よりData出力用)
　├ Endopint:3 Bulk Input (USB調査よりCommand入力用)
　└ Endopint:4 Bulk Output (USB調査よりCommand出力用)

ip link set can0～, ip link set can0 upを実行した際のUSB通信の解析

# ip link set can0 type can bitrate 500000
# ip link set can0 up
別ターミナル(事前に起動しておく)
# cat /sys/kernel/debug/usb/usbmon/2u
  ※cat /sys/kernel/debug/usb/devicesの "T:  Bus=02" より2uを使用
ffff880232886d80 3205323523 S Bo:2:007:4 -115 16 = 11000209 000d0201 00040000 00080022
　※Endpoint:4でのbulk送信
ffff880232886d80 3205450861 C Bi:2:007:3 0 16 = 11000200 00000100 01000000 00000022
　※Endpoint:3でのbulk送信

※Bo/Biの意味はUSBのBulk通信のIn/Outという意味
Ci Co   Control input and output
Zi Zo   Isochronous input and output
Ii Io   Interrupt input and output
Bi Bo   Bulk input and output

※ 送信データの解析
"11 00 02 09 00 0d020100040000000800 22"
/* Format of both received and transmitted USB command messages. */
typedef struct __packed usb_8dev_cmd_msg {
    uint8_t start;       // start of message byte
    uint8_t channel;     // unknown - always 0
    uint8_t command;     // command to execute
    uint8_t opt1;        // optional parameter / return value
    uint8_t opt2;        // optional parameter 2
    uint8_t data[10];    // optional parameter and data
    uint8_t end;         // end of message byte
} Msg_CmdTypeDef;
// https://github.com/kkuchera/canalyze-fw/blob/master/src/usbd_8dev_if.c

ip link set can0 downを実行した際のUSB通信の解析

# ip link can0 down
別ターミナル(事前に起動しておく)
# cat /sys/kernel/debug/usb/usbmon/2u
ffff880232886cc0 3256831570 S Bo:2:007:4 -115 16 = 11000300 00000000 00000000 00000022
ffff880232886cc0 3256843005 C Bi:2:007:3 0 16 = 11000300 00000100 01000000 00000022

cansendのUSB通信取得

# cansend can0 555#1122334455667788
別ターミナル(事前に起動しておく)
# cat /sys/kernel/debug/usb/usbmon/2u
ffff880232887980 3421490964 S Bo:2:007:2 -115 16 = 55000000 05550811 22334455 667788aa
　※Endpoint:2でのbulk送信

# cansend can0 777#8888888888888888
別ターミナル(事前に起動しておく)
# cat /sys/kernel/debug/usb/usbmon/2u
ffff8802328866c0 3619076135 S Bo:2:007:2 -115 16 = 55000000 07770888 88888888 888888aa

cansendのUSB通信を解析

55 00 00000555 08 1122334455667788 aa
55 00 00000777 08 8888888888888888 aa
^^USB_8DEV_DATA_START=0x55
      ^^^^^^^^ID(標準11bit, 拡張29bit)
               ^^DLC(1byte)
                  ^^^^^^^^^^^^^^^^Data(8bye 0x00パディング)
                                   ^^USB_8DEV_DATA_END=0xAA

/* Format of received USB data messages. */
typedef struct __packed usb_8dev_rx_msg {
    uint8_t start;      // start of message byte
    uint8_t flags;      // RTR and EXT_ID flag
    uint32_t id;        // upper 3 bits not used
    uint8_t dlc;        // data length code 0-8 bytes
    uint8_t data[8];    // 64-bit data
    uint8_t end;        // end of message byte
} Msg_RxTypeDef;

candumpのUSB通信取得

# candump can0
別ターミナル(事前に起動しておく)
# cat /sys/kernel/debug/usb/usbmon/2u
ffff880232886cc0 3667542562 C Bi:2:007:1 0 21 = 55000100 00055508 11223344 55667788 e65e0700 aa
　※Endpoint:1でのbulk受信
　※ID 0x555, 0x1122334455667788を受信
ffff8800b8d76c00 220316803 C Bi:2:008:1 0 21 = 55000100 00077708 88888888 88888888 b0990000 aa
　※Endpoint:1でのbulk受信
　※ID 0x777, 0x8888888888888888を受信

candumpのUSB通信を解析

55 00 01 00000555 08 1122334455667788 e65e0700 aa
55 00 01 00000777 08 8888888888888888 b0990000 aa
^^USB_8DEV_DATA_START=0x55
   ^^type
      ^^flags
         ^^^^^^^^ID(標準11bit, 拡張29bit)
                  ^^DLC(1byte)
                     ^^^^^^^^^^^^^^^^Data(8bye 0x00パディング)
                                      ^^^^^^^^timestamp
                                               ^^USB_8DEV_DATA_END=0xAA

/* Format of transmitted USB data messages. */
typedef struct __packed usb_8dev_tx_msg {
    uint8_t start;      // start of message byte
    uint8_t type;       // frame type
    uint8_t flags;      // RTR and EXT_ID flag
    uint32_t id;        // upper 3 bits not used
    uint8_t dlc;        // data length code 0-8 bytes
    uint8_t data[8];    // 64-bit data
    uint32_t timestamp; // 32-bit timestamp
    uint8_t end;        // end of message byte
} Msg_TxTypeDef;

Python on windowsでの制御

Windowsで行っているがlibusb経由であればLinuxでも使えるはず。但しLinuxの場合SocketCANの方が対応ツールも多く便利

ドライバーのインストール

zadigを使用して、Options > List All Devicesを選択 > リストボックスから"USB2CAN converter"(=CANalyze)を選択
Pythonの32/64bitに合わせてドライバーをインストールする。32bitの場合はlibusb-win32を、64bitの場合はWinUSB(例、下図)を選択

pythonでの通信

"""
# pip install can
# pip install pyusb
"""

import binascii
import usb.core
import usb.util

USB_8DEV_CMD_START      = 0x11
USB_8DEV_CMD_END        = 0x22
USB_8DEV_DATA_START     = 0x55
USB_8DEV_DATA_END       = 0xAA
USB_8DEV_OPEN           = 0x2
USB_8DEV_CLOSE          = 0x3
USB_8DEV_GET_SOFTW_HARDW_VER= 0xC

def init():
  dev = usb.core.find(idVendor=0x0483, idProduct=0x1234)
  dev.set_configuration()
  cfg = dev.get_active_configuration()
  itfs = cfg[(0,0)]
  dat_in = usb.util.find_descriptor(itfs, bEndpointAddress=0x81)
  dat_out = usb.util.find_descriptor(itfs, bEndpointAddress=0x2)
  cmd_in = usb.util.find_descriptor(itfs, bEndpointAddress=0x83)
  cmd_out = usb.util.find_descriptor(itfs, bEndpointAddress=0x4)
  s = [dat_in, dat_out, cmd_in, cmd_out]
  return s

def send_wait_cmd(s, msg):
  cmd_in, cmd_out = s[2], s[3]
  if cmd_out.write(msg) != len(msg):
    print("send error")
  return cmd_in.read(128)

def make_cmd(command, opt1 = 0, opt2 = 0, data = ""):
  channel = 0
  dlc_max = 8
  data += "0" * (2*dlc_max - len(data))
  msg = "1100%02x%02x%02x%s22" % (command, opt1, opt2, data)
  return binascii.unhexlify(msg)

def version(s):
  # get firmwate and hardware version
  msg = make_cmd(USB_8DEV_GET_SOFTW_HARDW_VER, data="00"*10)
  ret = send_wait_cmd(s, msg)
  print(ret)

def open(s):
  ctrlmode = 0x08
  msg = make_cmd(USB_8DEV_OPEN, opt1=0x09, data="0d02010004000000%02x00" % ctrlmode)
  ret = send_wait_cmd(s, msg)
  print(ret)

def close(s):
  msg = make_cmd(USB_8DEV_CLOSE, data="00"*10)
  ret = send_wait_cmd(s, msg)
  print(ret)

def make_dat(flags, id, data = ""):
  # flags : RTR and EXT_ID (USB_8DEV_EXTID) flag
  # id : 11 / 29bit
  dlc = len(data)
  if len(data)%2 != 0:
    data = "0" + data
  data += "00" * (8-dlc)
  msg = "55%02x%08x%02x%saa" % (flags, id, dlc, data)
  return binascii.unhexlify(msg)

def send(s, msg_id, data):
  dat_out = s[1]
  msg = make_dat(flags = 0, id = msg_id, data = data)
  if dat_out.write(msg) != len(msg):
    return False
  return True

def recv(s, timeout = 1000):
  dat_in = s[0]
  msg = dat_in.read(64, timeout = timeout)
  if msg[0] != USB_8DEV_DATA_START or msg[-1] != USB_8DEV_DATA_END or 13 + msg[7] != len(msg):
    return None
  frame_type, flags, id, dlc, data, timestamp = msg[1], msg[2], msg[3:7], msg[7], msg[8:-5], msg[-5:-1]
  id = (((((id[0]<<8) + id[1])<<8) + id[2])<<8) + id[3]
  data = [i for i in data]
  timestamp = (((((timestamp[3]<<8) + timestamp[2])<<8) + timestamp[1])<<8) + timestamp[0]
  return [id, data, timestamp]

送受信テスト

s = init()
open(s)
msg_id = 0x7ff
data = "1122334455667788"
send(s, msg_id, data)
while 1:
  ret = recv(s)
  if ret is not None:
    break

msg_id, data, timestamp = ret
close(s)