from __future__ import annotations import json from pathlib import Path import sys ROOT = Path(r"D:\TMCEPCW3\epcdata") MARKETS = ["EU", "GR", "US", "JP"] def decode_acd(data: bytes) -> bytes: return bytes(byte ^ 0xC8 for byte in data) def text(raw: bytes) -> str: return raw.decode("latin1", errors="ignore").strip() def rows_fixed(path: Path, length: int, encoded: bool = False): data = path.read_bytes() if encoded: data = decode_acd(data) for row_no, offset in enumerate(range(0, len(data), length), start=1): row = data[offset : offset + length] if len(row) == length: yield row_no, offset, row def iter_vin_index(market: str): for source in ["JOHOCL", "JOHOVN"]: path = ROOT / market / f"{source}.ACD" if not path.exists(): continue for row_no, offset, row in rows_fixed(path, 35, encoded=True): yield { "market": market, "source": source, "source_file": str(path), "row_no": row_no, "offset": offset, "catalog": text(row[0:6]), "vin_key": text(row[6:15]), "model": text(row[15:35]), } def iter_vehicle_detail(market: str): for source in ["JOHOJP", "JOHOKT"]: path = ROOT / market / f"{source}.ACD" if not path.exists(): continue for row_no, offset, row in rows_fixed(path, 186, encoded=True): yield { "market": market, "source": source, "source_file": str(path), "row_no": row_no, "offset": offset, "catalog": text(row[0:6]), "vin_key": text(row[6:15]), "model": text(row[15:35]), "production_start": text(row[35:41]), "production_end": text(row[41:47]), "frame_code": text(row[47:54]), "spec_code": text(row[54:62]), "engine_epc": text(row[62:82]), "body": text(row[82:92]), "transmission": text(row[92:102]), "gear": text(row[102:107]), "steering": text(row[107:112]), "door": text(row[112:117]), "grade": text(row[117:122]), "turbo": text(row[122:127]), "destination": text(row[127:132]), } def iter_shamei(market: str): path = ROOT / market / "SHAMEI.ACD" if not path.exists(): return for row_no, offset, row in rows_fixed(path, 111, encoded=True): yield { "market": market, "source": "SHAMEI", "source_file": str(path), "row_no": row_no, "offset": offset, "series_code": text(row[0:3]), "vehicle_name_epc": text(row[3:23]), "catalog": text(row[23:29]), "model_family": text(row[29:79]), "vehicle_range_start": text(row[79:85]), "vehicle_range_end": text(row[85:91]), "release_code": text(row[91:99]), "flag": text(row[99:101]), } def normalize_serial(value: str) -> int | None: value = "".join(ch for ch in value if ch.isdigit()) if not value: return None return int(value) def serial_candidates(vin: str) -> list[str]: candidates = [] compact = "".join(ch for ch in vin.upper() if ch.isalnum()) for width in [7, 6]: if len(compact) >= width: tail = compact[-width:] if tail.isdigit(): padded = tail.zfill(7) if padded not in candidates: candidates.append(padded) return candidates def framno_points(market: str, frame_code: str): path = ROOT / market / "FRAMNO.DAT" if not path.exists(): return [] points = [] for row_no, offset, row in rows_fixed(path, 99, encoded=False): row_frame = text(row[0:7]) if row_frame != frame_code: continue year = text(row[7:11]) for month in range(1, 13): raw = row[11 + (month - 1) * 7 : 11 + month * 7].decode("latin1", errors="ignore").strip() serial = normalize_serial(raw) if serial is None: continue points.append( { "market": market, "source": "FRAMNO", "source_file": str(path), "row_no": row_no, "offset": offset, "frame_code": frame_code, "year": year, "month": f"{month:02d}", "serial_raw": raw, "serial_int": serial, } ) return sorted(points, key=lambda item: (item["year"], item["month"], item["serial_int"])) def find_framno_month(market: str, frame_code: str, serial_raw: str): serial = normalize_serial(serial_raw) if serial is None: return None points = framno_points(market, frame_code) if not points: return None best = None next_point = None for index, point in enumerate(points): if point["serial_int"] <= serial: best = point next_point = points[index + 1] if index + 1 < len(points) else None elif best is None: next_point = point break else: break if best is None: return { "frame_serial": serial_raw, "production_month_epc": None, "framno_rule": "before_first_point", "next_point": next_point, } return { "frame_serial": serial_raw, "production_month_epc": f"{best['year']}{best['month']}", "framno_rule": "serial >= month_start and serial < next_month_start", "month_start_point": best, "next_month_start_point": next_point, } def match_vin_key(vin_key: str, vin9: str) -> bool: if not vin_key: return False return vin_key == vin9 or (len(vin_key) >= 6 and vin9.startswith(vin_key)) def ym_in_range(ym: str | None, start: str, end: str) -> bool: if not ym: return True if start and ym < start: return False if end and end != "999999" and ym >= end: return False return True def decode_vin(vin: str, model_filter: str = ""): compact = "".join(ch for ch in vin.upper() if ch.isalnum()) vin9 = compact[:9] model_filter = model_filter.upper().strip() index_matches = [ row for market in MARKETS for row in iter_vin_index(market) if match_vin_key(row["vin_key"], vin9) and (not model_filter or row["model"].upper() == model_filter) ] detail_rows = [] seen_details = set() for index_row in index_matches: for detail in iter_vehicle_detail(index_row["market"]): if ( detail["catalog"] == index_row["catalog"] and detail["model"] == index_row["model"] and detail["vin_key"] == index_row["vin_key"] ): key = (detail["source_file"], detail["row_no"]) if key not in seen_details: seen_details.add(key) detail_rows.append(detail) results = [] for detail in detail_rows: shamei = next( ( row for row in iter_shamei(detail["market"]) if row["catalog"] == detail["catalog"] ), None, ) for frame_serial in serial_candidates(compact): month = find_framno_month(detail["market"], detail["frame_code"], frame_serial) production_month = month.get("production_month_epc") if month else None if not ym_in_range(production_month, detail["production_start"], detail["production_end"]): continue item = { "input_vin": vin, "vin9": vin9, "market": detail["market"], "catalog": detail["catalog"], "vehicle_name_epc": shamei["vehicle_name_epc"] if shamei else "", "model_family": shamei["model_family"] if shamei else "", "vin_key": detail["vin_key"], "model": detail["model"], "production_start": detail["production_start"], "production_end": detail["production_end"], "frame_code": detail["frame_code"], "frame_serial": frame_serial, "engine_epc": detail["engine_epc"], "spec_code": detail["spec_code"], "body": detail["body"], "transmission": detail["transmission"], "gear": detail["gear"], "steering": detail["steering"], "door": detail["door"], "grade": detail["grade"], "turbo": detail["turbo"], "destination": detail["destination"], "source_detail": { "file": detail["source_file"], "row_no": detail["row_no"], "offset": detail["offset"], "record_len": 186, }, "source_shamei": { "file": shamei["source_file"], "row_no": shamei["row_no"], "offset": shamei["offset"], "record_len": 111, } if shamei else None, "framno_month": month, } results.append(item) return results def main() -> None: vin = sys.argv[1] if len(sys.argv) > 1 else "JTEBN99J900077939" model_filter = sys.argv[2] if len(sys.argv) > 2 else "" results = decode_vin(vin, model_filter) print(json.dumps(results, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()