parser.py

#!/usr/bin/env python3
"""Parser DRM .dat berdasarkan workbook referensi Excel.

Program tidak menjalankan VBA dari workbook. File biner ditulis ke sheet raw
(`csv`/`csv_pre`) dengan susunan 16 byte per baris, lalu LibreOffice dipakai
untuk menghitung ulang rumus workbook secara headless.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import math
import os
import re
import shutil
import subprocess
import sys
import tempfile
import zipfile
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable
from xml.etree import ElementTree as ET


MAIN_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
DOC_REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
PKG_REL_NS = "http://schemas.openxmlformats.org/package/2006/relationships"
NS = {"m": MAIN_NS, "r": DOC_REL_NS, "pr": PKG_REL_NS}
ET.register_namespace("", MAIN_NS)
ET.register_namespace("r", DOC_REL_NS)

RAW_WIDTH = 16
DEFAULT_TEMPLATE = Path(__file__).with_name("Template.xlsm")
DATA_FOR_CONVERT_ENGLISH_REFERENCE = Path(__file__).with_name("data_for_convert_english.csv")

# Nilai suhu coolant pada DRM disimpan sebagai nilai Celsius + 40. Rumus
# telemetry ENG_DMU(CV) di template sudah mengurangi offset tersebut, tetapi
# blok Top 10 Maximum Coolant Temperature belum melakukannya.
MAX_COOLANT_TEMPERATURE_CELLS = tuple(
    f"{column}{row}"
    for row in range(3110, 3115)
    for column in ("I", "K")
)
COOLANT_TEMPERATURE_OFFSET_C = 40

# Kolom B:BS pada sheet ENG_DMU(CV). Kolom BT kosong, sedangkan BU:BW
# merupakan kolom helper untuk grafik Excel sehingga tidak diekspor ke JSON.
ENGINE_DMU_CV_FIELDS = (
    "sec",
    "accelerator_pedal_position_percent",
    "actual_engine_torque_nm",
    "engine_speed_rpm",
    "target_fuel_quantity_mm3_per_stroke",
    "current_gear",
    "lockup_clutch_actuator",
    "clutch_actuator",
    "pto_control_status",
    "air_conditioner_clutch_switch",
    "exhaust_brake_switch",
    "refrigerator_switch",
    "turbo_vane_magnetic_valve_1",
    "turbo_vane_magnetic_valve_2",
    "turbo_vane_magnetic_valve_3",
    "exhaust_brake_valve_control",
    "starter_switch",
    "neutral_switch",
    "parking_brake_switch",
    "vehicle_speed_km_per_h",
    "clutch_switch",
    "brake_switch",
    "auto_cruise_control_status",
    "engine_coolant_temperature_c",
    "fuel_temperature_c",
    "post_injection_quantity_mm3_per_stroke",
    "common_rail_pressure_mpa",
    "dpd_differential_pressure_kpa",
    "barometric_pressure_kpa",
    "intake_air_temperature_c",
    "intake_air_pressure_kpa",
    "csf_entrance_temperature_c",
    "doc_entrance_temperature_c",
    "dpd_status",
    "dpd_error_count",
    "dpd_warning_count",
    "dpd_pm_storage_status",
    "dpd_trip_status",
    "intake_throttle_duty_limit_status",
    "egr_duty_limit_status",
    "egr2_duty_limit_status",
    "dpd_mode",
    "mass_air_flow_before_stage_15_g_per_cyl",
    "mass_air_flow_from_stage_15_g_per_cyl",
    "egr1_duty_percent",
    "egr1_target_position_percent",
    "egr1_actual_position_percent",
    "intake_throttle_duty_percent",
    "intake_throttle_target_position_percent",
    "intake_throttle_actual_position_percent",
    "ign_voltage_v",
    "scv_duty_or_pcv_close_timing_percent",
    "scv_actual_current_ma_or_pcv_feedback_control_quantity",
    "scv_target_current_ma_or_egr2_actual_position_percent",
    "scv_target_fuel_flow_mm3_per_stroke_or_egr2_duty_percent",
    "target_rail_pressure_mpa",
    "turbo_vane_actual_position_percent",
    "turbo_vane_target_position_percent",
    "target_boost_percent",
    "engine_mode",
    "main_soi",
    "pilot_soi",
    "cam_crank_syncro_status",
    "cylinder_1_balancing_fuel_compensation_mm3_per_stroke",
    "cylinder_2_balancing_fuel_compensation_mm3_per_stroke",
    "cylinder_3_balancing_fuel_compensation_mm3_per_stroke",
    "cylinder_4_balancing_fuel_compensation_mm3_per_stroke",
    "cylinder_5_balancing_fuel_compensation_mm3_per_stroke",
    "cylinder_6_balancing_fuel_compensation_mm3_per_stroke",
    "target_idle_rpm",
)


class ParserError(RuntimeError):
    pass


def qname(namespace: str, local: str) -> str:
    return f"{{{namespace}}}{local}"


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def column_name(index: int) -> str:
    result = ""
    while index:
        index, remainder = divmod(index - 1, 26)
        result = chr(65 + remainder) + result
    return result


def normalized_text(raw: bytes) -> str:
    return raw.decode("latin-1", errors="replace").replace("\x00", "").strip(" \x1e\xff")


def normalized_workbook_text(value: Any) -> Any:
    if not isinstance(value, str):
        return value
    # LibreOffice dapat menggandakan escape OOXML untuk karakter kontrol.
    cleaned = value.replace("_x005F_", "_")
    cleaned = re.sub(r"_x[0-9A-Fa-f]{4}_", "", cleaned)
    return cleaned.strip()


def quick_header(data: bytes) -> dict[str, Any]:
    result: dict[str, Any] = {}
    if len(data) >= 5:
        year, month, day, hour, minute = data[:5]
        try:
            parsed = datetime(2000 + year, month, day, hour, minute)
            result["download_time"] = parsed.isoformat(timespec="minutes")
        except ValueError:
            result["download_time"] = None
    if len(data) >= 26:
        result["vin"] = normalized_text(data[6:16] + data[18:26])
    if len(data) >= 34:
        result["engine_type"] = normalized_text(data[26:34])
    if len(data) >= 42:
        result["transmission_type"] = normalized_text(data[34:42])
    if len(data) >= 50:
        result["drm_type_code"] = int.from_bytes(data[46:50], "big")
    if len(data) >= 54:
        result["country_code"] = normalized_text(data[50:54])
    if len(data) >= 60:
        result["distributor_code"] = normalized_text(data[54:60])
    if len(data) >= 64:
        result["rear_body_code"] = normalized_text(data[60:64])
    if len(data) >= 68:
        result["load_code"] = normalized_text(data[64:68])
    if len(data) >= 72:
        result["odometer_km"] = int.from_bytes(data[68:72], "big")
    return result


def write_raw_csv(data: bytes, destination: Path) -> None:
    with destination.open("w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.writer(handle)
        writer.writerow(["offset_hex", "offset_dec", *[f"byte_{i:02d}" for i in range(16)], "ascii"])
        for offset in range(0, len(data), RAW_WIDTH):
            chunk = data[offset : offset + RAW_WIDTH]
            padded: list[Any] = list(chunk) + [""] * (RAW_WIDTH - len(chunk))
            ascii_preview = "".join(chr(value) if 32 <= value <= 126 else "." for value in chunk)
            writer.writerow([f"0x{offset:08X}", offset, *padded, ascii_preview])


def sheet_paths_from_package(package: zipfile.ZipFile) -> dict[str, str]:
    workbook = ET.fromstring(package.read("xl/workbook.xml"))
    relationships = ET.fromstring(package.read("xl/_rels/workbook.xml.rels"))
    targets = {
        rel.attrib["Id"]: rel.attrib["Target"]
        for rel in relationships.findall("pr:Relationship", NS)
    }
    result: dict[str, str] = {}
    for sheet in workbook.findall("m:sheets/m:sheet", NS):
        rel_id = sheet.attrib[qname(DOC_REL_NS, "id")]
        target = targets[rel_id].lstrip("/")
        if not target.startswith("xl/"):
            target = f"xl/{target}"
        result[sheet.attrib["name"]] = target
    return result


def formula_row_capacity(xml_data: bytes) -> int:
    root = ET.fromstring(xml_data)
    max_row = 0
    for cell in root.findall(".//m:c", NS):
        ref = cell.attrib.get("r", "")
        letters = "".join(ch for ch in ref if ch.isalpha())
        digits = "".join(ch for ch in ref if ch.isdigit())
        if not digits or not letters or cell.find("m:f", NS) is None:
            continue
        column = 0
        for ch in letters:
            column = column * 26 + ord(ch.upper()) - 64
        if column <= RAW_WIDTH:
            max_row = max(max_row, int(digits))
    return max_row


def replace_sheet_data(xml_data: bytes, data: bytes) -> bytes:
    root = ET.fromstring(xml_data)
    sheet_data = root.find("m:sheetData", NS)
    if sheet_data is None:
        raise ParserError("Sheet raw tidak memiliki elemen sheetData.")
    sheet_data.clear()

    row_count = math.ceil(len(data) / RAW_WIDTH) if data else 0
    for row_number in range(1, row_count + 1):
        offset = (row_number - 1) * RAW_WIDTH
        chunk = data[offset : offset + RAW_WIDTH]
        row = ET.SubElement(sheet_data, qname(MAIN_NS, "row"), {"r": str(row_number)})
        for column_number, value in enumerate(chunk, start=1):
            ref = f"{column_name(column_number)}{row_number}"
            cell = ET.SubElement(row, qname(MAIN_NS, "c"), {"r": ref, "t": "n"})
            ET.SubElement(cell, qname(MAIN_NS, "v")).text = str(value)

    dimension = root.find("m:dimension", NS)
    if dimension is not None:
        dimension.attrib["ref"] = f"A1:P{row_count}" if row_count else "A1"
    return ET.tostring(root, encoding="utf-8", xml_declaration=True)


def set_inline_string(xml_data: bytes, reference: str, value: str) -> bytes:
    root = ET.fromstring(xml_data)
    sheet_data = root.find("m:sheetData", NS)
    if sheet_data is None:
        return xml_data
    row_number = int("".join(ch for ch in reference if ch.isdigit()))
    row = next((item for item in sheet_data.findall("m:row", NS) if int(item.attrib["r"]) == row_number), None)
    if row is None:
        row = ET.SubElement(sheet_data, qname(MAIN_NS, "row"), {"r": str(row_number)})
    cell = next((item for item in row.findall("m:c", NS) if item.attrib.get("r") == reference), None)
    style = cell.attrib.get("s") if cell is not None else None
    if cell is not None:
        row.remove(cell)
    attributes = {"r": reference, "t": "inlineStr"}
    if style is not None:
        attributes["s"] = style
    cell = ET.SubElement(row, qname(MAIN_NS, "c"), attributes)
    inline = ET.SubElement(cell, qname(MAIN_NS, "is"))
    ET.SubElement(inline, qname(MAIN_NS, "t")).text = value
    return ET.tostring(root, encoding="utf-8", xml_declaration=True)


def force_recalculation(xml_data: bytes) -> bytes:
    root = ET.fromstring(xml_data)
    calc = root.find("m:calcPr", NS)
    if calc is None:
        calc = ET.SubElement(root, qname(MAIN_NS, "calcPr"))
    calc.attrib.update({"calcMode": "auto", "fullCalcOnLoad": "1", "forceFullCalc": "1"})
    return ET.tostring(root, encoding="utf-8", xml_declaration=True)


def clear_formula_caches(xml_data: bytes) -> bytes:
    """Hapus hasil formula lama agar mesin spreadsheet wajib menghitung ulang."""
    root = ET.fromstring(xml_data)
    changed = False
    for cell in root.findall(".//m:c", NS):
        if cell.find("m:f", NS) is None:
            continue
        cached = cell.find("m:v", NS)
        if cached is not None:
            cell.remove(cached)
            changed = True
    if not changed:
        return xml_data
    return ET.tostring(root, encoding="utf-8", xml_declaration=True)


def correct_max_coolant_temperature_formulas(xml_data: bytes) -> bytes:
    """Terapkan offset DRM pada 10 record suhu coolant maksimum."""
    root = ET.fromstring(xml_data)
    cells = {
        cell.attrib.get("r"): cell
        for cell in root.findall(".//m:c", NS)
        if cell.attrib.get("r") in MAX_COOLANT_TEMPERATURE_CELLS
    }
    missing = sorted(set(MAX_COOLANT_TEMPERATURE_CELLS) - set(cells))
    if missing:
        raise ParserError(
            "Template tidak sesuai; cell suhu coolant maksimum hilang: "
            + ", ".join(missing)
        )

    suffix = f"-{COOLANT_TEMPERATURE_OFFSET_C}"
    for reference in MAX_COOLANT_TEMPERATURE_CELLS:
        formula = cells[reference].find("m:f", NS)
        if formula is None or not formula.text:
            raise ParserError(
                f"Template tidak sesuai; formula suhu coolant maksimum hilang: {reference}"
            )
        if not formula.text.endswith(suffix):
            formula.text = f"({formula.text}){suffix}"

    return ET.tostring(root, encoding="utf-8", xml_declaration=True)


def build_calculation_workbook(template: Path, current: Path, previous: Path | None, destination: Path) -> dict[str, int]:
    current_data = current.read_bytes()
    previous_data = previous.read_bytes() if previous else b""

    with zipfile.ZipFile(template, "r") as source:
        paths = sheet_paths_from_package(source)
        required = {"TOP", "csv", "csv_pre", "Current", "Previous", "Data for Convert"}
        missing = sorted(required - set(paths))
        if missing:
            raise ParserError(f"Template tidak sesuai; sheet hilang: {', '.join(missing)}")

        current_rows = formula_row_capacity(source.read(paths["Current"]))
        previous_rows = formula_row_capacity(source.read(paths["Previous"]))
        if len(current_data) > current_rows * RAW_WIDTH:
            raise ParserError(
                f"File current berisi {len(current_data):,} byte; kapasitas rumus template "
                f"hanya {current_rows * RAW_WIDTH:,} byte."
            )
        if len(previous_data) > previous_rows * RAW_WIDTH:
            raise ParserError(
                f"File previous berisi {len(previous_data):,} byte; kapasitas rumus template "
                f"hanya {previous_rows * RAW_WIDTH:,} byte."
            )

        data_for_convert_xml = correct_max_coolant_temperature_formulas(
            source.read(paths["Data for Convert"])
        )
        replacements = {
            paths["csv"]: replace_sheet_data(source.read(paths["csv"]), current_data),
            paths["csv_pre"]: replace_sheet_data(source.read(paths["csv_pre"]), previous_data),
            paths["TOP"]: set_inline_string(source.read(paths["TOP"]), "I4", str(current.resolve())),
            paths["Data for Convert"]: clear_formula_caches(data_for_convert_xml),
            "xl/workbook.xml": force_recalculation(source.read("xl/workbook.xml")),
        }
        raw_sheet_paths = {paths["csv"], paths["csv_pre"]}
        with zipfile.ZipFile(destination, "w") as target:
            for info in source.infolist():
                content = replacements.get(info.filename)
                if content is None:
                    content = source.read(info.filename)
                    if info.filename.startswith("xl/worksheets/") and info.filename not in raw_sheet_paths:
                        content = clear_formula_caches(content)
                target.writestr(info, content)

    return {
        "current_capacity_bytes": current_rows * RAW_WIDTH,
        "previous_capacity_bytes": previous_rows * RAW_WIDTH,
    }


def find_soffice(explicit: str | None) -> Path:
    candidates: list[Path] = []
    if explicit:
        candidates.append(Path(explicit).expanduser())
    found = shutil.which("soffice")
    if found:
        candidates.append(Path(found))
    candidates.extend(
        [
            Path("/Applications/LibreOffice.app/Contents/MacOS/soffice"),
            Path("/Applications/LibreOfficeDev.app/Contents/MacOS/soffice"),
            Path("C:/Program Files/LibreOffice/program/soffice.exe"),
            Path("C:/Program Files (x86)/LibreOffice/program/soffice.exe"),
        ]
    )
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    raise ParserError("LibreOffice/soffice tidak ditemukan. Gunakan --soffice PATH atau mode --raw-only.")


def recalculate_with_libreoffice(source_xlsm: Path, destination_xlsx: Path, soffice: Path, working_dir: Path) -> None:
    profile = working_dir / "libreoffice_profile"
    converted = working_dir / f"{source_xlsm.stem}.xlsx"
    profile.mkdir(parents=True, exist_ok=True)
    command = [
        str(soffice),
        f"-env:UserInstallation={profile.resolve().as_uri()}",
        "--headless",
        "--convert-to",
        "xlsx",
        "--outdir",
        str(working_dir),
        str(source_xlsm),
    ]
    env = os.environ.copy()
    env.setdefault("SAL_USE_VCLPLUGIN", "svp")
    completed = subprocess.run(command, text=True, capture_output=True, env=env, timeout=180)
    if completed.returncode != 0 or not converted.exists():
        details = (completed.stdout + "\n" + completed.stderr).strip()
        raise ParserError(f"LibreOffice gagal menghitung workbook.\n{details}")
    shutil.copy2(converted, destination_xlsx)


def json_value(value: Any) -> Any:
    if isinstance(value, (datetime,)):
        return value.isoformat()
    return value


def last_nonempty_bounds(worksheet: Any, max_column: int | None = None) -> tuple[int, int]:
    last_row = 0
    last_column = 0
    column_limit = min(worksheet.max_column, max_column or worksheet.max_column)
    for row in worksheet.iter_rows(min_row=1, max_row=worksheet.max_row, min_col=1, max_col=column_limit):
        for cell in row:
            if cell.value is not None:
                last_row = max(last_row, cell.row)
                last_column = max(last_column, cell.column)
    return last_row, last_column


def export_sheet_csv(workbook: Any, sheet_name: str, destination: Path, max_column: int | None = None) -> None:
    worksheet = workbook[sheet_name]
    last_row, last_column = last_nonempty_bounds(worksheet, max_column=max_column)
    with destination.open("w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.writer(handle)
        if not last_row or not last_column:
            return
        for row in worksheet.iter_rows(min_row=1, max_row=last_row, min_col=1, max_col=last_column, values_only=True):
            writer.writerow([json_value(value) for value in row])


def export_sheet_json(
    workbook: Any,
    sheet_name: str,
    destination: Path,
    max_column: int | None = None,
    english_reference: Path | None = None,
) -> dict[str, Any]:
    worksheet = workbook[sheet_name]
    last_row, last_column = last_nonempty_bounds(worksheet, max_column=max_column)
    rows = []
    if last_row and last_column:
        rows = [
            [json_value(value) for value in row]
            for row in worksheet.iter_rows(
                min_row=1,
                max_row=last_row,
                min_col=1,
                max_col=last_column,
                values_only=True,
            )
        ]
    payload = {
        "sheet": sheet_name,
        "range": f"A1:{column_name(last_column)}{last_row}" if last_row and last_column else None,
        "row_count": last_row,
        "column_count": last_column,
        "columns": [column_name(index) for index in range(1, last_column + 1)],
        "rows": rows,
    }
    if english_reference is not None:
        apply_data_for_convert_english_reference(payload, english_reference)
    destination.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return payload


def apply_data_for_convert_english_reference(payload: dict[str, Any], reference: Path) -> None:
    with reference.open("r", newline="", encoding="utf-8-sig") as handle:
        reference_rows = list(csv.reader(handle))

    row_count = payload["row_count"]
    column_count = payload["column_count"]
    if len(reference_rows) != row_count or any(len(row) != column_count for row in reference_rows):
        raise ParserError(
            "Ukuran data_for_convert_english.csv tidak cocok dengan sheet Data for Convert: "
            f"referensi={len(reference_rows)} baris, sheet={row_count}x{column_count}."
        )

    expected_header = ["Input Item", "Format", "Byte Count", "Retrieved Value"]
    if reference_rows[0][16:20] != expected_header:
        raise ParserError("Header data_for_convert_english.csv tidak valid.")

    # A:H berisi seluruh label laporan. Q:R berisi nama/format metadata.
    # S1:T1 adalah header; nilai aktual tetap berasal dari workbook di I:O dan T:U.
    label_columns = (*range(8), 16, 17)
    for row_index, output_row in enumerate(payload["rows"]):
        reference_row = reference_rows[row_index]
        for column_index in label_columns:
            if reference_row[column_index]:
                output_row[column_index] = reference_row[column_index]
    for column_index in (18, 19):
        payload["rows"][0][column_index] = reference_rows[0][column_index]


def cell_value(workbook: Any, sheet: str, address: str) -> Any:
    return json_value(workbook[sheet][address].value)


def main_data_summary(workbook: Any) -> dict[str, Any]:
    fields = {
        "download_time_current": "D5",
        "download_time_previous": "D6",
        "vin": "D7",
        "engine_type": "D8",
        "rear_body_type": "D9",
        "load_type": "D10",
        "odometer_km": "D11",
        "key_on_hours": "D12",
        "total_mileage_km": "D13",
        "vehicle_running_hours": "D14",
        "fuel_consumed_l": "D15",
        "average_fuel_economy_including_idle_km_per_l": "D16",
        "average_fuel_economy_excluding_idle_km_per_l": "D17",
        "average_speed_including_idle_km_per_h": "D18",
        "average_speed_excluding_idle_km_per_h": "D19",
        "total_idle_hours": "D20",
        "idle_under_5_min_hours": "D21",
        "idle_over_5_min_hours": "D22",
        "idle_fuel_l": "D23",
        "idle_fuel_l_per_h": "D24",
        "pto_activation_count": "D25",
        "pto_activation_hours": "D26",
        "engine_start_cycles": "D27",
        "engine_run_hours": "D28",
        "average_engine_rpm": "D29",
        "foot_brake_count": "D30",
        "foot_brake_hours": "D31",
        "exhaust_brake_switch_count": "D32",
        "exhaust_brake_working_hours": "D33",
        "clutch_pedal_count": "D34",
    }
    result = {key: cell_value(workbook, "Main Data", address) for key, address in fields.items()}
    for key in ("vin", "engine_type", "rear_body_type", "load_type"):
        result[key] = normalized_workbook_text(result[key])
    return result


def dtc_summary(workbook: Any) -> list[dict[str, Any]]:
    worksheet = workbook["DTC history"]
    result = []
    for row in range(5, min(worksheet.max_row, 54) + 1):
        code = worksheet[f"E{row}"].value
        if code in (None, "", "-"):
            continue
        result.append(
            {
                "sequence": worksheet[f"B{row}"].value,
                "dtc_code": code,
                "dtc_name": worksheet[f"F{row}"].value,
                "mileage_km": worksheet[f"H{row}"].value,
                "elapsed_engine_hours": worksheet[f"J{row}"].value,
            }
        )
    return result


def engine_dmu_cv_summary(workbook: Any) -> dict[str, Any]:
    """Ubah empat blok freeze-frame ENG_DMU(CV) menjadi JSON terstruktur."""
    sheet_name = "ENG_DMU(CV)"
    if sheet_name not in workbook.sheetnames:
        return {"sheet": sheet_name, "columns": [], "events": []}

    worksheet = workbook[sheet_name]
    first_column = 2
    last_column = first_column + len(ENGINE_DMU_CV_FIELDS) - 1
    rows = list(
        worksheet.iter_rows(
            min_row=1,
            max_row=worksheet.max_row,
            min_col=1,
            max_col=last_column,
            values_only=True,
        )
    )
    block_indexes = []
    for row_index, row in enumerate(rows):
        title = normalized_workbook_text(row[0])
        if isinstance(title, str) and title.startswith("ENG_DMU"):
            block_indexes.append(row_index)

    if not block_indexes:
        return {"sheet": sheet_name, "columns": [], "events": []}

    first_header_index = block_indexes[0] + 5
    columns = [
        {
            "key": key,
            "label": normalized_workbook_text(rows[first_header_index][column - 1]),
            "source_column": column_name(column),
        }
        for column, key in zip(range(first_column, last_column + 1), ENGINE_DMU_CV_FIELDS)
    ]

    events = []
    for sequence, block_index in enumerate(block_indexes, start=1):
        next_block_index = block_indexes[sequence] if sequence < len(block_indexes) else len(rows)
        header_index = block_index + 5
        samples = []
        for row in rows[header_index + 1 : next_block_index]:
            if row[first_column - 1] in (None, ""):
                continue
            samples.append(
                {
                    key: json_value(value)
                    for key, value in zip(
                        ENGINE_DMU_CV_FIELDS,
                        row[first_column - 1 : last_column],
                    )
                }
            )

        dtc_code = normalized_workbook_text(rows[block_index + 1][4])
        if dtc_code not in (None, ""):
            dtc_code = str(dtc_code)
        events.append(
            {
                "sequence": sequence,
                "name": normalized_workbook_text(rows[block_index][0]),
                "dtc_code": dtc_code,
                "total_mileage_km": json_value(rows[block_index + 2][4]),
                "elapsed_time_from_key_on_min": json_value(rows[block_index + 3][4]),
                "samples": samples,
            }
        )

    return {"sheet": sheet_name, "columns": columns, "events": events}


def load_openpyxl() -> Any:
    try:
        from openpyxl import load_workbook
    except ImportError as exc:
        raise ParserError("Modul openpyxl belum terpasang. Jalankan: pip install -r requirements.txt") from exc
    return load_workbook


def ensure_output_dir(path: Path, force: bool) -> None:
    if path.exists() and any(path.iterdir()) and not force:
        raise ParserError(f"Folder output sudah berisi file: {path}. Gunakan --force untuk menimpa file keluaran.")
    path.mkdir(parents=True, exist_ok=True)


def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Parse DRM .dat memakai aturan workbook (16 byte per baris + rumus workbook)."
    )
    parser.add_argument("current", type=Path, help="File DRM .dat current")
    parser.add_argument("--previous", type=Path, help="File DRM .dat previous untuk analisis delta")
    parser.add_argument("--template", type=Path, default=DEFAULT_TEMPLATE, help="Workbook template .xlsm")
    parser.add_argument("--output-dir", type=Path, help="Folder output (default: <nama_dat>_parsed)")
    parser.add_argument("--soffice", help="Path executable LibreOffice soffice")
    parser.add_argument("--raw-only", action="store_true", help="Hanya ekspor byte mentah dan metadata header")
    parser.add_argument("--force", action="store_true", help="Izinkan menimpa file keluaran dengan nama yang sama")
    return parser.parse_args(argv)


def run(argv: Iterable[str] | None = None) -> int:
    args = parse_args(argv)
    current = args.current.expanduser().resolve()
    previous = args.previous.expanduser().resolve() if args.previous else None
    template = args.template.expanduser().resolve()
    output_dir = (args.output_dir or Path.cwd() / f"{current.stem}_parsed").expanduser().resolve()

    if not current.is_file():
        raise ParserError(f"File current tidak ditemukan: {current}")
    if current.suffix.lower() != ".dat":
        raise ParserError(f"File current harus berekstensi .dat: {current.name}")
    if previous and (not previous.is_file() or previous.suffix.lower() != ".dat"):
        raise ParserError(f"File previous tidak valid: {previous}")

    ensure_output_dir(output_dir, args.force)
    current_data = current.read_bytes()
    write_raw_csv(current_data, output_dir / "raw_bytes.csv")

    result: dict[str, Any] = {
        "parser": "DRM DAT parser",
        "parsing_basis": "Workbook; VBA dat_2_csv uses 16 bytes per row",
        "current": {"path": str(current), "size_bytes": len(current_data), "sha256": sha256(current)},
        "previous": None,
        "quick_header": quick_header(current_data),
    }
    if previous:
        previous_data = previous.read_bytes()
        result["previous"] = {
            "path": str(previous),
            "size_bytes": len(previous_data),
            "sha256": sha256(previous),
            "quick_header": quick_header(previous_data),
        }

    if args.raw_only:
        result["mode"] = "raw-only"
        (output_dir / "summary.json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"Selesai (raw-only): {output_dir}")
        return 0

    if not template.is_file():
        raise ParserError(f"Template workbook tidak ditemukan: {template}")
    if not DATA_FOR_CONVERT_ENGLISH_REFERENCE.is_file():
        raise ParserError(
            "Referensi terjemahan tidak ditemukan: "
            f"{DATA_FOR_CONVERT_ENGLISH_REFERENCE}"
        )
    soffice = find_soffice(args.soffice)
    load_workbook = load_openpyxl()

    with tempfile.TemporaryDirectory(prefix="drm_dat_parser_") as temp_name:
        temp_dir = Path(temp_name)
        calculation_book = temp_dir / "drm_calculation.xlsm"
        capacity = build_calculation_workbook(template, current, previous, calculation_book)
        final_workbook = output_dir / "parsed_report.xlsx"
        recalculate_with_libreoffice(calculation_book, final_workbook, soffice, temp_dir)

    workbook = load_workbook(final_workbook, read_only=True, data_only=True)
    try:
        exports = {
            "Main Data": ("main_data.csv", 13),
            "DTC history": ("dtc_history.csv", 10),
            "Data for Convert": ("data_for_convert.csv", 30),
            "ENG_DMU(CV)": ("engine_dmu_cv.csv", None),
            "ENG_DMU(LCV RT85 RT50_4J)": ("engine_dmu_lcv_4j.csv", None),
            "ENG_DMU(LCV_RT85_RZ4E)": ("engine_dmu_lcv_rz4e.csv", None),
            "ENG_DMU(LCV RT75)": ("engine_dmu_lcv_rt75.csv", None),
        }
        written = []
        for sheet_name, (filename, max_column) in exports.items():
            if sheet_name not in workbook.sheetnames:
                continue
            export_sheet_csv(workbook, sheet_name, output_dir / filename, max_column=max_column)
            written.append(filename)
        data_for_convert_json_filename = "data_for_convert.json"
        if "Data for Convert" in workbook.sheetnames:
            export_sheet_json(
                workbook,
                "Data for Convert",
                output_dir / data_for_convert_json_filename,
                max_column=30,
                english_reference=DATA_FOR_CONVERT_ENGLISH_REFERENCE,
            )
            written.append(data_for_convert_json_filename)
        main_summary = main_data_summary(workbook)
        if previous is None:
            main_summary["download_time_previous"] = None
        engine_dmu_cv = engine_dmu_cv_summary(workbook)
        engine_dmu_cv_filename = "engine_dmu_cv.json"
        (output_dir / engine_dmu_cv_filename).write_text(
            json.dumps(engine_dmu_cv, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        written.append(engine_dmu_cv_filename)
        result.update(
            {
                "mode": "full-workbook",
                "template": str(template),
                "template_sha256": sha256(template),
                "capacity": capacity,
                "main_data": main_summary,
                "dtc_history": dtc_summary(workbook),
                "engine_dmu_cv": engine_dmu_cv,
                "outputs": ["parsed_report.xlsx", "raw_bytes.csv", *written],
            }
        )
    finally:
        workbook.close()

    (output_dir / "summary.json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"Selesai: {output_dir}")
    return 0


def main() -> None:
    try:
        raise SystemExit(run())
    except ParserError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(2)


if __name__ == "__main__":
    main()