import streamlit as st import json import pandas as pd import re import xml.etree.ElementTree as ET st.set_page_config(layout="wide") st.title("FORScan Config Viewer & Diff Engine") # --- CACHED DATA LOADERS (The Speed Upgrade!) --- @st.cache_data def load_json_data(json_bytes): """Loads the JSON file once and keeps it in memory.""" return json.loads(json_bytes.decode('utf-8')) @st.cache_data def process_vehicle_file(file_name, file_bytes): """Parses the AB/ABT files once and keeps the dictionary in memory.""" vehicle_dict = {} content_str = file_bytes.decode("utf-8", errors="ignore").strip() # 1. Attempt to parse as Ford Motorcraft XML (.ab format) if "" in content_str or "" in content_str: try: start_idx = content_str.find("<") root = ET.fromstring(content_str[start_idx:]) bce_module = root.find(".//BCE_MODULE") if bce_module is not None: for data in bce_module.findall("DATA"): label = data.get("LABEL") if label: codes = [code.text for code in data.findall("CODE") if code.text] hex_data = "".join(codes).upper().replace(" ", "") vehicle_dict[label.upper()] = hex_data if vehicle_dict: return vehicle_dict except ET.ParseError: pass # 2. Match standard unencrypted Ford AS-BUILT format (Plain text) std_matches = re.finditer(r'([0-9A-Fa-f]{3}-\d{2}-\d{2})[^\dA-Fa-f]*([0-9A-Fa-f]+(?:\s+[0-9A-Fa-f]+)*)', content_str) for match in std_matches: address = match.group(1).upper() hex_data = re.sub(r'[^0-9A-Fa-f]', '', match.group(2)).upper() vehicle_dict[address] = hex_data # 3. Match encrypted FORScan .abt format abt_matches = re.finditer(r'([0-9A-Fa-f]{3}[G-V][0-9A-Fa-f][G-V][0-9A-Fa-f])([0-9A-Fa-f]+)', content_str) for match in abt_matches: raw_addr = match.group(1).upper() hex_data = match.group(2).upper() module = raw_addr[:3] b1 = int(ord(raw_addr[3]) - ord('G')) b2 = int(raw_addr[4], 16) block_dec = (b1 << 4) + b2 l1 = int(ord(raw_addr[5]) - ord('G')) l2 = int(raw_addr[6], 16) line_dec = (l1 << 4) + l2 decoded_addr = f"{module}-{block_dec:02d}-{line_dec:02d}" vehicle_dict[decoded_addr] = hex_data return vehicle_dict @st.cache_data def build_feature_index(data): """Scans the JSON to build the search index once and caches the list.""" all_features = set() for block in data: if block['type'] == 'discrete_matrix': for s in block['schema']: base_name = s.get('base_name', 'Unknown') raw_name = s.get('raw_name', 'Unknown') major_group = s.get('major_group', '') display_text = f"{base_name} ({raw_name})" all_features.add((display_text, major_group)) return sorted(list(all_features), key=lambda x: x[0]) # --- HELPER FUNCTIONS --- def extract_active_hex_string(mask, vehicle_hex): clean_mask = mask.replace(" ", "") clean_hex = vehicle_hex.replace(" ", "") active_val = "" for i, m_char in enumerate(clean_mask): if m_char == '*' and i < len(clean_hex): active_val += clean_hex[i] return active_val.upper() if active_val else None def highlight_active_row(row, active_str): if active_str is None: return [''] * len(row) row_hex = str(row['Hex Input']).strip().upper() if row_hex == active_str: return ['background-color: #1e4620; color: #4ade80; font-weight: bold'] * len(row) return [''] * len(row) # --- UI LAYOUT --- data = None vehicles = {} diff_only = False with st.sidebar: # --- 1. DATA LOADING PANEL --- with st.expander("📁 1. Data Loading", expanded=True): json_file = st.file_uploader("Upload parsed_forscan.json", type="json") vehicle_files = st.file_uploader("Upload Vehicle File(s) (.abt / .ab)", type=["abt", "ab"], accept_multiple_files=True) if json_file is not None: # Use cached JSON loader data = load_json_data(json_file.getvalue()) if vehicle_files: st.divider() st.markdown("**Status:**") json_modules = set(block['address'][:3] for block in data) for f in vehicle_files: # Use cached Vehicle parsing (we pass f.name to keep the cache key unique per file) v_data = process_vehicle_file(f.name, f.getvalue()) vehicles[f.name] = v_data abt_modules = set(addr[:3] for addr in v_data.keys()) if not json_modules.intersection(abt_modules): st.error(f"⚠️ **{f.name}**\nNo matching module found!") else: st.success(f"✅ **{f.name}**\nLoaded successfully.") if len(vehicles) > 1: st.divider() diff_only = st.checkbox("⚡ Show Differences Only", value=False) # --- 2. SEARCH & FILTER PANEL --- if data is not None: with st.expander("🔍 2. Feature Filters", expanded=True): # Use cached feature index builder feature_options = build_feature_index(data) if "selected_features" not in st.session_state: st.session_state["selected_features"] = set() def clear_all_features(): st.session_state["selected_features"].clear() search_query = st.text_input("Search Attributes", placeholder="e.g., TSR, Cruise, HUD...") search_query = search_query.lower() active_count = len(st.session_state["selected_features"]) col1, col2 = st.columns([2, 1]) if active_count > 0: col1.info(f"**{active_count}** selected") col2.button("Clear", on_click=clear_all_features) filtered_features = [] for display_text, major_group in feature_options: if search_query in display_text.lower() or search_query in major_group.lower(): filtered_features.append(display_text) with st.container(height=500): for f in filtered_features: is_checked = f in st.session_state["selected_features"] def update_feature_state(feature_name=f): if st.session_state[f"chk_{feature_name}"]: st.session_state["selected_features"].add(feature_name) else: st.session_state["selected_features"].discard(feature_name) st.checkbox(f, value=is_checked, key=f"chk_{f}", on_change=update_feature_state) active_filters = st.session_state["selected_features"] # --- MAIN UI LOOP --- if data is not None: for block in data: addr = block['address'] mask = block['mask'] if active_filters: if block['type'] != 'discrete_matrix': continue block_feature_names = [f"{s.get('base_name', 'Unknown')} ({s.get('raw_name', 'Unknown')})" for s in block['schema']] if not active_filters.intersection(set(block_feature_names)): continue active_hexes = {} for v_name, v_data in vehicles.items(): if addr in v_data: active_hexes[v_name] = extract_active_hex_string(mask, v_data[addr]) else: active_hexes[v_name] = None if diff_only and len(vehicles) > 1: if len(set(active_hexes.values())) <= 1: continue st.subheader(f"{addr} | Mask: `{mask}`") if block['type'] == 'continuous_input': instruction = block['schema'][0].get('instruction', 'None Found') st.info(f"**Continuous/Direct Input:** {instruction}") if vehicles: cols = st.columns(len(vehicles)) for col, (v_name, v_active_str) in zip(cols, active_hexes.items()): with col: st.markdown(f"**📄 {v_name}**") st.code(f"Hex: {v_active_str if v_active_str else 'N/A'}") else: schema_names = [f"{s.get('readable_name', 'Unknown')} ({s.get('acronym', '')})" for s in block['schema']] meta_tags = [] for s in block['schema']: if s.get('is_config'): meta_tags.append(f"`{s['config_type']} Config`") if meta_tags: st.markdown(f"**Types:** {' '.join(set(meta_tags))}") st.markdown(f"**Mapped Attributes:** `{'` • `'.join(schema_names)}`") table_data = [] for state in block['states']: row_data = {"Hex Input": state['hex']} for parsed_item in state['parsed']: idx = parsed_item['index'] col_name = schema_names[idx] if idx < len(schema_names) else f"Unknown_Index_{idx}" row_data[col_name] = parsed_item['state_value'] table_data.append(row_data) if table_data: df = pd.DataFrame(table_data) if not vehicles: st.dataframe(df, use_container_width=True, hide_index=True) else: cols = st.columns(len(vehicles)) for col, (v_name, v_active_str) in zip(cols, active_hexes.items()): with col: st.markdown(f"**📄 {v_name}**") styled_df = df.style.apply(highlight_active_row, active_str=v_active_str, axis=1) st.dataframe(styled_df, use_container_width=True, hide_index=True) st.divider()