DWG Floor Plan to Routable GeoJSON in One Python Pipeline

import logging

import ezdxf
from shapely.affinity import scale as shp_scale
from shapely.geometry import LineString

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def parse_dwg(dwg_path: str, layer_map: dict[str, str], unit_scale: float) -> dict[str, list[LineString]]:
    """Parse a DXF into class-bucketed, metre-scaled LineStrings, exploding blocks."""
    try:
        doc = ezdxf.readfile(dwg_path)
    except (IOError, ezdxf.DXFStructureError) as exc:
        logger.exception("parse failed for %s: %s", dwg_path, exc)
        raise

    msp = doc.modelspace()
    inserts = [e for ins in msp.query("INSERT") for e in ins.virtual_entities()]
    buckets: dict[str, list[LineString]] = {cls: [] for cls in set(layer_map.values())}

    for e in list(msp) + inserts:
        cls = layer_map.get(e.dxf.layer)
        if cls is None:
            continue
        if e.dxftype() == "LINE":
            geom = LineString([e.dxf.start[:2], e.dxf.end[:2]])
        elif e.dxftype() == "LWPOLYLINE":
            geom = LineString([(x, y) for x, y, *_ in e.get_points()])
        else:
            continue
        buckets[cls].append(shp_scale(geom, xfact=unit_scale, yfact=unit_scale, origin=(0, 0)))

    logger.info("parsed %s: %s", dwg_path, {k: len(v) for k, v in buckets.items()})
    return buckets

import logging

from shapely.geometry import LineString, Point, Polygon
from shapely.ops import polygonize, unary_union

logger = logging.getLogger(__name__)


def detect_spaces(
    walls: list[LineString], doors: list[LineString], snap_tol: float
) -> tuple[list[Polygon], list[Point]]:
    """Polygonize walls into rooms and reduce doors to opening centroids."""
    noded = unary_union(walls)                      # split walls at every intersection
    rooms = [p for p in polygonize(noded) if p.area > snap_tol]
    if not rooms:
        logger.warning("no closed rooms formed; check wall connectivity / snap_tol")

    door_points = [d.interpolate(0.5, normalized=True) for d in doors]
    logger.info("detected %d rooms, %d doors", len(rooms), len(door_points))
    return rooms, door_points

import logging

import networkx as nx
from shapely.geometry import Point, Polygon
from shapely.validation import make_valid

logger = logging.getLogger(__name__)


def build_and_validate(
    rooms: list[Polygon], doors: list[Point], snap_tol: float
) -> nx.Graph:
    """Build a room-adjacency routing graph and fail closed on a fragmented topology."""
    graph = nx.Graph()
    fixed = [make_valid(r) if not r.is_valid else r for r in rooms]
    for i, room in enumerate(fixed):
        graph.add_node(i, geometry=room, centroid=(room.centroid.x, room.centroid.y))

    for door in doors:
        touching = [i for i, r in enumerate(fixed) if r.distance(door) <= snap_tol]
        for a, b in zip(touching, touching[1:]):
            graph.add_edge(a, b, weight=door.distance(fixed[a].centroid))

    components = nx.number_connected_components(graph) if graph.number_of_nodes() else 0
    if components > 1:
        logger.error("routing graph fragmented into %d components", components)
        raise ValueError(f"topology check failed: {components} disconnected components")
    logger.info("graph ok: %d nodes, %d edges", graph.number_of_nodes(), graph.number_of_edges())
    return graph

import hashlib
import json
import logging

import networkx as nx

logger = logging.getLogger(__name__)


def serialize(graph: nx.Graph, level: int) -> dict:
    """Serialize graph nodes into the canonical GeoJSON FeatureCollection envelope."""
    features = []
    for node_id, data in graph.nodes(data=True):
        room = data["geometry"]
        coords = [list(room.exterior.coords)]
        topo = hashlib.sha256(room.wkb).hexdigest()[:8]
        features.append({
            "type": "Feature",
            "geometry": {"type": "Polygon", "coordinates": coords},
            "properties": {
                "feature_id": f"L{level}-room-{node_id:04d}",
                "space_class": "room",
                "level": level,
                "is_routable": graph.degree(node_id) > 0,
                "accessibility_rating": "step_free",
                "topology_hash": topo,
            },
        })

    collection = {"type": "FeatureCollection", "features": features}
    logger.info("serialized %d features for level %d", len(features), level)
    return collection


def run_pipeline(cfg: dict) -> dict:
    """Orchestrate the four stages end to end for one floor level."""
    buckets = parse_dwg(cfg["dwg_path"], cfg["layer_map"], cfg["unit_scale"])
    rooms, doors = detect_spaces(buckets.get("wall", []), buckets.get("door", []), cfg["snap_tol"])
    graph = build_and_validate(rooms, doors, cfg["snap_tol"])
    return serialize(graph, cfg["level"])