SVG/DWG Parsing Workflows for Indoor Mapping Pipelines

import logging
from typing import Dict

logger = logging.getLogger("parse.dwg.units")

# DXF $INSUNITS codes -> metres-per-unit (AutoCAD standard subset).
_UNIT_SCALE: Dict[int, float] = {
    0: 1.0,        # Unitless — assume 1:1
    1: 0.0254,     # Inches
    2: 0.3048,     # Feet
    4: 0.001,      # Millimetres
    5: 0.01,       # Centimetres
    6: 1.0,        # Metres
    7: 1000.0,     # Kilometres
}

def dwg_unit_to_metres(dxf_units: int) -> float:
    """Map a DXF $INSUNITS code to a metres-per-unit scale factor."""
    scale = _UNIT_SCALE.get(dxf_units)
    if scale is None:
        logger.warning("Unknown $INSUNITS code %s; defaulting to 1:1", dxf_units)
        return 1.0
    return scale

import logging
from dataclasses import dataclass
from typing import Any, Dict, List

import ezdxf
from ezdxf.lldxf.const import DXFStructureError
from shapely.geometry import Polygon

logger = logging.getLogger("parse.dwg")

@dataclass
class UnifiedGeometry:
    geom_type: str               # "Polygon" | "LineString"
    coordinates: List[Any]
    properties: Dict[str, Any]
    source_file: str

def parse_dwg_entities(filepath: str) -> List[UnifiedGeometry]:
    """Read a DWG/DXF file and return metric, Y-up unified geometry records."""
    try:
        doc = ezdxf.readfile(filepath)
    except IOError as exc:                       # file missing / unreadable
        logger.exception("Cannot open DWG %s: %s", filepath, exc)
        raise
    except DXFStructureError as exc:             # corrupt or invalid DXF/DWG
        logger.exception("Malformed DWG structure in %s: %s", filepath, exc)
        raise

    scale = dwg_unit_to_metres(doc.header.get("$INSUNITS", 0))
    geometries: List[UnifiedGeometry] = []

    for entity in doc.modelspace():
        etype = entity.dxftype()
        if etype in ("LWPOLYLINE", "POLYLINE"):
            if etype == "LWPOLYLINE":
                raw = [(p[0], p[1]) for p in entity.get_points()]
            else:
                raw = [(v.dxf.location.x, v.dxf.location.y) for v in entity.vertices]
            pts = [(x * scale, y * scale) for x, y in raw]
            if entity.is_closed and len(pts) >= 3:
                poly = Polygon(pts)
                if poly.is_valid and poly.area > 0.1:   # drop sub-0.1 m² slivers
                    geometries.append(UnifiedGeometry(
                        "Polygon", list(poly.exterior.coords),
                        {"floor_layer": entity.dxf.layer, "space_class": "room",
                         "source_format": "dwg"}, filepath))
        elif etype == "LINE":
            p1 = (entity.dxf.start.x * scale, entity.dxf.start.y * scale)
            p2 = (entity.dxf.end.x * scale, entity.dxf.end.y * scale)
            geometries.append(UnifiedGeometry(
                "LineString", [p1, p2],
                {"floor_layer": entity.dxf.layer, "space_class": "wall",
                 "source_format": "dwg"}, filepath))

    logger.info("Extracted %d entities from %s (scale=%.4f m/unit)",
                len(geometries), filepath, scale)
    return geometries

import logging
from typing import List, Tuple

import numpy as np
from lxml import etree

logger = logging.getLogger("parse.svg")

def apply_affine(points: List[Tuple[float, float]], matrix: np.ndarray) -> np.ndarray:
    """Apply a 3x3 homogeneous affine matrix to a list of (x, y) points."""
    homo = np.array([(x, y, 1.0) for x, y in points])
    return (matrix @ homo.T).T[:, :2]

def parse_svg_paths(filepath: str, px_per_metre: float = 1000.0) -> List["UnifiedGeometry"]:
    """Parse SVG <path> M/L/Z geometry into metric, Y-up unified records."""
    try:
        # resolve_entities=False neutralises XXE in untrusted SVG exports.
        tree = etree.parse(filepath, etree.XMLParser(resolve_entities=False))
    except (OSError, etree.XMLSyntaxError) as exc:   # missing file / malformed XML
        logger.exception("Cannot parse SVG %s: %s", filepath, exc)
        raise

    root = tree.getroot()
    _, _, _, vb_h = map(float, root.get("viewBox", "0 0 1000 1000").split())
    ns = "{http://www.w3.org/2000/svg}"
    geometries: List[UnifiedGeometry] = []

    for path in root.iter(f"{ns}path"):
        d = path.get("d", "")
        nums = [float(t) for t in d.replace(",", " ").split()
                if _is_float(t)]
        if len(nums) < 6:
            continue
        raw = [(nums[i], nums[i + 1]) for i in range(0, len(nums) - 1, 2)]
        # Flip Y against the viewBox, then scale pixels -> metres.
        metric = [((x) / px_per_metre, (vb_h - y) / px_per_metre) for x, y in raw]
        if len(metric) >= 3:
            geometries.append(UnifiedGeometry(
                "Polygon", metric,
                {"floor_layer": path.get("class", "default"),
                 "space_class": "room", "source_format": "svg"}, filepath))

    logger.info("Extracted %d paths from %s (px/m=%.1f)",
                len(geometries), filepath, px_per_metre)
    return geometries

def _is_float(token: str) -> bool:
    try:
        float(token)
        return True
    except ValueError:
        return False

import hashlib
import json
import logging
from pathlib import Path
from typing import Any, Dict, List

from shapely.errors import GEOSException
from shapely.geometry import mapping, shape
from shapely.validation import make_valid

logger = logging.getLogger("parse.unify")

def build_feature_collection(geoms: List["UnifiedGeometry"]) -> Dict[str, Any]:
    """Validate, hash, and serialise unified geometry into a GeoJSON FeatureCollection."""
    features: List[Dict[str, Any]] = []
    for g in geoms:
        raw = ({"type": "Polygon", "coordinates": [g.coordinates]}
               if g.geom_type == "Polygon"
               else {"type": "LineString", "coordinates": g.coordinates})
        try:
            repaired = make_valid(shape(raw))
        except GEOSException as exc:              # degenerate / non-noded geometry
            logger.warning("Dropping unrepairable geometry from %s: %s",
                           g.source_file, exc)
            continue
        geom_json = mapping(repaired)
        props = dict(g.properties)
        props["topology_hash"] = hashlib.sha1(
            json.dumps(geom_json["coordinates"]).encode()).hexdigest()[:8]
        props.setdefault("is_routable", True)
        features.append({"type": "Feature", "geometry": geom_json, "properties": props})

    logger.info("Built FeatureCollection with %d validated features", len(features))
    return {"type": "FeatureCollection", "features": features}

def parse_floor_plan(filepath: str) -> Dict[str, Any]:
    """Dispatch a single file to the correct decoder and return GeoJSON."""
    suffix = Path(filepath).suffix.lower()
    if suffix in (".dwg", ".dxf"):
        geoms = parse_dwg_entities(filepath)
    elif suffix == ".svg":
        geoms = parse_svg_paths(filepath)
    else:
        raise ValueError(f"Unsupported floor-plan format: {suffix!r}")
    return build_feature_collection(geoms)

{
  "type": "FeatureCollection",
  "features": [
    {
      "type": "Feature",
      "geometry": { "type": "Polygon", "coordinates": [[[0,0],[6,0],[6,4],[0,4],[0,0]]] },
      "properties": {
        "floor_layer": "A-ROOM", "space_class": "room",
        "source_format": "dwg", "is_routable": true, "topology_hash": "9f3c2a7e"
      }
    }
  ]
}

from shapely.geometry import shape

def assert_metric_sanity(fc: dict, max_span_m: float = 500.0) -> None:
    """Fail fast if a FeatureCollection escaped the metric, Y-up contract."""
    for feat in fc["features"]:
        geom = shape(feat["geometry"])
        minx, miny, maxx, maxy = geom.bounds
        assert max(maxx - minx, maxy - miny) < max_span_m, "unit mismatch — span too large"
        if geom.geom_type == "Polygon":
            # CCW exterior ring => positive signed area (OGC convention).
            assert geom.exterior.is_ccw, "ring orientation inverted — check Y-axis flip"