Web app and API

satprint.app

FastAPI backend: terrain lookup, model generation, STL and GLB download.

BBoxIn

Bases: BaseModel

A bounding box in degrees, as the API receives it.

Job(id, status='running', stage='starting', done=0, total=0, started=time.time(), result=None, error=None, status_code=None) dataclass

A background model build and how far it has got.

as_dict()

The job as the JSON body of GET /api/jobs/{job_id}.

Source code in satprint/app.py

def as_dict(self) -> dict:
    """The job as the JSON body of ``GET /api/jobs/{job_id}``."""
    return {
        "job_id": self.id,
        "status": self.status,
        "stage": self.stage,
        "done": self.done,
        "total": self.total,
        "elapsed_s": round(time.time() - self.started, 1),
        "result": self.result,
        "error": self.error,
        "status_code": self.status_code,
    }

ModelRequest

Bases: BaseModel

Body of POST /api/model and POST /api/jobs: what to build and how.

ModelResponse

Bases: BaseModel

A built model: download URLs, build info and an inline preview.

StoredModel(stl, png, info, created, glb=None, threemf=None) dataclass

A built model's files and info, kept for download by model id.

create_app(tile_fetcher=None, imagery_fetcher=None, overpass=None, geocoder=None, vector_tiles=None)

Build the FastAPI app: the API routes and the static web page.

Each client defaults to a real one; tests pass fakes.

Parameters:

Name	Type	Description	Default
tile_fetcher	TileFetcher | None	elevation tile source.	None
imagery_fetcher	ImageryFetcher | None	satellite imagery tile source.	None
overpass	OverpassClient | None	Overpass client for buildings, bridges and landmarks.	None
geocoder	Geocoder | None	place search for /api/search.	None
vector_tiles	VectorTileClient | None	OpenFreeMap vector tiles for buildings, water and roads.	None

Source code in satprint/app.py

def create_app(
    tile_fetcher: TileFetcher | None = None,
    imagery_fetcher: ImageryFetcher | None = None,
    overpass: OverpassClient | None = None,
    geocoder: Geocoder | None = None,
    vector_tiles: VectorTileClient | None = None,
) -> FastAPI:
    """Build the FastAPI app: the API routes and the static web page.

    Each client defaults to a real one; tests pass fakes.

    :param tile_fetcher: elevation tile source.
    :param imagery_fetcher: satellite imagery tile source.
    :param overpass: Overpass client for buildings, bridges and landmarks.
    :param geocoder: place search for ``/api/search``.
    :param vector_tiles: OpenFreeMap vector tiles for buildings, water and roads.
    """
    app = FastAPI(
        title="satprint",
        version=__version__,
        description="Turn satellite elevation data into 3D-printable terrain models.",
    )
    fetcher = tile_fetcher or TileFetcher()
    imagery = imagery_fetcher or ImageryFetcher()
    osm = overpass or OverpassClient()
    vtiles = vector_tiles or VectorTileClient()

    def _water_for(bbox: BBox, progress: Progress | None = None) -> list:
        """Water polygons in ``bbox``, cached."""
        key = _bbox_key(bbox)
        hit = water_cache.get(key)
        if hit is None:
            tiles = vtiles.tiles(
                bbox, progress=progress, zoom=water_zoom(bbox), stage="water"
            )
            hit = water_from_vector_tiles(tiles)
            water_cache.put(key, hit)
        return hit

    def _bridges_for(
        bbox: BBox, progress: Progress | None = None
    ) -> tuple[tuple[list, list], list]:
        """Bridge outlines and lines, and the water polygons, in ``bbox``."""
        key = _bbox_key(bbox)
        hit = bridges_cache.get(key)
        if hit is None:
            tiles = vtiles.tiles(bbox, progress=progress, stage="bridges")
            hit = (bridges_from_vector_tiles(tiles), water_from_vector_tiles(tiles))
            bridges_cache.put(key, hit)
        return hit

    def _roads_for(bbox: BBox, progress: Progress | None = None) -> tuple[list, list]:
        """Road lines and water polygons in ``bbox``."""
        key = _bbox_key(bbox)
        hit = roads_cache.get(key)
        if hit is None:
            tiles = vtiles.tiles(bbox, progress=progress, stage="roads")
            hit = (roads_from_vector_tiles(tiles), water_from_vector_tiles(tiles))
            roads_cache.put(key, hit)
        return hit

    geo = geocoder or Geocoder()

    def _bbox_key(bbox: BBox) -> tuple:
        """Cache key for ``bbox``, rounded to about a meter."""
        return tuple(
            round(v, 5) for v in (bbox.south, bbox.west, bbox.north, bbox.east)
        )

    def _buildings_for(
        bbox: BBox, source: str, progress: Progress | None = None
    ) -> tuple[list, str | None, str]:
        """Buildings in ``bbox``, a roof-shape warning, and the source used."""
        order = {"auto": ["openfreemap", "overpass"]}.get(source, [source])
        errors = []
        for name in order:
            key = (name, *_bbox_key(bbox))
            hit = buildings_cache.get(key)
            if hit is not None:
                return *hit, name
            try:
                hit = fetch_buildings(name, bbox, osm, vtiles, progress)
            except Exception as exc:
                errors.append(f"{name}: {exc}")
                continue
            buildings_cache.put(key, hit)
            return *hit, name
        raise RuntimeError("; ".join(errors))

    def _texture_for(
        bbox: BBox, progress: Progress | None = None
    ) -> tuple[bytes, dict]:
        """Satellite imagery of ``bbox`` as JPEG bytes and its metadata, cached."""
        key = _bbox_key(bbox)
        hit = imagery_cache.get(key)
        if hit is None:
            img, meta = fetch_imagery(bbox, fetcher=imagery, progress=progress)
            buf = io.BytesIO()
            img.save(buf, format="JPEG", quality=90)
            hit = (buf.getvalue(), meta)
            imagery_cache.put(key, hit)
        return hit

    def _terrain_for(req: ModelRequest, progress: Progress | None = None) -> Heightmap:
        """The heightmap ``req`` asks for: synthetic, uploaded or downloaded.

        :raises HTTPException: for a missing bbox or upload, or a failed download.
        """
        if req.source == "synthetic":
            key = ("synthetic", req.seed, req.resolution)
            hm = terrain_cache.get(key)
            if hm is None:
                hm = synthetic_heightmap(
                    rows=int(req.resolution * 0.75), cols=req.resolution, seed=req.seed
                )
                terrain_cache.put(key, hm)
            return hm
        if req.source == "upload":
            if not req.upload_id:
                raise HTTPException(400, "upload_id is required for source=upload")
            hm = upload_store.get(req.upload_id)
            if hm is None:
                raise HTTPException(
                    404, "upload not found (server restarted?) — upload the file again"
                )
            if req.ground_width_m:
                hm = Heightmap(
                    hm.data,
                    req.ground_width_m,
                    req.ground_width_m * hm.data.shape[0] / hm.data.shape[1],
                    hm.source,
                    hm.bbox,
                    hm.meta,
                )
            return hm
        if req.bbox is None:
            raise HTTPException(400, "bbox is required for source=terrarium")
        try:
            bbox = BBox(**req.bbox.model_dump())
        except ValueError as exc:
            raise HTTPException(400, str(exc)) from exc
        key = (
            "terrarium",
            round(bbox.south, 5),
            round(bbox.west, 5),
            round(bbox.north, 5),
            round(bbox.east, 5),
            req.resolution,
        )
        hm = terrain_cache.get(key)
        if hm is None:
            try:
                hm = fetch_terrarium(
                    bbox, target_cols=req.resolution, fetcher=fetcher, progress=progress
                )
            except ValueError as exc:
                raise HTTPException(400, str(exc)) from exc
            except LookupError as exc:
                raise HTTPException(404, str(exc)) from exc
            except Exception as exc:  # network / decode failures
                raise HTTPException(502, f"elevation download failed: {exc}") from exc
            try:  # data over water is noisy; levelling refines it, so a failure keeps it
                hm = level_water(
                    hm,
                    vtiles.tiles(
                        bbox,
                        progress=progress,
                        zoom=water_zoom(bbox),
                        stage="water",
                    ),
                )
            except Exception:
                pass
            terrain_cache.put(key, hm)
        return hm

    @app.get("/api/presets")
    def presets():
        """Named example areas for the place picker."""
        return PRESETS

    @app.get("/api/search")
    def search(q: str = Query(..., min_length=2, max_length=200)):
        """Places matching ``q``, from Nominatim."""
        try:
            return geo.search(q)
        except Exception as exc:
            raise HTTPException(502, f"place search failed: {exc}") from exc

    @app.get("/api/health")
    def health():
        """Liveness check with the running version."""
        return {"ok": True, "version": __version__}

    @app.post("/api/upload")
    async def upload(file: UploadFile = File(...)):
        """Store an uploaded heightmap file for ``source=upload``; 60 MB at most."""
        data = await file.read()
        if len(data) > 60 * 1024 * 1024:
            raise HTTPException(413, "file too large (60 MB limit)")
        try:
            hm = load_heightmap_file(data, file.filename or "")
        except Exception as exc:
            raise HTTPException(400, f"could not read heightmap: {exc}") from exc
        upload_id = uuid.uuid4().hex[:12]
        upload_store.put(upload_id, hm)
        return {
            "upload_id": upload_id,
            "rows": hm.shape[0],
            "cols": hm.shape[1],
            "min_elev_m": hm.min,
            "max_elev_m": hm.max,
            "ground_width_m": hm.ground_width_m,
            "ground_height_m": hm.ground_height_m,
            "meta": hm.meta,
        }

    def _build(req: ModelRequest, progress: Progress | None = None) -> ModelResponse:
        """Build a model; ``progress`` hears each stage as it advances."""
        hm = _terrain_for(req, progress)
        try:
            params = PrintParams(
                width_mm=req.width_mm,
                base_mm=req.base_mm,
                exaggeration=req.exaggeration,
                relief_mm=req.relief_mm,
                smoothing=req.smoothing,
                clamp_sea_level=req.clamp_sea_level,
            )
        except ValueError as exc:
            raise HTTPException(400, str(exc)) from exc
        t0 = time.perf_counter()
        if progress:
            progress("terrain mesh", 0, 0)
        relief, info = prepare_relief(hm, params)
        mesh: Mesh = heightmap_to_mesh(
            relief, info["width_mm"], info["depth_mm"], params.base_mm
        )
        frame: Mesh | None = None
        body = mesh  # terrain plus frame: what the STL and GLB carry
        if req.frame_mm > 0:
            frame_h = req.frame_height_mm or params.base_mm + 1.0
            frame = frame_mesh(
                info["width_mm"], info["depth_mm"], req.frame_mm, frame_h
            )
            body = merge_meshes(mesh, frame)
            info.update(
                {
                    "frame_mm": req.frame_mm,
                    "frame_height_mm": frame_h,
                    "outer_width_mm": info["width_mm"] + 2 * req.frame_mm,
                    "outer_depth_mm": info["depth_mm"] + 2 * req.frame_mm,
                    "height_mm": max(info["height_mm"], frame_h),
                }
            )
        rows, cols = relief.shape
        bmesh: BuildingMesh | None = None
        footprints: list = []  # roads stop at the buildings
        building_info: dict = {}
        if req.buildings and hm.bbox is not None:
            area = bbox_area_km2(hm.bbox)
            # Like the texture, a building failure should not cost the STL.
            if area > MAX_BUILDING_AREA_KM2:
                building_info["building_error"] = (
                    f"area is {area:.0f} km2; buildings are limited to "
                    f"{MAX_BUILDING_AREA_KM2:.0f} km2"
                )
            else:
                try:
                    found, warning, used = _buildings_for(
                        hm.bbox, req.building_source, progress
                    )
                except Exception as exc:
                    building_info["building_error"] = f"building download failed: {exc}"
                else:
                    footprints = [b.footprint for b in found]
                    bmesh = building_mesh(
                        found,
                        hm.bbox,
                        relief,
                        info["width_mm"],
                        info["depth_mm"],
                        params.base_mm,
                        info["mm_per_m_plan"],
                        scale=req.building_scale,
                        progress=progress,
                    )
                    building_info = {
                        "buildings": bmesh.count,
                        "building_source": used,
                        "building_attribution": (
                            OVERTURE_ATTRIBUTION
                            if used == "overture"
                            else OSM_ATTRIBUTION
                        ),
                    }
                    if warning:
                        building_info["building_warning"] = warning
                    if bmesh.count:
                        top = float(bmesh.vertices[:, 2].max())
                        info["height_mm"] = max(info["height_mm"], top)
        brmesh: BuildingMesh | None = None
        if req.bridges and hm.bbox is not None:
            area = bbox_area_km2(hm.bbox)
            # A bridge failure should not cost the STL either.
            if area > MAX_BUILDING_AREA_KM2:
                building_info["bridge_error"] = (
                    f"area is {area:.0f} km2; bridges are limited to "
                    f"{MAX_BUILDING_AREA_KM2:.0f} km2"
                )
            else:
                try:
                    features, water = _bridges_for(hm.bbox, progress)
                except Exception as exc:
                    building_info["bridge_error"] = f"bridge download failed: {exc}"
                else:
                    if progress:
                        progress("bridge mesh", 0, 0)
                    brmesh = bridge_mesh(
                        *features,
                        water,
                        hm.bbox,
                        relief,
                        info["width_mm"],
                        info["depth_mm"],
                        params.base_mm,
                        info["mm_per_m_plan"],
                        pier_spacing_mm=req.bridge_piers_mm,
                    )
                    building_info["bridges"] = brmesh.count
                    if brmesh.count:
                        top = float(brmesh.vertices[:, 2].max())
                        info["height_mm"] = max(info["height_mm"], top)
        rmesh: BuildingMesh | None = None
        if req.roads and hm.bbox is not None:
            area = bbox_area_km2(hm.bbox)
            # Nor a road failure.
            if area > MAX_BUILDING_AREA_KM2:
                building_info["road_error"] = (
                    f"area is {area:.0f} km2; roads are limited to "
                    f"{MAX_BUILDING_AREA_KM2:.0f} km2"
                )
            else:
                try:
                    lines, water = _roads_for(hm.bbox, progress)
                except Exception as exc:
                    building_info["road_error"] = f"road download failed: {exc}"
                else:
                    if progress:
                        progress("road mesh", 0, 0)
                    rmesh = road_mesh(
                        lines,
                        water,
                        footprints,
                        hm.bbox,
                        relief,
                        info["width_mm"],
                        info["depth_mm"],
                        params.base_mm,
                        info["mm_per_m_plan"],
                        detail=req.road_detail,
                        bridges=brmesh,
                    )
                    building_info["roads"] = rmesh.count
                    if rmesh.count:
                        top = float(rmesh.vertices[:, 2].max())
                        info["height_mm"] = max(info["height_mm"], top)
        # buildings, bridges and roads: everything standing on the terrain
        extras = merge_building_meshes(*(m for m in (bmesh, brmesh, rmesh) if m))
        credit = building_info.get("building_attribution", OSM_ATTRIBUTION)
        if progress:
            progress("writing files", 0, 0)
        solid = merge_meshes(body, extras.as_mesh()) if extras.count else body
        stl = write_binary_stl(solid, name=req.name)
        png = heightmap_png(hm, params.clamp_sea_level)
        threemf = None
        multicolor_info: dict = {}
        if req.multicolor and hm.bbox is not None:
            polygons: list = []
            try:
                polygons = _water_for(hm.bbox, progress)
            except Exception as exc:
                # Coasts still color from the flattened sea; say why rivers
                # and lakes are missing.
                multicolor_info["water_error"] = f"water download failed: {exc}"
            parts, mask = multicolor_parts(
                polygons,
                hm.bbox,
                relief,
                info["width_mm"],
                info["depth_mm"],
                params.base_mm,
                sea_level_flat=params.clamp_sea_level and info["min_elev_m"] <= 0,
                buildings=bmesh,
                frame=frame,
                roads=rmesh,
                bridges=brmesh,
            )
            sources = [OSM_ATTRIBUTION] if polygons else []
            if bmesh and bmesh.count:
                sources.append(credit)
            if (rmesh and rmesh.count) or (brmesh and brmesh.count):
                sources.append(OSM_ATTRIBUTION)
            credits = "; ".join(dict.fromkeys(sources)) or None
            threemf = write_3mf(
                parts,
                name=req.name,
                attribution=credits,
                plate_center=plate_center(req.plate),
            )
            multicolor_info.update(
                {
                    "multicolor_parts": [p[0] for p in parts if p[1].faces.shape[0]],
                    "water_fraction": round(float(mask.mean()), 4),
                    "threemf_bytes": len(threemf),
                }
            )
        glb = None
        texture_info: dict = {"textured": False}
        if req.texture and hm.bbox is not None:
            # A missing texture should not cost the user their STL.
            try:
                jpeg, meta = _texture_for(hm.bbox, progress)
            except Exception as exc:
                texture_info["texture_error"] = f"imagery download failed: {exc}"
            else:
                credits = IMAGERY_ATTRIBUTION
                if bmesh and bmesh.count:
                    credits += "; " + credit
                osm = (rmesh and rmesh.count) or (brmesh and brmesh.count)
                if osm and OSM_ATTRIBUTION not in credits:
                    credits += "; " + OSM_ATTRIBUTION
                glb = write_glb(
                    body,
                    rows,
                    cols,
                    jpeg,
                    name=req.name,
                    copyright=credits,
                    buildings=extras,
                )
                texture_info = {
                    "textured": True,
                    "texture_meta": meta,
                    "glb_bytes": len(glb),
                    "texture_attribution": IMAGERY_ATTRIBUTION,
                }
        vol_cm3 = solid.volume_mm3() / 1000.0
        info.update(
            {
                "triangles": solid.triangle_count,
                "volume_cm3": vol_cm3,
                "est_weight_g_pla_solid": vol_cm3 * 1.24,
                "est_weight_g_pla_20pct": vol_cm3
                * 1.24
                * 0.35,  # shell + 20 % infill rule of thumb
                "stl_bytes": len(stl),
                "source": hm.source,
                "source_meta": hm.meta,
                "ground_width_m": hm.ground_width_m,
                "ground_height_m": hm.ground_height_m,
                "generate_seconds": round(time.perf_counter() - t0, 3),
                **texture_info,
                **building_info,
                **multicolor_info,
            }
        )
        model_id = uuid.uuid4().hex[:12]
        model_store.put(
            model_id,
            StoredModel(
                stl=stl,
                png=png,
                info=info,
                created=time.time(),
                glb=glb,
                threemf=threemf,
            ),
        )
        safe = (
            "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in req.name)
            or "terrain"
        )
        return ModelResponse(
            model_id=model_id,
            stl_url=f"/api/model/{model_id}/{safe}.stl",
            png_url=f"/api/model/{model_id}/heightmap.png",
            glb_url=f"/api/model/{model_id}/{safe}.glb" if glb else None,
            threemf_url=f"/api/model/{model_id}/{safe}.3mf" if threemf else None,
            info=info,
            preview_png="data:image/png;base64," + base64.b64encode(png).decode(),
        )

    @app.post("/api/model", response_model=ModelResponse)
    def make_model(req: ModelRequest):
        """Build a model and wait for it."""
        return _build(req)

    @app.post("/api/jobs")
    def start_job(req: ModelRequest):
        """Build a model in the background; poll ``GET /api/jobs/{id}``."""
        job = Job(id=uuid.uuid4().hex[:12])
        jobs.put(job.id, job)

        def report(stage: str, done: int, total: int) -> None:
            """Record build progress on the job."""
            job.stage, job.done, job.total = stage, done, total

        def run() -> None:
            """Build the model and record the result or the error on the job."""
            try:
                job.result = _build(req, report).model_dump()
                job.status = "done"
            except HTTPException as exc:
                job.error, job.status_code = str(exc.detail), exc.status_code
                job.status = "error"
            except Exception as exc:
                job.error, job.status_code = f"model build failed: {exc}", 500
                job.status = "error"

        threading.Thread(target=run, name=f"job-{job.id}", daemon=True).start()
        return {"job_id": job.id}

    @app.get("/api/jobs/{job_id}")
    def job_status(job_id: str):
        """Progress, and the result once done, of a background build."""
        job = jobs.get(job_id)
        if job is None:
            raise HTTPException(404, "job not found (server restarted?)")
        return job.as_dict()

    @app.get("/api/model/{model_id}/{filename}")
    def download(model_id: str, filename: str):
        """One file of a built model: PNG, STL, GLB, 3MF or ``info.json``."""
        m = model_store.get(model_id)
        if m is None:
            raise HTTPException(404, "model not found — generate it again")
        if filename.endswith(".png"):
            return Response(m.png, media_type="image/png")
        if filename.endswith(".stl"):
            return Response(
                m.stl,
                media_type="model/stl",
                headers={"Content-Disposition": f'attachment; filename="{filename}"'},
            )
        if filename.endswith(".glb") and m.glb is not None:
            return Response(
                m.glb,
                media_type="model/gltf-binary",
                headers={"Content-Disposition": f'attachment; filename="{filename}"'},
            )
        if filename.endswith(".3mf") and m.threemf is not None:
            return Response(
                m.threemf,
                media_type="model/3mf",
                headers={"Content-Disposition": f'attachment; filename="{filename}"'},
            )
        if filename == "info.json":
            return m.info
        raise HTTPException(404, "unknown file")

    @app.get("/", include_in_schema=False)
    def index():
        """The web page."""
        # The script and stylesheet URLs carry the version, so after an
        # upgrade the browser fetches them again instead of pairing the new
        # page with an old cached copy.
        with open(os.path.join(STATIC_DIR, "index.html"), encoding="utf-8") as fh:
            return HTMLResponse(fh.read().replace("__VERSION__", __version__))

    app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")

    @app.middleware("http")
    async def revalidate_page(request, call_next):
        """Have browsers check the page and its files each time, so a new
        version shows on a plain reload; unchanged files come back as 304."""
        response = await call_next(request)
        if request.url.path == "/" or request.url.path.startswith("/static/"):
            response.headers["Cache-Control"] = "no-cache"
        return response

    return app