CANopen 日志分析与自动化

# 日志文件命名规范
can_<interface>_<timestamp>_<scenario>.log

# 示例
can_vcan0_20240520_143022_sdo_timeout.log
can_can0_20240520_150100_production.log

#!/bin/bash
# log_header.sh - 生成标准日志头

cat << 'EOF'
# CANopen Log File
# Generated: $(date -Iseconds)
# Interface: $INTERFACE
# Bitrate: $BITRATE
# Node IDs: $NODES
# Scenario: $SCENARIO
# Tool: candump
# Version: $(candump -h | head -1)
#
# Frame Format: (timestamp) interface id#data
# Timestamp Precision: microseconds
#
EOF

#!/bin/bash
# log_stats.sh - 日志统计工具

LOG_FILE=$1

if [ -z "$LOG_FILE" ]; then
    echo "Usage: $0 <log_file>"
    exit 1
fi

echo "=== CANopen 日志分析 ==="
echo "文件: $LOG_FILE"
echo ""

# 总帧数
total_frames=$(wc -l < "$LOG_FILE")
echo "总帧数: $total_frames"

# 各消息类型统计
echo ""
echo "=== 消息类型分布 ==="
awk '{print $3}' "$LOG_FILE" | cut -d'#' -f1 | sort | uniq -c | sort -rn

# 节点统计
echo ""
echo "=== 节点分布 ==="
awk '{print $3}' "$LOG_FILE" | cut -d'#' -f1 | grep -o '.$' | sort | uniq -c | sort -rn

# 时间跨度
first_time=$(head -1 "$LOG_FILE" | grep -oP '^\(\K[0-9.]+')
last_time=$(tail -1 "$LOG_FILE" | grep -oP '^\(\K[0-9.]+')
duration=$(echo "$last_time - $first_time" | bc)
echo ""
echo "时间跨度: $duration 秒"
echo "平均帧率: $(echo "scale=2; $total_frames / $duration" | bc) 帧/秒"

#!/bin/bash
# extract_sdo.sh - 提取 SDO 会话

LOG_FILE=$1

echo "=== SDO 会话 ==="
grep -E "(60[0-9A-F]{2}|58[0-9A-F]{2})#" "$LOG_FILE" | \
awk '
BEGIN { last_cmd = ""; }
{
    frame = $0
    id = substr($3, 1, 3)
    data = substr($3, 5)

    # SDO Request
    if (id ~ /^60/) {
        ccs = substr(data, 1, 1)
        idx = substr(data, 4, 2) substr(data, 2, 2)
        subidx = substr(data, 6, 2)

        last_cmd = frame
        last_id = id
        last_idx = idx
        last_subidx = subidx

        # Decode CCS
        if (ccs == "2") printf "Read 0x%s:%s -> ", idx, subidx
        if (ccs == "3") printf "Write 0x%s:%s -> ", idx, subidx
    }
    # SDO Response
    else if (id ~ /^58/ && last_id == substr(id, 2, 2)) {
        scs = substr(data, 1, 1)

        # Decode SCS
        if (scs == "4") {
            value = substr(data, 8, 2) substr(data, 6, 2) substr(data, 4, 2) substr(data, 2, 2)
            printf "Value: 0x%s\n", value
        } else if (scs == "8") {
            abort = substr(data, 8, 2) substr(data, 6, 2) substr(data, 4, 2) substr(data, 2, 2)
            printf "ABORT: 0x%s\n", abort
        }

        last_cmd = ""
    }
}'

#!/bin/bash
# heartbeat_monitor.sh - 心跳状态监控

LOG_FILE=$1
TIMEOUT_MS=3000

echo "=== 心跳状态监控 (超时: ${TIMEOUT_MS}ms) ==="

# 提取心跳消息
grep -E "7[0-9A-F]{2}#" "$LOG_FILE" | \
awk -v timeout=$TIMEOUT_MS '
BEGIN {
    last_time[256] = 0
}

{
    # 解析帧
    match($0, "\\(([^)]+)\\)", ts_arr)
    match($3, "([0-9A-F]+)#", id_arr)
    match($3, "#([0-9A-F]+)", data_arr)

    time = ts_arr[1]
    node_id = strtonum("0x" substr(id_arr[1], 2, 2))
    state = strtonum("0x" data_arr[1])

    # 检查超时
    if (last_time[node_id] > 0) {
        elapsed = (time - last_time[node_id]) * 1000
        if (elapsed > timeout) {
            printf "[%.3f] 节点 %d 心跳超时: %.1f ms\n", time, node_id, elapsed
        }
    }

    # 输出状态变化
    if (last_state[node_id] != "" && last_state[node_id] != state) {
        states["00"] = "Boot-up"
        states["04"] = "Stopped"
        states["05"] = "Operational"
        states["7F"] = "Pre-Operational"
        printf "[%.3f] 节点 %d: %s -> %s\n", time, node_id, states[last_state[node_id]], states[state]
    }

    last_time[node_id] = time
    last_state[node_id] = state
}'

#!/usr/bin/env python3
# canopen_log_parser.py - CANopen 日志解析器

import re
from dataclasses import dataclass
from typing import List, Optional, Dict
from datetime import datetime

@dataclass
class CANFrame:
    """CAN 帧数据结构"""
    timestamp: float
    interface: str
    can_id: int
    data: bytes
    is_extended: bool = False
    is_error: bool = False

    @property
    def cob_id(self) -> int:
        """获取 COB-ID（不含功能码）"""
        return self.can_id & 0x7F

    @property
    def function_code(self) -> int:
        """获取功能码"""
        return self.can_id & 0x780

    @property
    def is_nmt(self) -> bool:
        return self.can_id == 0x000

    @property
    def is_sync(self) -> bool:
        return self.can_id == 0x080

    @property
    def is_emergency(self) -> bool:
        return self.function_code == 0x080

    @property
    def is_time_stamped(self) -> bool:
        return self.function_code == 0x100

    @property
    def is_sdo_request(self) -> bool:
        return (self.can_id & 0x7F0) == 0x600

    @property
    def is_sdo_response(self) -> bool:
        return (self.can_id & 0x7F0) == 0x580

    @property
    def is_heartbeat(self) -> bool:
        return (self.can_id & 0x780) == 0x700

    @property
    def is_tpdo(self) -> bool:
        return 0x180 <= self.can_id <= 0x4FF

    @property
    def is_rpdo(self) -> bool:
        return 0x200 <= self.can_id <= 0x57F


class CANopenLogParser:
    """CANopen 日志解析器"""

    def __init__(self, log_file: str):
        self.log_file = log_file
        self.frames: List[CANFrame] = []

    def parse(self) -> List[CANFrame]:
        """解析日志文件"""
        log_pattern = re.compile(
            r'\((?P<timestamp>[0-9.]+)\)\s+'
            r'(?P<interface>\w+)\s+'
            r'(?P<id>[0-9A-Fa-f]+)#'
            r'(?P<data>[0-9A-Fa-f]*)'
        )

        with open(self.log_file, 'r') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue

                match = log_pattern.match(line)
                if match:
                    frame = CANFrame(
                        timestamp=float(match.group('timestamp')),
                        interface=match.group('interface'),
                        can_id=int(match.group('id'), 16),
                        data=bytes.fromhex(match.group('data'))
                    )
                    self.frames.append(frame)

        return self.frames

    def filter_by_node(self, node_id: int) -> List[CANFrame]:
        """过滤特定节点的帧"""
        return [f for f in self.frames if f.cob_id == node_id]

    def get_sdo_sessions(self) -> List[Dict]:
        """提取 SDO 会话"""
        sessions = []
        pending = {}

        for frame in self.frames:
            if frame.is_sdo_request:
                # 解析 SDO Request
                ccs = frame.data[0] & 0xF
                idx = (frame.data[2] << 8) | frame.data[1]
                subidx = frame.data[3]

                node_id = frame.cob_id
                pending[node_id] = {
                    'request_time': frame.timestamp,
                    'node_id': node_id,
                    'ccs': ccs,
                    'index': idx,
                    'subindex': subidx
                }

            elif frame.is_sdo_response:
                node_id = frame.cob_id
                if node_id in pending:
                    session = pending[node_id]
                    session['response_time'] = frame.timestamp
                    session['latency_ms'] = (frame.timestamp - session['request_time']) * 1000

                    # 解析 SDO Response
                    scs = frame.data[0] & 0xF
                    session['scs'] = scs
                    session['success'] = (scs == 0x4) or (scs == 0x2)

                    sessions.append(session)
                    del pending[node_id]

        return sessions

    def get_heartbeat_states(self) -> Dict[int, List[tuple]]:
        """获取心跳状态历史"""
        states = {}

        for frame in self.frames:
            if frame.is_heartbeat and len(frame.data) >= 1:
                node_id = frame.cob_id
                state = frame.data[0]

                if node_id not in states:
                    states[node_id] = []
                states[node_id].append((frame.timestamp, state))

        return states


if __name__ == '__main__':
    import sys

    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <log_file>")
        sys.exit(1)

    parser = CANopenLogParser(sys.argv[1])
    frames = parser.parse()

    print(f"总帧数: {len(frames)}")

    # SDO 会话分析
    sdo_sessions = parser.get_sdo_sessions()
    print(f"SDO 会话: {len(sdo_sessions)}")

    timeout_sessions = [s for s in sdo_sessions if not s['success']]
    print(f"SDO 失败: {len(timeout_sessions)}")

    # 心跳分析
    states = parser.get_heartbeat_states()
    print(f"心跳节点: {len(states)}")

#!/usr/bin/env python3
# canopen_detector.py - CANopen 问题自动检测

from canopen_log_parser import CANopenLogParser, CANFrame

class CANopenIssueDetector:
    """CANopen 问题检测器"""

    def __init__(self, parser: CANopenLogParser):
        self.parser = parser
        self.issues = []

    def detect_all(self):
        """检测所有问题"""
        self.detect_sdo_timeouts()
        self.detect_heartbeat_missing()
        self.detect_nmt_start_failure()
        self.detect_bus_overload()
        self.detect_emergency_messages()
        return self.issues

    def detect_sdo_timeouts(self, timeout_ms=100):
        """检测 SDO 超时"""
        sdo_sessions = self.parser.get_sdo_sessions()

        for session in sdo_sessions:
            if not session.get('success'):
                self.issues.append({
                    'type': 'SDO_TIMEOUT',
                    'severity': 'ERROR',
                    'node_id': session['node_id'],
                    'index': hex(session['index']),
                    'subindex': hex(session['subindex']),
                    'message': f"SDO 读取失败: 0x{session['index']:04X}:{session['subindex']:02X}"
                })

    def detect_heartbeat_missing(self, timeout_ms=3000, node_ids=None):
        """检测心跳丢失"""
        if node_ids is None:
            # 自动检测所有节点
            frames = self.parser.frames
            node_ids = set(f.cob_id for f in frames if f.is_heartbeat)

        for node_id in node_ids:
            heartbeat_frames = [f for f in self.parser.frames
                              if f.is_heartbeat and f.cob_id == node_id]

            for i in range(1, len(heartbeat_frames)):
                prev_time = heartbeat_frames[i-1].timestamp
                curr_time = heartbeat_frames[i].timestamp
                interval_ms = (curr_time - prev_time) * 1000

                if interval_ms > timeout_ms:
                    self.issues.append({
                        'type': 'HEARTBEAT_MISSING',
                        'severity': 'WARNING',
                        'node_id': node_id,
                        'interval_ms': interval_ms,
                        'message': f"节点 {node_id} 心跳间隔过大: {interval_ms:.1f} ms"
                    })

    def detect_nmt_start_failure(self):
        """检测 NMT 启动失败"""
        # 查找 NMT Start 命令
        nmt_start = [f for f in self.parser.frames if f.is_nmt and f.data[0] == 0x01]

        # 检查是否所有节点都响应
        for frame in nmt_start:
            target_node = frame.data[1]
            if target_node == 0:  # 广播，检查所有节点
                continue

            # 检查是否有 Bootup 消息
            bootup = [f for f in self.parser.frames
                     if f.is_heartbeat and f.cob_id == target_node
                     and f.data[0] == 0x00]

            if not bootup:
                self.issues.append({
                    'type': 'NMT_START_FAILURE',
                    'severity': 'ERROR',
                    'node_id': target_node,
                    'message': f"节点 {target_node} 未响应 NMT Start 命令"
                })

    def detect_bus_overload(self, threshold=0.8, window_sec=1.0):
        """检测总线过载"""
        frames = self.parser.frames
        if not frames:
            return

        start_time = frames[0].timestamp
        end_time = frames[-1].timestamp
        duration = end_time - start_time

        # 计算平均帧率
        avg_rate = len(frames) / duration

        # 计算瞬时负载
        windows = int(duration / window_sec)
        for i in range(windows):
            window_start = start_time + i * window_sec
            window_end = window_start + window_sec

            window_frames = [f for f in frames if window_start <= f.timestamp < window_end]
            instant_rate = len(window_frames) / window_sec

            # 500kbps CANopen 理论最大帧率约为 5000 帧/秒
            max_rate = 5000
            load = instant_rate / max_rate

            if load > threshold:
                self.issues.append({
                    'type': 'BUS_OVERLOAD',
                    'severity': 'WARNING',
                    'timestamp': window_start,
                    'load_pct': load * 100,
                    'message': f"总线过载: 负载 {load*100:.1f}% ({instant_rate:.0f} 帧/秒)"
                })

    def detect_emergency_messages(self):
        """检测紧急消息"""
        emergency_frames = [f for f in self.parser.frames if f.is_emergency]

        for frame in emergency_frames:
            error_code = (frame.data[1] << 8) | frame.data[0]
            error_register = frame.data[2]

            self.issues.append({
                'type': 'EMERGENCY',
                'severity': 'ERROR',
                'node_id': frame.cob_id,
                'error_code': hex(error_code),
                'error_register': hex(error_register),
                'message': f"紧急消息: 节点 {frame.cob_id}, 错误码 0x{error_code:04X}"
            })

    def print_report(self):
        """打印检测报告"""
        if not self.issues:
            print("未检测到问题")
            return

        print(f"\n=== CANopen 问题检测报告 ===")
        print(f"检测到 {len(self.issues)} 个问题\n")

        for issue in self.issues:
            severity = issue['severity'].ljust(8)
            msg = issue['message']
            print(f"[{severity}] {msg}")


if __name__ == '__main__':
    import sys

    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <log_file>")
        sys.exit(1)

    parser = CANopenLogParser(sys.argv[1])
    parser.parse()

    detector = CANopenIssueDetector(parser)
    detector.detect_all()
    detector.print_report()

#!/usr/bin/env python3
# canopen_monitor.py - 实时 CANopen 监控

import curses
import time
from threading import Thread
from queue import Queue
import subprocess
import re

class CANopenMonitor:
    def __init__(self, interface):
        self.interface = interface
        self.frame_queue = Queue()
        self.running = False
        self.stats = {
            'total_frames': 0,
            'sdo_requests': 0,
            'sdo_responses': 0,
            'pdo_tx': 0,
            'pdo_rx': 0,
            'heartbeat': 0,
            'errors': 0,
            'nodes': {}
        }

    def start_capture(self):
        """启动 CAN 帧捕获"""
        self.running = True

        def capture_thread():
            process = subprocess.Popen(
                ['candump', self.interface],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )

            while self.running:
                line = process.stdout.readline()
                if line:
                    self.frame_queue.put(line.strip())

            process.terminate()

        Thread(target=capture_thread, daemon=True).start()

    def update_stats(self, frame_str):
        """更新统计信息"""
        match = re.match(r'.*\s+([0-9A-F]+)#([0-9A-Fa-f]*)', frame_str)
        if not match:
            return

        can_id = int(match.group(1), 16)
        node_id = can_id & 0x7F

        self.stats['total_frames'] += 1

        # 消息类型统计
        if (can_id & 0x7F0) == 0x600:
            self.stats['sdo_requests'] += 1
        elif (can_id & 0x7F0) == 0x580:
            self.stats['sdo_responses'] += 1
        elif 0x180 <= can_id <= 0x1FF:
            self.stats['pdo_tx'] += 1
        elif 0x200 <= can_id <= 0x27F:
            self.stats['pdo_rx'] += 1
        elif (can_id & 0x780) == 0x700:
            self.stats['heartbeat'] += 1
            if node_id not in self.stats['nodes']:
                self.stats['nodes'][node_id] = {'last_seen': time.time(), 'count': 0}
            self.stats['nodes'][node_id]['last_seen'] = time.time()
            self.stats['nodes'][node_id]['count'] += 1

    def run(self, stdscr):
        """运行监控界面"""
        self.start_capture()

        curses.curs_set(0)
        stdscr.nodelay(1)

        while self.running:
            # 处理新帧
            while not self.frame_queue.empty():
                frame = self.frame_queue.get()
                self.update_stats(frame)

            # 检查按键
            key = stdscr.getch()
            if key == ord('q'):
                break

            # 更新显示
            self.render(stdscr)

            stdscr.refresh()
            time.sleep(0.1)

        self.running = False

    def render(self, stdscr):
        """渲染监控界面"""
        stdscr.clear()

        h, w = stdscr.getmaxyx()

        # 标题
        title = f"CANopen 实时监控 - {self.interface}"
        stdscr.addstr(0, (w - len(title)) // 2, title, curses.A_BOLD)

        # 统计信息
        line = 2
        stdscr.addstr(line, 2, f"总帧数: {self.stats['total_frames']}")
        stdscr.addstr(line+1, 2, f"SDO 请求: {self.stats['sdo_requests']}")
        stdscr.addstr(line+2, 2, f"SDO 响应: {self.stats['sdo_responses']}")
        stdscr.addstr(line+3, 2, f"TPDO: {self.stats['pdo_tx']}")
        stdscr.addstr(line+4, 2, f"RPDO: {self.stats['pdo_rx']}")
        stdscr.addstr(line+5, 2, f"心跳: {self.stats['heartbeat']}")

        # 节点状态
        line = 8
        stdscr.addstr(line, 2, "活动节点:", curses.A_BOLD)
        line += 1

        for node_id, info in self.stats['nodes'].items():
            elapsed = time.time() - info['last_seen']
            status = "OK" if elapsed < 3.0 else "TIMEOUT"
            stdscr.addstr(line, 4, f"节点 {node_id}: {info['count']} 帧 [{status}]")
            line += 1

        # 帧率
        stdscr.addstr(h-2, 2, f"按 'q' 退出")


if __name__ == '__main__':
    import sys

    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <interface>")
        sys.exit(1)

    interface = sys.argv[1]

    try:
        monitor = CANopenMonitor(interface)
        curses.wrapper(monitor.run)
    except KeyboardInterrupt:
        pass