SVG/DWG Parsing Workflows for Indoor Mapping Pipelines

import math
import json
import logging
from pathlib import Path
from typing import List, Dict, Any, Tuple
from dataclasses import dataclass, field

import ezdxf
from shapely.geometry import Polygon, LineString, shape, mapping
from shapely.validation import make_valid
from lxml import etree
import numpy as np

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

@dataclass
class UnifiedGeometry:
    geom_type: str
    coordinates: List[Any]
    properties: Dict[str, Any]
    source_file: str

def dwg_unit_to_meters(dxf_units: int) -> float:
    """Map DXF unit codes to meters. Based on AutoCAD standard unit codes."""
    unit_map = {
        0: 1.0,   # Unitless
        1: 0.0254, # Inches
        2: 1.0,   # Feet
        3: 1609.344, # Miles
        4: 0.001, # Millimeters
        5: 0.01,  # Centimeters
        6: 1.0,   # Meters
        7: 1000.0, # Kilometers
    }
    return unit_map.get(dxf_units, 1.0)

def parse_dwg_entities(filepath: str) -> List[UnifiedGeometry]:
    doc = ezdxf.readfile(filepath)
    msp = doc.modelspace()
    unit_scale = dwg_unit_to_meters(doc.header.get('$INSUNITS', 0))
    
    geometries = []
    for entity in msp:
        if entity.dxftype() in ('LWPOLYLINE', 'POLYLINE'):
            points = [(p[0] * unit_scale, p[1] * unit_scale) for p in entity.get_points()]
            if entity.is_closed and len(points) >= 3:
                poly = Polygon(points)
                if poly.is_valid and poly.area > 0.1:
                    geometries.append(UnifiedGeometry(
                        geom_type="Polygon",
                        coordinates=list(poly.exterior.coords),
                        properties={"layer": entity.dxf.layer, "type": "room_or_corridor"},
                        source_file=filepath
                    ))
        elif entity.dxftype() == 'LINE':
            p1 = (entity.dxf.start.x * unit_scale, entity.dxf.start.y * unit_scale)
            p2 = (entity.dxf.end.x * unit_scale, entity.dxf.end.y * unit_scale)
            geometries.append(UnifiedGeometry(
                geom_type="LineString",
                coordinates=[p1, p2],
                properties={"layer": entity.dxf.layer, "type": "wall_or_edge"},
                source_file=filepath
            ))
    return geometries

def apply_svg_transform(points: List[Tuple[float, float]], matrix: np.ndarray) -> List[Tuple[float, float]]:
    """Apply 3x3 affine transform matrix to SVG coordinate points."""
    pts = np.array([(x, y, 1) for x, y in points])
    transformed = (matrix @ pts.T).T
    return [(x, y) for x, y in transformed[:, :2]]

def parse_svg_path(filepath: str, pixel_to_meter: float = 0.001) -> List[UnifiedGeometry]:
    parser = etree.XMLParser(resolve_entities=False)
    tree = etree.parse(filepath, parser)
    root = tree.getroot()
    
    # Default SVG viewBox parsing
    vb = root.get('viewBox', '0 0 1000 1000').split()
    vb_x, vb_y, vb_w, vb_h = map(float, vb)
    scale_x = pixel_to_meter
    scale_y = -pixel_to_meter  # Invert Y for GIS alignment
    
    geometries = []
    for path in root.iter('{http://www.w3.org/2000/svg}path'):
        d = path.get('d', '')
        if not d: continue
        
        # Simplified path parser for M, L, Z commands
        coords = []
        for cmd in d.replace(',', ' ').split():
            try:
                coords.append(float(cmd))
            except ValueError:
                continue
                
        if len(coords) >= 4:
            raw_pts = [(coords[i], coords[i+1]) for i in range(0, len(coords)-1, 2)]
            # Apply scale and Y-inversion
            transformed = [(x * scale_x, y * scale_y) for x, y in raw_pts]
            
            if len(transformed) >= 3:
                poly = Polygon(transformed)
                if poly.is_valid and poly.area > 0.1:
                    geometries.append(UnifiedGeometry(
                        geom_type="Polygon",
                        coordinates=list(poly.exterior.coords),
                        properties={"layer": path.get('class', 'default'), "type": "svg_room"},
                        source_file=filepath
                    ))
    return geometries

def build_geojson(geometries: List[UnifiedGeometry]) -> Dict[str, Any]:
    features = []
    for g in geometries:
        if g.geom_type == "Polygon":
            raw_geom = {"type": "Polygon", "coordinates": [g.coordinates]}
        else:
            raw_geom = {"type": "LineString", "coordinates": g.coordinates}

        # Repair invalid geometries via shapely, then emit GeoJSON-compatible dict.
        repaired = make_valid(shape(raw_geom))
        features.append({
            "type": "Feature",
            "geometry": mapping(repaired),
            "properties": g.properties
        })
    return {"type": "FeatureCollection", "features": features}

if __name__ == "__main__":
    # Example pipeline execution
    input_files = ["floorplan.dwg", "floorplan_export.svg"]
    all_geoms = []
    
    for f in input_files:
        if not Path(f).exists(): continue
        if f.endswith('.dwg'):
            all_geoms.extend(parse_dwg_entities(f))
        elif f.endswith('.svg'):
            all_geoms.extend(parse_svg_path(f))
            
    output = build_geojson(all_geoms)
    Path("normalized_indoor_map.geojson").write_text(json.dumps(output, indent=2))
    logging.info(f"Exported {len(output['features'])} validated features to GeoJSON.")