ATP 行車紀錄分析系統 - 完整技術規格書 v2.0

版本： 2.0 (Python 重構版)

文件日期： 2025-10-27

原始系統版本： 1.0 (Java)

開發單位：

撰寫者： Lawliet

📑 文件目錄

系統概述
系統架構設計
解碼邏輯詳細規格
資料結構定義
API 設計規格
測試策略
Java → Python 對照指南
部署與維護

1. 系統概述

1.1 專案背景

系統名稱： ATP (Automatic Train Protection) 行車紀錄分析系統

開發目的：

解析 ATP 系統產生的行車紀錄資料（RU 檔案、MMI 檔案）
視覺化列車運行軌跡、速度曲線、事件記錄
提供組員考核、安全分析、異常事件偵測功能
產出符合規範的報表

原始系統：

開發時間：2004-2012 年
開發廠商：MiTAC Inc（神通電腦）
技術棧：Java + Swing + JDBC
程式碼規模：約 50,000 行，433 個檔案

重構目標：

保留所有已驗證的解碼邏輯（100% 對照移植）
使用 Python 重寫，提升開發效率與維護性
採用現代化 Web 介面取代 Swing
利用 Pandas 加速資料處理
程式碼量預計減少 90%

1.2 系統範圍

輸入資料：

RU 檔案（.RU 格式）
- Recording Unit（記錄單元）產生的二進位檔案
- 包含：ATP 封包、BTM 電報、VDX 訊號
- 檔案大小：通常 1-50 MB
MMI 檔案（.MMI 格式）
- Man-Machine Interface（人機介面）記錄
- 包含：司機員操作、系統狀態、警告訊息
- 檔案大小：通常 0.5-10 MB

處理功能：

資料解碼
- 40+ 種 MMI 封包類型解析
- BTM 電報 5 片段重組
- 19 種 ETCS 路側封包解碼
資料分析
- 速度曲線分析
- 車站停靠判斷
- EB/SB 事件統計
- 無端 SB 偵測
- 超速警告分析
視覺化呈現
- 時間模式 / 距離模式雙軸顯示
- 速度曲線、ATP 限速線、目標速度線
- 事件標記（EB/SB/故障）
- 車站與 Balise 位置標示
- 坡度曲線疊加
報表產出
- 組員考核報表
- 異常事件清單
- 統計摘要
- 匯出格式：CSV / Excel / PDF

輸出資料：

結構化資料（資料庫）
視覺化圖表（互動式網頁）
報表檔案（CSV/Excel/PDF）

1.3 系統限制與假設

限制：

僅支援台鐵 ATP 系統的資料格式（Bombardier 規格）
RU/MMI 檔案必須完整且未損壞
解碼邏輯基於 2004-2012 年的系統規格

假設：

ATP 系統產生的資料格式保持穩定
二進位檔案使用 Big-Endian 位元組序
時間戳記採用 YY-MM-DD HH:MM:SS 格式（年份 + 2000）
位置資訊單位為公尺
速度資訊單位為 km/h

相依性：

Python 3.9+
資料庫：PostgreSQL 或 SQLite
瀏覽器：Chrome/Firefox/Edge（支援 WebGL）

2. 系統架構設計

2.1 整體架構

┌─────────────────────────────────────────────────────────────┐
│                        使用者介面層                          │
│   (Web UI - Streamlit / FastAPI + React)                   │
├─────────────────────────────────────────────────────────────┤
│                        應用邏輯層                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  檔案管理    │  │  任務管理    │  │  報表產出    │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                        視覺化層                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  速度曲線    │  │  事件圖表    │  │  統計儀表板  │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                        業務邏輯層                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  資料分析    │  │  事件偵測    │  │  統計計算    │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                        資料存取層                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  資料模型    │  │  資料庫DAO   │  │  快取管理    │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                        解碼引擎層                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  RU 解碼器   │  │  MMI 解碼器  │  │  BTM 解碼器  │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                        檔案處理層                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  檔案讀取    │  │  格式驗證    │  │  位元組解析  │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
└─────────────────────────────────────────────────────────────┘

2.2 模組設計

模組 1：檔案處理層 (file_handler)

職責：

讀取 RU/MMI 二進位檔案
驗證檔案格式與完整性
提供位元組流存取介面

主要類別：

# file_handler/ru_file_[reader.py](http://reader.py)
class RUFileReader:
    """RU 檔案讀取器"""
    def read(self, file_path: str) -> bytes
    def validate(self, data: bytes) -> bool
    def get_file_info(self, file_path: str) -> dict

對應 Java 原始碼：

decoder_re/[DataFeeder.java](http://DataFeeder.java)
decoder_re/[FileRead.java](http://FileRead.java)

模組 2：解碼引擎層 (decoder)

職責：

解析二進位資料為結構化資訊
處理 40+ 種 MMI 封包類型
BTM 5 片段重組
ETCS 路側封包解碼

主要類別請見完整規格書第 3 章

對應 Java 原始碼：

decode_re/[Decoder.java](http://Decoder.java)
decode_re/[DecodeATP.java](http://DecodeATP.java)
decoder_re/[HeadDecoder.java](http://HeadDecoder.java)
decoder_re/[BTMDecoder.java](http://BTMDecoder.java)
decoder_re/waySidePacket/*.java（19 個檔案）

3. 解碼邏輯詳細規格

3.1 RU 檔案格式

3.1.1 檔案結構

RU 檔案 (.MMI)
├─ 記錄集合 (Records)
│  ├─ 記錄 1
│  │  ├─ 標頭 (15 bytes)
│  │  └─ 資料 (變長)
│  ├─ 記錄 2
│  │  └─ ...
│  └─ 記錄 N

3.1.2 記錄標頭格式 (15 bytes)

位元組	欄位	型別	說明	Java 對照
0	年份	uint8	YY (需加 2000)	`data[pos]`
1	月份	uint8	MM (1-12)	`data[pos+1]`
2	日期	uint8	DD (1-31)	`data[pos+2]`
3	小時	uint8	HH (0-23)	`data[pos+3]`
4	分鐘	uint8	mm (0-59)	`data[pos+4]`
5	秒數	uint8	ss (0-59)	`data[pos+5]`
6-7	速度	uint16	單位: 0.1 km/h	`DecodeATP.getIntFromByte2()`
8-11	位置	uint32	單位: 公尺	`DecodeATP.getIntFromByte4()`
12-13	封包長度	uint16	含標頭的總長度	`data[pos+12] << 8
14	封包類型	uint8	0x50=ATP, 0xA0=MMI	`data[pos+14]`

Python 實作範例:

@dataclass
class RecordHeader:
    timestamp: datetime
    speed: float  # km/h
    position: int  # meters
    packet_length: int
    packet_type: int
    
    @classmethod
    def parse(cls, data: bytes) -> 'RecordHeader':
        year = data[0] + 2000
        month = data[1]
        day = data[2]
        hour = data[3]
        minute = data[4]
        second = data[5]
        
        timestamp = datetime(year, month, day, hour, minute, second)
        speed = struct.unpack('>H', data[6:8])[0] / 10.0
        position = struct.unpack('>I', data[8:12])[0]
        packet_length = struct.unpack('>H', data[12:14])[0]
        packet_type = data[14]
        
        return cls(timestamp, speed, position, packet_length, packet_type)

Java 對照:

decode_re/[DecodeATP.java](http://DecodeATP.java): getIntFromByte2(), getIntFromByte4()
decoder_re/[HeadDecoder.java](http://HeadDecoder.java): decodeHead()

3.2 MMI 封包解碼規格

3.2.1 封包類型定義

類型碼	封包名稱	優先級	說明
1	MMI_DYNAMIC	⭐⭐⭐	動態資料(速度/位置/時間)
2	MMI_STATUS	⭐⭐⭐	系統狀態
3	MMI_TRACK_DESCRIPTION	⭐⭐	軌道描述
4	MMI_TRACK_DATA	⭐	軌道資料
5	MMI_TRACK_AHEAD_FREE	⭐	前方軌道淨空
6	MMI_DRIVER_REQUEST	⭐⭐	司機員請求
7	MMI_DRIVER_ACTION	⭐⭐	司機員動作
8	MMI_DRIVER_MESSAGE	⭐⭐⭐	司機員訊息
9	MMI_FAILURE_REPORT_ATP	⭐⭐⭐	ATP 故障報告
10	MMI_FAILURE_REPORT_BTM	⭐⭐	BTM 故障報告
11	MMI_SET_TIME	⭐	時間設定
12	MMI_ECHO	⭐	Echo 封包
19	MMI_ICON	⭐	圖示顯示
20	MMI_GEOLOCATION	⭐⭐	地理位置
25	MMI_NTC_STATUS	⭐	NTC 狀態

3.2.2 MMI_DYNAMIC (封包 1) - 最重要的封包

資料長度: 15 bytes (固定)

資料格式:

位元組	欄位	型別	單位	說明
0-1	V_TRAIN	uint16	0.1 km/h	列車速度
2-5	L_TRAIN_POS	int32	cm	列車位置
6-9	L_TRAIN_POS_FRONT	int32	cm	列車前端位置
10-13	T_TRAIN	uint32	ms	列車時間 (UTC)
14	M_MODE	uint8	-	系統模式

模式定義 (M_MODE):

class ATPMode(Enum):
    FULL_SUPERVISION = 0      # FS - 完全監控
    ON_SIGHT = 1              # OS - 目視行駛
    STAFF_RESPONSIBLE = 2     # SR - 司機員責任
    SHUNTING = 3              # SH - 調車
    UNFITTED = 4              # UN - 未裝備
    STANDBY = 5               # SB - 待命
    TRIP = 6                  # TR - 跳脫
    POST_TRIP = 7             # PT - 跳脫後
    SYSTEM_FAILURE = 8        # SF - 系統故障
    ISOLATION = 9             # IS - 隔離
    NON_LEADING = 10          # NL - 非主控
    LIMITED_SUPERVISION = 11  # LS - 有限監控

Python 實作:

@dataclass
class MMIDynamic:
    v_train: float          # km/h
    l_train_pos: int        # cm
    l_train_pos_front: int  # cm
    t_train: int            # ms (UTC)
    m_mode: ATPMode
    
    @classmethod
    def decode(cls, data: bytes) -> 'MMIDynamic':
        v_train = struct.unpack('>H', data[0:2])[0] / 10.0
        l_train_pos = struct.unpack('>i', data[2:6])[0]
        l_train_pos_front = struct.unpack('>i', data[6:10])[0]
        t_train = struct.unpack('>I', data[10:14])[0]
        m_mode = ATPMode(data[14])
        
        return cls(v_train, l_train_pos, l_train_pos_front, t_train, m_mode)

Java 對照:

decode_re/[PacketMMI.java](http://PacketMMI.java): setPacketOne() 方法
變數名稱對照:
- Java: v_train, l_train_location, l_train_location_front
- Python: 保持相同命名

3.2.3 MMI_STATUS (封包 2)

資料長度: 可變 (最少 13 bytes)

核心欄位:

位元組	欄位	型別	說明
0	M_LEVEL	uint8	ETCS Level (0/1/2/3)
1	M_MODE	uint8	ATP 模式
2-3	V_PERMITTED	uint16	允許速度 (0.1 km/h)
4-5	V_TARGET	uint16	目標速度 (0.1 km/h)
6-7	V_RELEASE	uint16	釋放速度 (0.1 km/h)
8-11	D_TARGET	int32	目標距離 (cm)
12	Q_TEXT_CLASS	uint8	文字類別

Python 實作:

@dataclass
class MMIStatus:
    m_level: int
    m_mode: ATPMode
    v_permitted: float  # km/h
    v_target: float     # km/h
    v_release: float    # km/h
    d_target: int       # cm
    q_text_class: int
    
    @classmethod
    def decode(cls, data: bytes) -> 'MMIStatus':
        return cls(
            m_level=data[0],
            m_mode=ATPMode(data[1]),
            v_permitted=struct.unpack('>H', data[2:4])[0] / 10.0,
            v_target=struct.unpack('>H', data[4:6])[0] / 10.0,
            v_release=struct.unpack('>H', data[6:8])[0] / 10.0,
            d_target=struct.unpack('>i', data[8:12])[0],
            q_text_class=data[12]
        )

3.2.4 MMI_DRIVER_MESSAGE (封包 8)

用途: 記錄系統訊息、警告、錯誤

資料格式:

位元組	欄位	型別	說明
0	Q_TEXT	uint8	訊息ID
1	Q_TEXT_CLASS	uint8	訊息類別
2	Q_TEXT_CRITERIA	uint8	顯示條件
3-N	TEXT	string	UTF-8 文字 (可選)

訊息類別 (Q_TEXT_CLASS):

class MessageClass(Enum):
    AUXILIARY_INFO = 0      # 輔助資訊
    IMPORTANT_INFO = 1      # 重要資訊  
    LEVEL_1_WARNING = 2     # Level 1 警告
    LEVEL_2_WARNING = 3     # Level 2 警告

常見訊息 ID (Q_TEXT):

256: "EB 作用" (Emergency Brake)
257: "SB 作用" (Service Brake)
258: "超速警告"
259: "ATP 故障"
260: "BTM 故障"

Python 實作:

@dataclass
class MMIDriverMessage:
    q_text: int
    q_text_class: MessageClass
    q_text_criteria: int
    text: str = ""
    
    @classmethod
    def decode(cls, data: bytes) -> 'MMIDriverMessage':
        q_text = data[0]
        q_text_class = MessageClass(data[1])
        q_text_criteria = data[2]
        text = data[3:].decode('utf-8', errors='ignore').rstrip('\x00')
        
        return cls(q_text, q_text_class, q_text_criteria, text)

3.3 BTM 電報解碼規格

3.3.1 BTM 5 片段重組機制

背景:

BTM 電報最大 1023 bits,透過無線傳輸可能分散在 5 個片段
必須收集完整 5 個片段才能重組完整電報

片段格式:

位元組	欄位	說明
0	FRAGMENT_ID	片段編號 (0-4)
1-N	DATA	片段資料

重組演算法:

class BTMReassembler:
    def __init__(self):
        self.fragments = {}  # {telegram_id: {fragment_id: data}}
        
    def add_fragment(self, telegram_id: int, fragment_id: int, data: bytes) -> Optional[bytes]:
        """
        加入片段,當收集完 5 個片段時返回完整電報
        """
        if telegram_id not in self.fragments:
            self.fragments[telegram_id] = {}
        
        self.fragments[telegram_id][fragment_id] = data
        
        # 檢查是否收集完整
        if len(self.fragments[telegram_id]) == 5:
            # 按照片段 ID 順序組合
            complete_telegram = b''.join(
                self.fragments[telegram_id][i] for i in range(5)
            )
            
            # 清除已重組的片段
            del self.fragments[telegram_id]
            
            return complete_telegram
        
        return None

Java 對照:

decoder_re/[BTMDecoder.java](http://BTMDecoder.java): collectFragments() 方法
使用 HashMap<Integer, List<byte[]>> 儲存片段

3.3.2 ETCS 路側封包格式

Balise 電報結構:

電報 (Telegram)
├─ 電報標頭 (10 bytes)
├─ 封包 1 (P3 - Movement Authority)
├─ 封包 2 (P27 - Speed Profile)  
├─ 封包 3 (P44 - Data Used By Applications)
└─ ...

電報標頭格式:

位元	長度	欄位	說明
0-9	10 bits	Q_UPDOWN	方向 (上行/下行)
10-23	14 bits	M_VERSION	ETCS 版本
24-33	10 bits	Q_MEDIA	媒體類型
34-47	14 bits	N_PIG	Balise 群組編號
48-50	3 bits	N_TOTAL	總 Balise 數
51-53	3 bits	M_DUP	重複標記
54-55	2 bits	M_MCOUNT	訊息計數
56-79	24 bits	NID_C	國家代碼
80-93	14 bits	NID_BG	Balise 群組ID

Python 實作:

@dataclass
class TelegramHeader:
    q_updown: int      # 0=上行, 1=下行
    m_version: int     # ETCS 版本
    q_media: int       # 媒體類型
    n_pig: int         # Balise 編號
    n_total: int       # 群組總數
    m_dup: int         # 重複標記
    m_mcount: int      # 訊息計數
    nid_c: int         # 國家代碼
    nid_bg: int        # Balise 群組 ID
    
    @classmethod
    def parse(cls, bits: str) -> 'TelegramHeader':
        """
        從 bit string 解析標頭
        bits: '01001110...' (94 bits)
        """
        return cls(
            q_updown=int(bits[0:10], 2),
            m_version=int(bits[10:24], 2),
            q_media=int(bits[24:34], 2),
            n_pig=int(bits[34:48], 2),
            n_total=int(bits[48:51], 2),
            m_dup=int(bits[51:54], 2),
            m_mcount=int(bits[54:56], 2),
            nid_c=int(bits[56:80], 2),
            nid_bg=int(bits[80:94], 2)
        )

3.3.3 重要封包類型

P3 - Movement Authority (行車權限)

位元	長度	欄位	說明
0-7	8	NID_PACKET	封包ID = 3
8-20	13	Q_DIR	方向
21-35	15	L_PACKET	封包長度
36-50	15	Q_SCALE	比例尺
51-65	15	V_LOA	最低允許速度
66-80	15	T_LOA	時間限制
81-95	15	N_ITER	區段數量
96+	var	SECTIONS	速度區段列表

P27 - Speed Profile (速度限制)

位元	長度	欄位	說明
0-7	8	NID_PACKET	封包ID = 27
8-20	13	Q_DIR	方向
21-35	15	L_PACKET	封包長度
36-50	15	Q_SCALE	比例尺
51-57	7	D_STATIC	靜態速度區段長度
58-64	7	V_STATIC	靜態速度值
65-71	7	Q_FRONT	列車前端標記
72-76	5	N_ITER	速度區段數量
77+	var	PROFILES	速度曲線列表

Python 實作:

@dataclass
class P3_MovementAuthority:
    nid_packet: int = 3
    q_dir: int = 0
    l_packet: int = 0
    q_scale: int = 0
    v_loa: int = 0
    t_loa: int = 0
    sections: List['MASection'] = field(default_factory=list)
    
    @classmethod
    def parse(cls, bits: str) -> 'P3_MovementAuthority':
        # 實作封包解析邏輯
        pass

@dataclass  
class P27_SpeedProfile:
    nid_packet: int = 27
    q_dir: int = 0
    l_packet: int = 0
    q_scale: int = 0
    d_static: int = 0
    v_static: int = 0
    q_front: int = 0
    profiles: List['SpeedSection'] = field(default_factory=list)
    
    @classmethod
    def parse(cls, bits: str) -> 'P27_SpeedProfile':
        # 實作封包解析邏輯
        pass

Java 對照:

decoder_re/waySidePacket/P3_[MovementAuthority.java](http://MovementAuthority.java)
decoder_re/waySidePacket/P27_[SpeedProfile.java](http://SpeedProfile.java)
共 19 個封包類別 (P3, P5, P12, P15, P21, P27, P41, P42, P44, P45, P58, P63, P64, P65, P66, P67, P68, P72, P80)

3.4 VDX 數位訊號解碼

VDX (Digital I/O Signals): 列車數位輸入輸出訊號

訊號類型:

位元	訊號名稱	說明
0	EB_CMD	緊急煞車指令
1	SB_CMD	常用煞車指令
2	WARNING	警告訊號
3	ISOLATION	隔離狀態
4	DOOR_CLOSED	車門關閉
5	TRACTION_CUT	牽引切斷
6-7	RESERVED	保留

Python 實作:

@dataclass
class VDXSignals:
    eb_cmd: bool       # 緊急煞車指令
    sb_cmd: bool       # 常用煞車指令
    warning: bool      # 警告
    isolation: bool    # 隔離
    door_closed: bool  # 車門關閉
    traction_cut: bool # 牽引切斷
    
    @classmethod
    def decode(cls, byte_value: int) -> 'VDXSignals':
        return cls(
            eb_cmd=bool(byte_value & 0x01),
            sb_cmd=bool(byte_value & 0x02),
            warning=bool(byte_value & 0x04),
            isolation=bool(byte_value & 0x08),
            door_closed=bool(byte_value & 0x10),
            traction_cut=bool(byte_value & 0x20)
        )

4. 資料結構定義

4.1 核心資料模型

4.1.1 ATPMission (任務)

@dataclass
class ATPMission:
    """代表一次完整的行車任務"""
    
    # 基本資訊
    id: str                    # 任務 ID (UUID)
    ru_filename: str           # RU 檔案名稱
    mmi_filename: str          # MMI 檔案名稱
    upload_time: datetime      # 上傳時間
    
    # 行車資訊
    train_number: str          # 車次
    driver_name: str           # 司機員姓名
    start_station: str         # 起站
    end_station: str           # 迄站
    start_time: datetime       # 發車時間
    end_time: datetime         # 到站時間
    
    # 資料集合
    records: List[Record]      # 所有記錄
    events: List[Event]        # 事件列表
    stations: List[Station]    # 車站列表
    balises: List[Balise]      # Balise 列表
    
    # 統計資訊
    total_distance: int        # 總距離 (m)
    max_speed: float           # 最高速度 (km/h)
    avg_speed: float           # 平均速度 (km/h)
    eb_count: int              # EB 次數
    sb_count: int              # SB 次數
    warning_count: int         # 警告次數
    
    # 狀態
    status: str                # 處理狀態
    error_message: str = ""    # 錯誤訊息

4.1.2 Record (記錄)

@dataclass
class Record:
    """單筆記錄"""
    
    # 標頭資訊
    timestamp: datetime        # 時間戳記
    speed: float               # 速度 (km/h)
    position: int              # 位置 (m)
    packet_type: int           # 封包類型
    
    # 封包資料
    packet_data: Union[
        MMIDynamic,
        MMIStatus,
        MMIDriverMessage,
        BTMTelegram,
        VDXSignals,
        bytes  # 未解析的原始資料
    ]
    
    # 衍生資訊
    acceleration: float = 0.0  # 加速度 (m/s²)
    gradient: float = 0.0      # 坡度 (‰)
    mode: ATPMode = ATPMode.STANDBY  # ATP 模式

4.1.3 Event (事件)

class EventType(Enum):
    """事件類型"""
    EB = "EB"                  # 緊急煞車
    SB = "SB"                  # 常用煞車
    WARNING = "WARNING"        # 警告
    OVERSPEED = "OVERSPEED"    # 超速
    ATP_FAULT = "ATP_FAULT"    # ATP 故障
    BTM_FAULT = "BTM_FAULT"    # BTM 故障
    MODE_CHANGE = "MODE_CHANGE" # 模式切換
    STATION_STOP = "STATION_STOP" # 車站停車
    
@dataclass
class Event:
    """事件記錄"""
    
    id: str                    # 事件 ID
    mission_id: str            # 所屬任務
    event_type: EventType      # 事件類型
    timestamp: datetime        # 發生時間
    position: int              # 發生位置 (m)
    speed: float               # 當時速度 (km/h)
    
    # 事件詳細資訊
    description: str           # 事件描述
    severity: int              # 嚴重程度 (1-5)
    related_records: List[int] # 相關記錄 ID
    
    # 分析資訊
    is_valid: bool = True      # 是否有效事件
    is_false_positive: bool = False  # 是否誤判
    note: str = ""             # 備註

4.1.4 Station (車站)

@dataclass
class Station:
    """車站資訊"""
    
    id: str                    # 車站 ID
    name: str                  # 車站名稱
    position: int              # 位置 (m)
    platform_start: int        # 月台起點 (m)
    platform_end: int          # 月台終點 (m)
    
    # 停車資訊
    arrival_time: Optional[datetime] = None     # 到達時間
    departure_time: Optional[datetime] = None   # 發車時間
    stop_position: Optional[int] = None         # 停車位置 (m)
    is_in_platform: bool = False                # 是否停在月台內
    
    # 分析資訊
    overrun_distance: int = 0  # 超過停車點距離 (m)
    stop_accuracy_grade: str = "" # 停車精準度等級

4.1.5 Balise (應答器)

@dataclass  
class Balise:
    """Balise 應答器"""
    
    id: str                    # Balise ID
    nid_bg: int                # Balise 群組 ID
    position: int              # 位置 (m)
    direction: int             # 方向 (0=上行, 1=下行)
    
    # 電報資訊
    telegram_count: int = 0    # 收到電報數
    last_telegram_time: Optional[datetime] = None
    
    # 封包資料
    movement_authority: Optional[P3_MovementAuthority] = None
    speed_profile: Optional[P27_SpeedProfile] = None
    gradient_profile: Optional[P21_GradientProfile] = None
    
    # 狀態
    is_valid: bool = True      # 是否有效
    error_count: int = 0       # 錯誤次數

4.2 資料庫 Schema

4.2.1 PostgreSQL Schema

-- 任務表
CREATE TABLE missions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    ru_filename VARCHAR(255) NOT NULL,
    mmi_filename VARCHAR(255) NOT NULL,
    upload_time TIMESTAMP NOT NULL DEFAULT NOW(),
    
    train_number VARCHAR(50),
    driver_name VARCHAR(100),
    start_station VARCHAR(100),
    end_station VARCHAR(100),
    start_time TIMESTAMP,
    end_time TIMESTAMP,
    
    total_distance INTEGER,
    max_speed FLOAT,
    avg_speed FLOAT,
    eb_count INTEGER DEFAULT 0,
    sb_count INTEGER DEFAULT 0,
    warning_count INTEGER DEFAULT 0,
    
    status VARCHAR(50) DEFAULT 'pending',
    error_message TEXT,
    
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_missions_train_number ON missions(train_number);
CREATE INDEX idx_missions_driver_name ON missions(driver_name);
CREATE INDEX idx_missions_upload_time ON missions(upload_time);

-- 記錄表
CREATE TABLE records (
    id BIGSERIAL PRIMARY KEY,
    mission_id UUID REFERENCES missions(id) ON DELETE CASCADE,
    
    timestamp TIMESTAMP NOT NULL,
    speed FLOAT NOT NULL,
    position INTEGER NOT NULL,
    packet_type SMALLINT NOT NULL,
    
    -- 封包資料 (JSON 格式)
    packet_data JSONB,
    
    acceleration FLOAT DEFAULT 0.0,
    gradient FLOAT DEFAULT 0.0,
    mode VARCHAR(50),
    
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_records_mission_id ON records(mission_id);
CREATE INDEX idx_records_timestamp ON records(timestamp);
CREATE INDEX idx_records_position ON records(position);
CREATE INDEX idx_records_packet_data ON records USING gin(packet_data);

-- 事件表
CREATE TABLE events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    mission_id UUID REFERENCES missions(id) ON DELETE CASCADE,
    
    event_type VARCHAR(50) NOT NULL,
    timestamp TIMESTAMP NOT NULL,
    position INTEGER NOT NULL,
    speed FLOAT NOT NULL,
    
    description TEXT,
    severity INTEGER CHECK (severity BETWEEN 1 AND 5),
    related_records INTEGER[],
    
    is_valid BOOLEAN DEFAULT TRUE,
    is_false_positive BOOLEAN DEFAULT FALSE,
    note TEXT,
    
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_events_mission_id ON events(mission_id);
CREATE INDEX idx_events_type ON events(event_type);
CREATE INDEX idx_events_timestamp ON events(timestamp);

-- 車站表  
CREATE TABLE stations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    mission_id UUID REFERENCES missions(id) ON DELETE CASCADE,
    
    name VARCHAR(100) NOT NULL,
    position INTEGER NOT NULL,
    platform_start INTEGER NOT NULL,
    platform_end INTEGER NOT NULL,
    
    arrival_time TIMESTAMP,
    departure_time TIMESTAMP,
    stop_position INTEGER,
    is_in_platform BOOLEAN DEFAULT FALSE,
    
    overrun_distance INTEGER DEFAULT 0,
    stop_accuracy_grade VARCHAR(10),
    
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_stations_mission_id ON stations(mission_id);
CREATE INDEX idx_stations_name ON stations(name);

-- Balise 表
CREATE TABLE balises (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    mission_id UUID REFERENCES missions(id) ON DELETE CASCADE,
    
    nid_bg INTEGER NOT NULL,
    position INTEGER NOT NULL,
    direction SMALLINT NOT NULL,
    
    telegram_count INTEGER DEFAULT 0,
    last_telegram_time TIMESTAMP,
    
    -- 封包資料 (JSON 格式)
    movement_authority JSONB,
    speed_profile JSONB,
    gradient_profile JSONB,
    
    is_valid BOOLEAN DEFAULT TRUE,
    error_count INTEGER DEFAULT 0,
    
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_balises_mission_id ON balises(mission_id);
CREATE INDEX idx_balises_nid_bg ON balises(nid_bg);
CREATE INDEX idx_balises_position ON balises(position);

-- 組員表 (用於考核)
CREATE TABLE crew_members (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(100) NOT NULL UNIQUE,
    employee_id VARCHAR(50) NOT NULL UNIQUE,
    department VARCHAR(100),
    position VARCHAR(50),
    
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- 組員考核記錄表
CREATE TABLE crew_evaluations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    mission_id UUID REFERENCES missions(id) ON DELETE CASCADE,
    crew_member_id UUID REFERENCES crew_members(id) ON DELETE CASCADE,
    
    evaluation_date DATE NOT NULL,
    
    -- 考核項目
    eb_count INTEGER DEFAULT 0,
    sb_count INTEGER DEFAULT 0,
    false_sb_count INTEGER DEFAULT 0,
    overspeed_count INTEGER DEFAULT 0,
    stop_accuracy_score FLOAT,
    
    -- 評分
    total_score FLOAT,
    grade VARCHAR(10),
    note TEXT,
    
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_crew_evaluations_mission_id ON crew_evaluations(mission_id);
CREATE INDEX idx_crew_evaluations_member_id ON crew_evaluations(crew_member_id);
CREATE INDEX idx_crew_evaluations_date ON crew_evaluations(evaluation_date);

5. API 設計規格

5.1 REST API 端點

5.1.1 檔案管理 API

上傳檔案

POST /api/v1/files/upload
Content-Type: multipart/form-data

Request Body:
{
    "ru_file": <binary>,
    "mmi_file": <binary>,
    "train_number": "1234",
    "driver_name": "張三"
}

Response:
{
    "mission_id": "uuid",
    "status": "processing",
    "message": "檔案上傳成功,開始處理"
}

查詢處理狀態

GET /api/v1/missions/{mission_id}/status

Response:
{
    "mission_id": "uuid",
    "status": "completed",  // pending, processing, completed, failed
    "progress": 100,
    "message": "處理完成",
    "created_at": "2025-10-27T10:00:00Z",
    "completed_at": "2025-10-27T10:05:00Z"
}

5.1.2 資料查詢 API

查詢任務列表

GET /api/v1/missions?
    driver_name=張三&
    start_date=2025-10-01&
    end_date=2025-10-27&
    page=1&
    page_size=20

Response:
{
    "total": 150,
    "page": 1,
    "page_size": 20,
    "missions": [
        {
            "id": "uuid",
            "train_number": "1234",
            "driver_name": "張三",
            "start_time": "2025-10-27T08:00:00Z",
            "end_time": "2025-10-27T10:00:00Z",
            "total_distance": 50000,
            "max_speed": 130.5,
            "eb_count": 0,
            "sb_count": 2
        }
    ]
}

查詢記錄資料

GET /api/v1/missions/{mission_id}/records?
    start_time=2025-10-27T08:00:00Z&
    end_time=2025-10-27T09:00:00Z&
    packet_type=1

Response:
{
    "mission_id": "uuid",
    "total_records": 3600,
    "records": [
        {
            "timestamp": "2025-10-27T08:00:00Z",
            "speed": 80.5,
            "position": 1000,
            "packet_type": 1,
            "packet_data": {
                "v_train": 80.5,
                "l_train_pos": 100000,
                "m_mode": "FULL_SUPERVISION"
            }
        }
    ]
}

查詢事件列表

GET /api/v1/missions/{mission_id}/events?
    event_type=SB&
    severity_min=3

Response:
{
    "mission_id": "uuid",
    "events": [
        {
            "id": "uuid",
            "event_type": "SB",
            "timestamp": "2025-10-27T08:15:30Z",
            "position": 15000,
            "speed": 95.0,
            "description": "超速觸發 SB",
            "severity": 3,
            "is_false_positive": false
        }
    ]
}

5.1.3 分析 API

速度曲線資料

GET /api/v1/missions/{mission_id}/speed-curve?
    mode=time  // time | distance

Response:
{
    "mission_id": "uuid",
    "mode": "time",
    "data": {
        "timestamps": ["2025-10-27T08:00:00Z", ...],
        "speeds": [0, 10.5, 20.3, ...],
        "positions": [0, 100, 250, ...],
        "atp_limits": [120, 120, 110, ...],
        "target_speeds": [80, 80, 70, ...]
    }
}

停車精準度分析

GET /api/v1/missions/{mission_id}/stop-accuracy

Response:
{
    "mission_id": "uuid",
    "stations": [
        {
            "name": "台北",
            "platform_start": 1000,
            "platform_end": 1200,
            "stop_position": 1100,
            "is_in_platform": true,
            "overrun_distance": 0,
            "grade": "A"
        }
    ]
}

無端 SB 偵測

GET /api/v1/missions/{mission_id}/false-sb

Response:
{
    "mission_id": "uuid",
    "false_sb_events": [
        {
            "timestamp": "2025-10-27T08:30:00Z",
            "position": 20000,
            "speed": 0.0,
            "reason": "換端時誤觸發",
            "is_confirmed": true
        }
    ]
}

5.1.4 報表 API

產生組員考核報表

POST /api/v1/reports/crew-evaluation
Content-Type: application/json

Request Body:
{
    "crew_member_id": "uuid",
    "start_date": "2025-10-01",
    "end_date": "2025-10-27",
    "format": "pdf"  // csv | excel | pdf
}

Response:
{
    "report_id": "uuid",
    "download_url": "/api/v1/reports/uuid/download",
    "expires_at": "2025-10-28T00:00:00Z"
}

下載報表

GET /api/v1/reports/{report_id}/download

Response:
Content-Type: application/pdf
Content-Disposition: attachment; filename="crew_evaluation_20251027.pdf"
<binary data>

5.2 WebSocket API (即時資料)

// 連接 WebSocket
ws = new WebSocket('ws://[localhost:8000/ws/missions/{mission_id}](http://localhost:8000/ws/missions/{mission_id})')

// 訂閱即時速度資料
ws.send(JSON.stringify({
    type: 'subscribe',
    channel: 'speed_updates'
}))

// 接收即時資料
ws.onmessage = (event) => {
    const data = JSON.parse([event.data](http://event.data))
    // {
    //     type: 'speed_update',
    //     timestamp: '2025-10-27T08:00:00Z',
    //     speed: 80.5,
    //     position: 1000
    // }
}

6. 測試策略

6.1 單元測試

6.1.1 解碼器測試

class TestRUDecoder(unittest.TestCase):
    def setUp(self):
        self.decoder = RUDecoder()
        self.test_data = self.load_test_data('test_ru.mmi')
    
    def test_record_header_parsing(self):
        """測試記錄標頭解析"""
        header = RecordHeader.parse(self.test_data[:15])
        
        self.assertEqual(header.timestamp.year, 2025)
        self.assertAlmostEqual(header.speed, 80.5, delta=0.1)
        self.assertEqual(header.position, 1000)
        self.assertEqual(header.packet_type, 0xA0)
    
    def test_mmi_dynamic_decoding(self):
        """測試 MMI_DYNAMIC 封包解碼"""
        packet = MMIDynamic.decode(self.test_data[15:30])
        
        self.assertAlmostEqual(packet.v_train, 80.5, delta=0.1)
        self.assertEqual(packet.m_mode, ATPMode.FULL_SUPERVISION)
    
    def test_btm_reassembly(self):
        """測試 BTM 5 片段重組"""
        reassembler = BTMReassembler()
        
        # 加入 5 個片段
        for i in range(5):
            fragment = self.load_fragment(i)
            result = reassembler.add_fragment(telegram_id=1, fragment_id=i, data=fragment)
            
            if i < 4:
                self.assertIsNone(result)  # 尚未完成
            else:
                self.assertIsNotNone(result)  # 完成重組
                self.assertEqual(len(result), 128)  # 完整電報長度

6.1.2 資料結構測試

class TestATPMission(unittest.TestCase):
    def test_mission_creation(self):
        """測試任務建立"""
        mission = ATPMission(
            id=str(uuid.uuid4()),
            ru_filename='test.mmi',
            mmi_filename='test.mmi',
            upload_time=[datetime.now](http://datetime.now)(),
            train_number='1234',
            driver_name='測試司機',
            start_station='台北',
            end_station='高雄',
            records=[],
            events=[],
            stations=[],
            balises=[]
        )
        
        self.assertIsNotNone([mission.id](http://mission.id))
        self.assertEqual(mission.train_number, '1234')
    
    def test_event_filtering(self):
        """測試事件篩選"""
        events = [
            Event(..., event_type=EventType.EB, severity=5),
            Event(..., event_type=[EventType.SB](http://EventType.SB), severity=3),
            Event(..., event_type=EventType.WARNING, severity=1)
        ]
        
        high_severity_events = [e for e in events if e.severity >= 3]
        self.assertEqual(len(high_severity_events), 2)

6.2 整合測試

6.2.1 端對端測試

class TestEndToEnd(unittest.TestCase):
    def test_complete_workflow(self):
        """測試完整工作流程"""
        
        # 1. 上傳檔案
        ru_file = open('test_data/sample.mmi', 'rb')
        mmi_file = open('test_data/sample.mmi', 'rb')
        
        response = [self.client.post](http://self.client.post)('/api/v1/files/upload', files={
            'ru_file': ru_file,
            'mmi_file': mmi_file
        })
        
        self.assertEqual(response.status_code, 200)
        mission_id = response.json()['mission_id']
        
        # 2. 等待處理完成
        while True:
            status_response = self.client.get(f'/api/v1/missions/{mission_id}/status')
            status = status_response.json()['status']
            
            if status == 'completed':
                break
            elif status == 'failed':
                [self.fail](http://self.fail)('處理失敗')
            
            time.sleep(1)
        
        # 3. 查詢資料
        records_response = self.client.get(f'/api/v1/missions/{mission_id}/records')
        records = records_response.json()['records']
        
        self.assertGreater(len(records), 0)
        
        # 4. 查詢事件
        events_response = self.client.get(f'/api/v1/missions/{mission_id}/events')
        events = events_response.json()['events']
        
        # 驗證事件資料
        for event in events:
            self.assertIn(event['event_type'], [e.value for e in EventType])

6.3 對照驗證測試

關鍵: 使用 Java 系統的輸出作為測試基準

class TestJavaCompatibility(unittest.TestCase):
    """對照 Java 系統輸出進行驗證"""
    
    def setUp(self):
        # 載入 Java 系統的輸出結果
        [self.java](http://self.java)_output = self.load_java_reference('java_output.json')
    
    def test_decoder_output_matches_java(self):
        """驗證解碼器輸出與 Java 系統一致"""
        
        # 使用相同的測試檔案
        ru_file = 'test_data/sample.mmi'
        
        # Python 解碼
        python_decoder = RUDecoder()
        python_records = python_decoder.decode_file(ru_file)
        
        # 載入 Java 解碼結果
        java_records = [self.java](http://self.java)_output['records']
        
        # 比對記錄數量
        self.assertEqual(len(python_records), len(java_records))
        
        # 逐筆比對
        for i, (py_rec, java_rec) in enumerate(zip(python_records, java_records)):
            with self.subTest(record=i):
                # 比對時間戳記
                self.assertEqual(
                    py_rec.timestamp.isoformat(),
                    java_rec['timestamp']
                )
                
                # 比對速度 (允許 0.1 km/h 誤差)
                self.assertAlmostEqual(
                    py_rec.speed,
                    java_rec['speed'],
                    delta=0.1
                )
                
                # 比對位置 (允許 1 公尺誤差)
                self.assertAlmostEqual(
                    py_rec.position,
                    java_rec['position'],
                    delta=1
                )
    
    def test_event_detection_matches_java(self):
        """驗證事件偵測與 Java 系統一致"""
        
        python_events = self.detect_events('test_data/sample.mmi')
        java_events = [self.java](http://self.java)_output['events']
        
        # 比對事件數量
        self.assertEqual(len(python_events), len(java_events))
        
        # 比對每個事件
        for py_event, java_event in zip(python_events, java_events):
            self.assertEqual(py_event.event_type.value, java_event['type'])
            self.assertEqual(py_event.position, java_event['position'])

6.4 效能測試

class TestPerformance(unittest.TestCase):
    def test_large_file_processing(self):
        """測試大檔案處理效能"""
        
        large_file = 'test_data/large_50mb.mmi'
        
        start_time = time.time()
        decoder = RUDecoder()
        records = decoder.decode_file(large_file)
        end_time = time.time()
        
        processing_time = end_time - start_time
        
        # 期望: 50 MB 檔案在 10 秒內處理完成
        self.assertLess(processing_time, 10.0)
        
        # 期望: 至少產生 10000 筆記錄
        self.assertGreater(len(records), 10000)
    
    def test_concurrent_processing(self):
        """測試並行處理效能"""
        
        files = [f'test_data/test_{i}.mmi' for i in range(10)]
        
        start_time = time.time()
        
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(self.process_file, f) for f in files]
            results = [f.result() for f in futures]
        
        end_time = time.time()
        
        # 期望: 10 個檔案並行處理時間 < 序列處理時間 / 3
        self.assertLess(end_time - start_time, 30.0)

7. Java → Python 對照指南

7.1 類別對照表

Java 類別	Python 類別/模組	說明
`Decoder.java`	`decoder/ru_[decoder.py](http://decoder.py)`	RU 主解碼器
`DecodeATP.java`	`decoder/atp_[decoder.py](http://decoder.py)`	ATP 封包解碼
`PacketMMI.java`	`decoder/mmi_[decoder.py](http://decoder.py)`	MMI 封包解碼
`HeadDecoder.java`	`decoder/header_[parser.py](http://parser.py)`	標頭解析器
`BTMDecoder.java`	`decoder/btm_[decoder.py](http://decoder.py)`	BTM 解碼器
`DataFeeder.java`	`decoder/data_[feeder.py](http://feeder.py)`	資料供給器
`WaySideTelegramPacketDecoder.java`	`decoder/wayside_[decoder.py](http://decoder.py)`	路側電報解碼
`P3_[MovementAuthority.java](http://MovementAuthority.java)`	`models/packets/[p3.py](http://p3.py)`	P3 封包
`ATPMission.java`	`models/[mission.py](http://mission.py)`	任務模型

7.2 關鍵差異

7.2.1 位元組序處理

Java:

// Big-Endian 讀取
int value = (data[0] << 8) | (data[1] & 0xFF);

Python:

# 使用 struct 模組
value = struct.unpack('>H', data[0:2])[0]

7.2.2 集合類別

Java:

// 使用 Vector (同步化,較慢)
Vector<Record> records = new Vector<>();
records.add(record);

Python:

# 使用 List (更快)
records: List[Record] = []
records.append(record)

7.2.3 錯誤處理

Java:

try {
    // 解碼邏輯
} catch (Exception e) {
    // 通常只印出錯誤,不處理
    e.printStackTrace();
}

Python:

try:
    # 解碼邏輯
except struct.error as e:
    logger.error(f"解碼失敗: {e}")
    raise DecoderException(f"封包格式錯誤") from e

7.3 移植步驟

Step 1: 分析 Java 原始碼

# 1. 找出核心類別
grep -r "class.*Decoder" src_decompiled/

# 2. 建立類別依賴圖
javap -v *.class > dependencies.txt

# 3. 識別關鍵方法
grep -r "public.*decode" src_decompiled/

Step 2: 建立對照表

列出所有 Java 類別
規劃對應的 Python 模組
識別共用邏輯

Step 3: 實作核心解碼器

# 從最底層開始實作
# 1. header_[parser.py](http://parser.py) (標頭解析)
# 2. mmi_[decoder.py](http://decoder.py) (MMI 封包)
# 3. btm_[decoder.py](http://decoder.py) (BTM 重組)
# 4. ru_[decoder.py](http://decoder.py) (主解碼器)

Step 4: 對照驗證

# 使用 Java 系統的輸出作為測試基準
java_output = load_java_reference('test_case_1.json')
python_output = python_decoder.decode('test_case_1.mmi')

assert_equal(python_output, java_output)

Step 5: 效能優化

# 使用 cProfile 找出瓶頸
python -m cProfile -o [output.prof](http://output.prof) [main.py](http://main.py)

# 優化熱點函數
# - 使用 struct 代替手動位元組操作
# - 使用 NumPy 加速數值運算
# - 使用 concurrent.futures 平行處理

8. 部署與維護

8.1 系統需求

硬體需求:

CPU: 4 核心以上
RAM: 8 GB 以上
硬碟: 100 GB 以上 (SSD 建議)

軟體需求:

OS: macOS, Linux, Windows
Python: 3.9+
PostgreSQL: 13+
Redis: 6+ (快取)

8.2 安裝部署

Step 1: 環境設定

# 建立虛擬環境
python -m venv venv
source venv/bin/activate  # macOS/Linux
# venv\Scripts\activate  # Windows

# 安裝相依套件
pip install -r requirements.txt

Step 2: 資料庫設定

# 建立資料庫
createdb atp_analysis

# 執行 Schema
psql atp_analysis < schema.sql

# 建立索引
psql atp_analysis < indexes.sql

Step 3: 設定檔

# config.yaml
database:
  host: [localhost](http://localhost)
  port: 5432
  name: atp_analysis
  user: postgres
  password: your_password

redis:
  host: [localhost](http://localhost)
  port: 6379

file_storage:
  base_path: /Users/lawliet/dev-workspace/atp-data
  max_size_mb: 100

logging:
  level: INFO
  file: logs/atp.log

Step 4: 啟動服務

# 啟動 API 伺服器
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

# 啟動背景工作程序
celery -A tasks worker --loglevel=info

# 啟動 Web UI
streamlit run [app.py](http://app.py) --server.port 8501

8.3 監控維護

日誌監控:

# 使用 structlog 進行結構化日誌
import structlog

logger = structlog.get_logger()
[logger.info](http://logger.info)("decoder.start", mission_id=mission_id, file_size=file_size)
logger.error("decoder.failed", mission_id=mission_id, error=str(e))

效能監控:

# 使用 Prometheus 收集指標
from prometheus_client import Counter, Histogram

decode_duration = Histogram('atp_decode_duration_seconds', 'Decode duration')
decode_errors = Counter('atp_decode_errors_total', 'Total decode errors')

@decode_duration.time()
def decode_file(filename):
    try:
        # 解碼邏輯
        pass
    except Exception as e:
        decode_[errors.inc](http://errors.inc)()
        raise

備份策略:

# 每日備份資料庫
pg_dump atp_analysis > backup_$(date +%Y%m%d).sql

# 每週備份上傳檔案
tar -czf uploads_$(date +%Y%m%d).tar.gz /Users/lawliet/dev-workspace/atp-data/

8.4 疑難排解

常見問題:

問題	原因	解決方法
解碼失敗	檔案格式錯誤	檢查檔案完整性
記憶體不足	大檔案處理	增加 RAM 或使用串流處理
處理速度慢	CPU 效能不足	使用多執行緒或更快的 CPU
資料庫連線失敗	PostgreSQL 未啟動	檢查資料庫服務狀態

除錯模式:

# 啟用詳細日誌
export ATP_LOG_LEVEL=DEBUG
python [main.py](http://main.py)

# 使用 pdb 除錯
python -m pdb [main.py](http://main.py)

重要提醒

這份規格書是從 Java 原始系統完整移植到 Python 的指南：

100% 對照原則：所有解碼邏輯必須與 Java 版本完全一致
測試驅動：使用 Java 系統的輸出作為測試基準
保留核心價值：Java 系統經過 10+ 年驗證，不可隨意更改邏輯
現代化改進：在保證邏輯正確的前提下，使用現代工具提升效能

關鍵成功因素：

✅ 詳細閱讀 Java 原始碼
✅ 為每個模組建立對照測試
✅ 使用實際 RU 檔案驗證
✅ 保持與 Java 團隊溝通

下一步行動

立即可做：

設定開發環境（Python 3.11+, PostgreSQL）
Clone Java 原始碼到 reference/ 目錄
準備測試資料（實際 RU 檔案 + Java 輸出）
開始實作核心解碼器

開發順序建議：

階段 1 (2-3週): 核心解碼器
  ├─ PacketHeaderParser
  ├─ MMIDecoder (先實作 MMI_DYNAMIC)
  ├─ BTMDecoder
  └─ 單元測試 + Java 對照驗證

階段 2 (1-2週): 資料模型
  ├─ ATPMission
  ├─ DatabaseManager
  └─ 整合測試

階段 3 (2-3週): 視覺化
  ├─ SpeedCurveDrawer
  ├─ EventDrawer
  └─ 互動式圖表

階段 4 (1-2週): Web UI
  ├─ Streamlit 主介面
  ├─ 檔案上傳功能
  └─ 報表產出

聯絡方式：

開發者：Lawliet
單位：
專案路徑：/Users/lawliet/dev-workspace/atp-analysis-python

文件版本： 2.0

最後更新： 2025-10-27

狀態： 規格書完成，準備開始實作

附錄：快速參考

專有名詞

ATP: Automatic Train Protection (列車自動防護系統)
MMI: Man-Machine Interface (人機介面)
RU: Recording Unit (記錄單元)
BTM: Balise Transmission Module (地面應答器傳輸模組)
EB: Emergency Brake (緊急煞車)
SB: Service Brake (常用煞車)
ETCS: European Train Control System (歐洲列車控制系統)

重要封包

MMI_DYNAMIC (1): 動態資料（速度、位置）⭐
MMI_STATUS (2): 系統狀態 ⭐
MMI_DRIVER_MESSAGE (8): 司機員訊息 ⭐
MMI_FAILURE_REPORT_ATP (9): ATP 故障 ⭐
P3: Movement Authority (行車權限) ⭐
P27: Speed Profile (速度限制) ⭐

關鍵檔案位置

Java 原始碼：

專案檔案：/mnt/project/
本地參考：/Users/lawliet/dev-workspace/atp-analysis-python/reference/

Python 專案：

主目錄：/Users/lawliet/dev-workspace/atp-analysis-python/
測試資料：tests/test_data/
Java 輸出參考：tests/java_reference_outputs/

FilesExpand file tree

ATP 行車紀錄分析系統 - 完整技術規格書 v2 0 29855714413f81b1b980eedb85bea559.md

Latest commit

History

ATP 行車紀錄分析系統 - 完整技術規格書 v2 0 29855714413f81b1b980eedb85bea559.md

File metadata and controls

ATP 行車紀錄分析系統 - 完整技術規格書 v2.0

📑 文件目錄

1. 系統概述

1.1 專案背景

1.2 系統範圍

1.3 系統限制與假設

2. 系統架構設計

2.1 整體架構

2.2 模組設計

模組 1：檔案處理層 (file_handler)

模組 2：解碼引擎層 (decoder)

3. 解碼邏輯詳細規格

3.1 RU 檔案格式

3.1.1 檔案結構

3.1.2 記錄標頭格式 (15 bytes)

3.2 MMI 封包解碼規格

3.2.1 封包類型定義

3.2.2 MMI_DYNAMIC (封包 1) - 最重要的封包

3.2.3 MMI_STATUS (封包 2)

3.2.4 MMI_DRIVER_MESSAGE (封包 8)

3.3 BTM 電報解碼規格

3.3.1 BTM 5 片段重組機制

3.3.2 ETCS 路側封包格式

3.3.3 重要封包類型

3.4 VDX 數位訊號解碼

4. 資料結構定義

4.1 核心資料模型

4.1.1 ATPMission (任務)

4.1.2 Record (記錄)

4.1.3 Event (事件)

4.1.4 Station (車站)

4.1.5 Balise (應答器)

4.2 資料庫 Schema

4.2.1 PostgreSQL Schema

5. API 設計規格

5.1 REST API 端點

5.1.1 檔案管理 API

5.1.2 資料查詢 API

5.1.3 分析 API

5.1.4 報表 API

5.2 WebSocket API (即時資料)

6. 測試策略

6.1 單元測試

6.1.1 解碼器測試

6.1.2 資料結構測試

6.2 整合測試

6.2.1 端對端測試

6.3 對照驗證測試

6.4 效能測試

7. Java → Python 對照指南

7.1 類別對照表

7.2 關鍵差異

7.2.1 位元組序處理

7.2.2 集合類別

7.2.3 錯誤處理

7.3 移植步驟

8. 部署與維護

8.1 系統需求

8.2 安裝部署

8.3 監控維護

8.4 疑難排解

重要提醒

下一步行動

附錄：快速參考

專有名詞

重要封包

關鍵檔案位置