#!/usr/bin/env python3
"""
FASE 3: Populate exploded diagrams and hotspots
This script creates FASE 3 tables and populates them with sample diagrams,
vehicle-diagram relationships, and clickable hotspots linked to parts.
"""
import sqlite3
import os
from typing import List, Dict, Optional
# Database path configuration
DB_PATH = os.path.join(os.path.dirname(__file__), '..', 'vehicle_database.db')
SCHEMA_PATH = os.path.join(os.path.dirname(__file__), '..', 'sql', 'schema.sql')
DIAGRAMS_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'dashboard', 'static', 'diagrams')
class Fase3Manager:
"""Manager for FASE 3 tables: diagrams, vehicle_diagrams, and diagram_hotspots"""
def __init__(self, db_path: str = DB_PATH):
self.db_path = db_path
self.connection = None
def connect(self):
"""Connect to the SQLite database"""
self.connection = sqlite3.connect(self.db_path)
self.connection.row_factory = sqlite3.Row
print(f"Connected to database: {self.db_path}")
def disconnect(self):
"""Close the database connection"""
if self.connection:
self.connection.close()
print("Disconnected from database")
def create_fase3_tables(self):
"""Create FASE 3 tables from schema file"""
if not os.path.exists(SCHEMA_PATH):
raise FileNotFoundError(f"Schema file not found: {SCHEMA_PATH}")
with open(SCHEMA_PATH, 'r') as f:
schema = f.read()
if self.connection:
cursor = self.connection.cursor()
cursor.executescript(schema)
self.connection.commit()
print("FASE 3 tables created successfully")
def create_diagrams_directory(self):
"""Create the diagrams directory structure"""
if not os.path.exists(DIAGRAMS_DIR):
os.makedirs(DIAGRAMS_DIR)
print(f"Created diagrams directory: {DIAGRAMS_DIR}")
else:
print(f"Diagrams directory already exists: {DIAGRAMS_DIR}")
def get_diagram_by_name(self, name: str) -> Optional[int]:
"""Get diagram ID by name, returns None if not found"""
cursor = self.connection.cursor()
cursor.execute("SELECT id FROM diagrams WHERE name = ?", (name,))
result = cursor.fetchone()
return result[0] if result else None
def insert_diagram(self, name: str, name_es: str, group_id: int, image_path: str,
thumbnail_path: str = None, svg_content: str = None,
width: int = 600, height: int = 400, display_order: int = 0,
source: str = None) -> int:
"""Insert a diagram if it doesn't exist, return its ID"""
existing_id = self.get_diagram_by_name(name)
if existing_id:
print(f" Diagram '{name}' already exists (ID: {existing_id})")
return existing_id
cursor = self.connection.cursor()
cursor.execute(
"""INSERT INTO diagrams (name, name_es, group_id, image_path, thumbnail_path,
svg_content, width, height, display_order, source)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(name, name_es, group_id, image_path, thumbnail_path, svg_content,
width, height, display_order, source)
)
self.connection.commit()
diagram_id = cursor.lastrowid
print(f" Inserted diagram: {name} (ID: {diagram_id})")
return diagram_id
def get_vehicle_diagram(self, diagram_id: int, model_year_engine_id: int) -> Optional[int]:
"""Check if a vehicle-diagram link already exists"""
cursor = self.connection.cursor()
cursor.execute(
"""SELECT id FROM vehicle_diagrams
WHERE diagram_id = ? AND model_year_engine_id = ?""",
(diagram_id, model_year_engine_id)
)
result = cursor.fetchone()
return result[0] if result else None
def insert_vehicle_diagram(self, diagram_id: int, model_year_engine_id: int,
notes: str = None) -> int:
"""Link a diagram to a vehicle configuration"""
existing_id = self.get_vehicle_diagram(diagram_id, model_year_engine_id)
if existing_id:
return existing_id
cursor = self.connection.cursor()
cursor.execute(
"""INSERT INTO vehicle_diagrams (diagram_id, model_year_engine_id, notes)
VALUES (?, ?, ?)""",
(diagram_id, model_year_engine_id, notes)
)
self.connection.commit()
return cursor.lastrowid
def get_hotspot(self, diagram_id: int, callout_number: int) -> Optional[int]:
"""Check if a hotspot already exists for this diagram and callout"""
cursor = self.connection.cursor()
cursor.execute(
"""SELECT id FROM diagram_hotspots
WHERE diagram_id = ? AND callout_number = ?""",
(diagram_id, callout_number)
)
result = cursor.fetchone()
return result[0] if result else None
def insert_hotspot(self, diagram_id: int, part_id: int = None, callout_number: int = None,
label: str = None, shape: str = 'rect', coords: str = '',
color: str = '#e74c3c') -> int:
"""Insert a hotspot for a diagram"""
if callout_number:
existing_id = self.get_hotspot(diagram_id, callout_number)
if existing_id:
return existing_id
cursor = self.connection.cursor()
cursor.execute(
"""INSERT INTO diagram_hotspots (diagram_id, part_id, callout_number, label,
shape, coords, color)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(diagram_id, part_id, callout_number, label, shape, coords, color)
)
self.connection.commit()
return cursor.lastrowid
def get_part_by_name_pattern(self, pattern: str) -> Optional[Dict]:
"""Get a part by name pattern"""
cursor = self.connection.cursor()
cursor.execute(
"SELECT id, oem_part_number, name FROM parts WHERE name LIKE ?",
(f"%{pattern}%",)
)
result = cursor.fetchone()
return dict(result) if result else None
def get_group_by_name(self, name: str) -> Optional[int]:
"""Get group ID by name"""
cursor = self.connection.cursor()
cursor.execute("SELECT id FROM part_groups WHERE name LIKE ?", (f"%{name}%",))
result = cursor.fetchone()
return result[0] if result else None
def get_all_model_year_engines(self) -> List[int]:
"""Get all model_year_engine IDs"""
cursor = self.connection.cursor()
cursor.execute("SELECT id FROM model_year_engine")
return [row[0] for row in cursor.fetchall()]
# SVG Templates for diagrams
def generate_brake_assembly_svg() -> str:
"""Generate SVG for front brake assembly diagram"""
return '''
'''
def generate_oil_filter_system_svg() -> str:
"""Generate SVG for oil filter system diagram"""
return '''
'''
def generate_suspension_diagram_svg() -> str:
"""Generate SVG for front suspension diagram"""
return '''
'''
def save_svg_file(filename: str, content: str):
"""Save SVG content to file"""
filepath = os.path.join(DIAGRAMS_DIR, filename)
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
print(f" Saved SVG file: {filepath}")
return filepath
def populate_diagrams(manager: Fase3Manager) -> Dict[str, int]:
"""Populate the diagrams table with sample diagrams"""
print("\n=== Populating Diagrams ===")
diagram_ids = {}
# Get group IDs for different diagram types
brake_rotor_group = manager.get_group_by_name("Brake Rotors")
oil_filter_group = manager.get_group_by_name("Oil Filters")
struts_group = manager.get_group_by_name("Struts")
# Use fallback group IDs if not found
if not brake_rotor_group:
brake_rotor_group = 17 # Default Brake Rotors group
if not oil_filter_group:
oil_filter_group = 70 # Default Oil Filters group
if not struts_group:
struts_group = 157 # Default Struts group
diagrams_data = [
{
'name': 'Front Brake Assembly',
'name_es': 'Ensamble de Freno Delantero',
'group_id': brake_rotor_group,
'image_path': 'diagrams/brake_assembly.svg',
'svg_generator': generate_brake_assembly_svg,
'svg_filename': 'brake_assembly.svg',
'source': 'System Generated'
},
{
'name': 'Oil Filter System',
'name_es': 'Sistema de Filtro de Aceite',
'group_id': oil_filter_group,
'image_path': 'diagrams/oil_filter_system.svg',
'svg_generator': generate_oil_filter_system_svg,
'svg_filename': 'oil_filter_system.svg',
'source': 'System Generated'
},
{
'name': 'Front Suspension Assembly',
'name_es': 'Ensamble de Suspension Delantera',
'group_id': struts_group,
'image_path': 'diagrams/suspension_assembly.svg',
'svg_generator': generate_suspension_diagram_svg,
'svg_filename': 'suspension_assembly.svg',
'source': 'System Generated'
}
]
for diagram_data in diagrams_data:
# Generate SVG content
svg_content = diagram_data['svg_generator']()
# Save SVG file
save_svg_file(diagram_data['svg_filename'], svg_content)
# Insert diagram record
diagram_id = manager.insert_diagram(
name=diagram_data['name'],
name_es=diagram_data['name_es'],
group_id=diagram_data['group_id'],
image_path=diagram_data['image_path'],
thumbnail_path=None,
svg_content=svg_content,
width=600,
height=400,
display_order=0,
source=diagram_data['source']
)
diagram_ids[diagram_data['name']] = diagram_id
print(f"\nTotal diagrams created: {len(diagram_ids)}")
return diagram_ids
def populate_vehicle_diagrams(manager: Fase3Manager, diagram_ids: Dict[str, int]):
"""Link diagrams to vehicle configurations"""
print("\n=== Linking Diagrams to Vehicles ===")
# Get all model_year_engine entries
mye_ids = manager.get_all_model_year_engines()
if not mye_ids:
print("No vehicle configurations found. Skipping vehicle-diagram links.")
return
total_links = 0
# Link each diagram to all vehicle configurations (diagrams are generic)
for diagram_name, diagram_id in diagram_ids.items():
for mye_id in mye_ids:
manager.insert_vehicle_diagram(
diagram_id=diagram_id,
model_year_engine_id=mye_id,
notes=f"Standard {diagram_name.lower()} diagram"
)
total_links += 1
print(f" Created {total_links} vehicle-diagram links")
def populate_hotspots(manager: Fase3Manager, diagram_ids: Dict[str, int]):
"""Create hotspots for each diagram linking to actual parts"""
print("\n=== Creating Diagram Hotspots ===")
total_hotspots = 0
# Hotspots for Brake Assembly diagram
if 'Front Brake Assembly' in diagram_ids:
diagram_id = diagram_ids['Front Brake Assembly']
print(f"\n Creating hotspots for Front Brake Assembly (ID: {diagram_id})")
# Find parts to link
brake_rotor = manager.get_part_by_name_pattern("Brake Rotor")
brake_pads = manager.get_part_by_name_pattern("Brake Pads")
brake_caliper = manager.get_part_by_name_pattern("Caliper")
# Hotspot 1: Brake Rotor - rectangle around the rotor area
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=brake_rotor['id'] if brake_rotor else None,
callout_number=1,
label="Brake Rotor / Disco de Freno",
shape='circle',
coords='250,200,120', # cx,cy,r for circle
color='#3498db'
)
total_hotspots += 1
print(f" Added hotspot 1: Brake Rotor")
# Hotspot 2: Brake Caliper - rectangle around caliper
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=brake_caliper['id'] if brake_caliper else None,
callout_number=2,
label="Brake Caliper / Calibrador",
shape='rect',
coords='320,140,80,120', # x,y,width,height for rect
color='#e74c3c'
)
total_hotspots += 1
print(f" Added hotspot 2: Brake Caliper")
# Hotspot 3: Brake Pads - rectangle around pad area
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=brake_pads['id'] if brake_pads else None,
callout_number=3,
label="Brake Pads / Balatas",
shape='rect',
coords='300,160,120,80', # x,y,width,height
color='#8b7355'
)
total_hotspots += 1
print(f" Added hotspot 3: Brake Pads")
# Hotspots for Oil Filter System diagram
if 'Oil Filter System' in diagram_ids:
diagram_id = diagram_ids['Oil Filter System']
print(f"\n Creating hotspots for Oil Filter System (ID: {diagram_id})")
# Find oil filter part
oil_filter = manager.get_part_by_name_pattern("Oil Filter")
# Hotspot 1: Oil Filter
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=oil_filter['id'] if oil_filter else None,
callout_number=1,
label="Oil Filter / Filtro de Aceite",
shape='rect',
coords='320,140,80,120', # x,y,width,height
color='#2980b9'
)
total_hotspots += 1
print(f" Added hotspot 1: Oil Filter")
# Hotspots for Suspension Assembly diagram
if 'Front Suspension Assembly' in diagram_ids:
diagram_id = diagram_ids['Front Suspension Assembly']
print(f"\n Creating hotspots for Front Suspension Assembly (ID: {diagram_id})")
# Find parts
strut = manager.get_part_by_name_pattern("Strut")
ball_joint = manager.get_part_by_name_pattern("Ball Joint")
control_arm = manager.get_part_by_name_pattern("Control Arm")
# Hotspot 1: Strut Assembly
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=strut['id'] if strut else None,
callout_number=1,
label="Strut Assembly / Amortiguador",
shape='rect',
coords='275,95,50,150', # x,y,width,height
color='#27ae60'
)
total_hotspots += 1
print(f" Added hotspot 1: Strut Assembly")
# Hotspot 2: Ball Joint
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=ball_joint['id'] if ball_joint else None,
callout_number=2,
label="Ball Joint / Rotula",
shape='circle',
coords='300,280,20', # cx,cy,r
color='#c0392b'
)
total_hotspots += 1
print(f" Added hotspot 2: Ball Joint")
# Hotspot 3: Control Arm
manager.insert_hotspot(
diagram_id=diagram_id,
part_id=control_arm['id'] if control_arm else None,
callout_number=3,
label="Control Arm / Brazo de Control",
shape='poly',
coords='150,320,300,280,450,320,430,340,300,310,170,340', # polygon points
color='#444'
)
total_hotspots += 1
print(f" Added hotspot 3: Control Arm")
print(f"\nTotal hotspots created: {total_hotspots}")
def main():
"""Main entry point for FASE 3 population"""
print("=" * 60)
print("FASE 3: Exploded Diagrams and Hotspots Population")
print("=" * 60)
manager = Fase3Manager()
try:
# Connect to database
manager.connect()
# Create FASE 3 tables (idempotent)
manager.create_fase3_tables()
# Create diagrams directory
manager.create_diagrams_directory()
# Populate diagrams
diagram_ids = populate_diagrams(manager)
# Link diagrams to vehicles
populate_vehicle_diagrams(manager, diagram_ids)
# Create hotspots
populate_hotspots(manager, diagram_ids)
print("\n" + "=" * 60)
print("FASE 3 population completed successfully!")
print("=" * 60)
# Print summary
cursor = manager.connection.cursor()
cursor.execute("SELECT COUNT(*) FROM diagrams")
diagram_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM vehicle_diagrams")
vd_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM diagram_hotspots")
hotspot_count = cursor.fetchone()[0]
print(f"\nSummary:")
print(f" Diagrams: {diagram_count}")
print(f" Vehicle-Diagram Links: {vd_count}")
print(f" Hotspots: {hotspot_count}")
print(f"\nSVG files saved to: {DIAGRAMS_DIR}")
except Exception as e:
print(f"\nError: {e}")
raise
finally:
manager.disconnect()
if __name__ == "__main__":
main()