识别未知串行协议的帧长度

#!/usr/bin/env python3
"""
ProtocolFrameLength.py

确定具有等长帧且每帧包含相似字节的未知串行协议的帧长度。

有关说明，请参见
https://techoverflow.net/2017/07/30/identifying-the-frame-length-for-an-unknown-serial-protocol/

示例用法：
$ python3 ProtocolFrameLength.py -b 115200 /dev/ttyACM0
"""
import serial
import numpy as np
import math
from functools import reduce
import operator

__author__ = "Uli Köhler"
__license__ = "Apache License v2.0"
__version__ = "1.0"
__email__ = "ukoehler@techoverflow.net"

def match_score(c1, c2):
    """
    两个字符 c1 和 c2 的相关性分数。
    使用简单的二进制分数
    """
    if c1 is None or c2 is None:  # 填充字符
        return 0
    return 1 if c1 == c2 else 0

def string_match_score(s1, s2):
    assert len(s1) == len(s2)
    ln = len(s1)
    return sum(match_score(s1[i], s2[i]) for i in range(ln))

def compute_correlation_scores(chars, nomit=-1):
    # 省略最后 nomit 个字符，因为单字符匹配会被过高估值
    if nomit == -1: # 自动值
        nomit = len(chars) // 10
    corr = np.zeros(len(chars) - nomit)
    # 注意：零移位的自相关始终为 1，故意省略！
    for i in range(1, corr.size):
        # 用 None 构建前缀
        prefix = [None] * i
        s2 = prefix + list(chars[:-i])
        # 由于 None 和序列长度（有 i 个 None）可达到的最大分数进行归一化
        corr[i] = string_match_score(chars, s2) / (len(chars) - i)
    return corr

def print_most_likely_frame_length(correlations):
    # 找到最大的相关系数。此模型不需要阈值
    idx = np.argmax(correlations)
    print("帧长度可能是 {} 字节".format(idx))

def plot_correlations(correlations):
    from matplotlib import pyplot as plt
    plt.style.use("ggplot")

    plt.title("具有 16 字节帧的协议的相关性分数")
    plt.gcf().set_size_inches(20,10)
    plt.plot(correlations)

    plt.title("相关性分数")
    plt.ylabel("归一化相关性分数")
    plt.xlabel("移位")

    plt.show()

def multishift_adjust(correlations):
    """
    多移位感知算法
    """
    corr_multishift = np.zeros(correlations.size // 4)
    for i in range(1, corr_multishift.size):
        # 迭代 i 的倍数（包括 i 本身）
        corr_multishift[i] = reduce(operator.mul,
            (correlations[j] for j in range(i, correlations.size, i)), 1)
    return corr_multishift

if __name__ == "__main__":

    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument('port', help='要使用的串行端口')
    parser.add_argument('-b', '--baudrate', type=int, default=9600, help='要使用的波特率')
    parser.add_argument('-n', '--num-bytes', type=int, default=200, help='要读取的字符数')
    parser.add_argument('-m', '--multishift', action="store_true", help='使用多移位感知自相关算法')
    parser.add_argument('-p', '--plot', action="store_true", help='绘制结果相关性矩阵')
    args = parser.parse_args()

    ser = serial.Serial(args.port, args.baudrate)
    ser.reset_input_buffer()
    chars = ser.read(args.num_bytes)

    corr = compute_correlation_scores(chars)
    if args.multishift:
        corr = multishift_adjust(corr)
    print_most_likely_frame_length(corr)

    if args.plot:
        plot_correlations(corr)

python3 ProtocolFrameLength.py -b 115200 /dev/ttyACM0