352 lines
15 KiB
Python
352 lines
15 KiB
Python
import logging
|
|
import math
|
|
import os
|
|
import tempfile
|
|
from collections.abc import Iterator
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from shapely.affinity import affine_transform
|
|
from shapely.geometry import LineString, box
|
|
from shapely.geometry.base import BaseGeometry
|
|
|
|
from flight_map.ahn import terrain_data
|
|
from flight_map.curves import Cubic
|
|
from flight_map.geo import Projection
|
|
from flight_map.lettering import lettering
|
|
from flight_map.maps import WaterMaskError, map_data, map_lines, water_areas
|
|
from flight_map.storage import database, get_session, load_observations
|
|
from flight_map.terrain import Elevation
|
|
from flight_map.tracks import make_tracks
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
SVG_NS = "http://www.w3.org/2000/svg"
|
|
ET.register_namespace("", SVG_NS)
|
|
|
|
|
|
def tag(name: str) -> str:
|
|
return f"{{{SVG_NS}}}{name}"
|
|
|
|
|
|
def lines(geometry: BaseGeometry) -> Iterator[LineString]:
|
|
if isinstance(geometry, LineString):
|
|
if geometry.length > 0:
|
|
yield geometry
|
|
elif hasattr(geometry, "geoms"):
|
|
for part in geometry.geoms:
|
|
yield from lines(part)
|
|
|
|
|
|
def add_line(group: ET.Element, line: LineString) -> None:
|
|
coordinates = list(line.coords)
|
|
path = "M " + " L ".join(f"{x:.5f},{y:.5f}" for x, y in coordinates)
|
|
ET.SubElement(group, tag("path"), {"d": path})
|
|
|
|
|
|
def add_curves(group: ET.Element, curves: list[Cubic]) -> None:
|
|
commands = []
|
|
previous = None
|
|
for curve in curves:
|
|
if all(point == curve.start for point in curve.points):
|
|
continue
|
|
if previous is None or math.dist(previous, curve.start) > 1e-7:
|
|
commands.append(f"M {curve.start[0]:.5f},{curve.start[1]:.5f}")
|
|
commands.append(
|
|
"C "
|
|
+ " ".join(
|
|
f"{x:.5f},{y:.5f}"
|
|
for x, y in (
|
|
curve.control1,
|
|
curve.control2,
|
|
curve.end,
|
|
)
|
|
)
|
|
)
|
|
previous = curve.end
|
|
if commands:
|
|
ET.SubElement(group, tag("path"), {"d": " ".join(commands)})
|
|
|
|
|
|
def label(group: ET.Element, text: str, x: float, y: float, height: float) -> None:
|
|
for line in lettering(text, x, y, height):
|
|
add_line(group, line)
|
|
|
|
|
|
def utc(timestamp: float) -> str:
|
|
return datetime.fromtimestamp(timestamp, UTC).strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlotOptions:
|
|
width: float = 297
|
|
height: float = 210
|
|
margin: float = 12
|
|
extension: float = 4
|
|
pen_width: float = 0.3
|
|
font_height: float = 1.8
|
|
max_gap: float = 120
|
|
|
|
def validate(self) -> None:
|
|
if not all(
|
|
math.isfinite(value) and value > 0
|
|
for value in (
|
|
self.width,
|
|
self.height,
|
|
self.margin,
|
|
self.extension,
|
|
self.pen_width,
|
|
self.font_height,
|
|
self.max_gap,
|
|
)
|
|
):
|
|
raise ValueError(
|
|
"Page dimensions, margins, pen width and plot settings must be positive."
|
|
)
|
|
if 2 * (self.margin + self.extension) >= min(self.width, self.height):
|
|
raise ValueError("Margins and extensions leave no room for the map.")
|
|
|
|
|
|
def render(
|
|
db: Path,
|
|
session_id: str,
|
|
output: Path,
|
|
*,
|
|
start: float | None = None,
|
|
end: float | None = None,
|
|
options: PlotOptions | None = None,
|
|
cache: Path = Path(".flight-map-cache"),
|
|
local_map: Path | None = None,
|
|
terrain: Path | None = None,
|
|
terrain_reference: str = "unknown",
|
|
force: bool = False,
|
|
refresh_map: bool = False,
|
|
no_terrain: bool = False,
|
|
refresh_terrain: bool = False,
|
|
) -> int:
|
|
options = options or PlotOptions()
|
|
options.validate()
|
|
if no_terrain and (terrain is not None or refresh_terrain):
|
|
raise ValueError("--no-terrain cannot be combined with --terrain or --refresh-terrain.")
|
|
if terrain is not None and refresh_terrain:
|
|
raise ValueError("--refresh-terrain applies to automatic AHN data, not --terrain.")
|
|
if terrain is None and terrain_reference != "unknown":
|
|
raise ValueError("--terrain-reference requires a local --terrain raster.")
|
|
if output.exists() and not force:
|
|
raise ValueError(f"Output already exists: {output}. Use --force to replace it.")
|
|
if any(
|
|
output.resolve() == source.resolve()
|
|
for source in (db, local_map, terrain)
|
|
if source is not None
|
|
):
|
|
raise ValueError("SVG output must not overwrite the recording database or input data.")
|
|
with database(db, readonly=True) as connection:
|
|
session = get_session(connection, session_id)
|
|
start = session.started if start is None else start
|
|
end = session.ended if end is None else end
|
|
if (
|
|
not math.isfinite(start)
|
|
or not math.isfinite(end)
|
|
or start < session.started
|
|
or end > session.ended
|
|
or start >= end
|
|
):
|
|
raise ValueError("Time selection must be a nonempty interval within the recording.")
|
|
observations = load_observations(connection, session_id, start, end)
|
|
if not observations:
|
|
raise ValueError("No recorded observations in the selected interval.")
|
|
projection = Projection(session.bounds)
|
|
data = map_data(session.bounds, cache, local_map, refresh=refresh_map)
|
|
try:
|
|
water = water_areas(data)
|
|
except WaterMaskError as exc:
|
|
LOGGER.warning(
|
|
"Land elevation unavailable: %s. Reported flight altitudes are unaffected.", exc
|
|
)
|
|
water = None
|
|
terrain_source = "disabled" if no_terrain else "local terrain"
|
|
if terrain is None and not no_terrain:
|
|
terrain = terrain_data(session.bounds, cache / "terrain", refresh=refresh_terrain)
|
|
terrain_reference = "nap" if terrain else "unknown"
|
|
terrain_source = "PDOK AHN DTM dtm_05m; NAP; CC0"
|
|
if terrain is not None and terrain.resolve() == output.resolve():
|
|
raise ValueError("SVG output must not overwrite cached terrain data.")
|
|
with Elevation(terrain, terrain_reference) as elevation:
|
|
ground = elevation.mean_ground(session.bounds, water=water) if water is not None else None
|
|
land_coverage = elevation.land_coverage
|
|
land_samples = elevation.land_samples
|
|
valid_land_samples = elevation.valid_land_samples
|
|
water_samples = elevation.water_samples
|
|
tracks = make_tracks(observations, projection, options.max_gap)
|
|
if not tracks:
|
|
raise ValueError("No usable airborne tracks cross the map in the selected interval.")
|
|
if ground is None:
|
|
LOGGER.warning(
|
|
"Average land elevation unavailable: no usable height map or land coverage. "
|
|
"Reported flight altitudes are unaffected."
|
|
)
|
|
missing = sum(track.average_altitude is None for track in tracks)
|
|
if missing:
|
|
LOGGER.warning(
|
|
"%d track altitude(s) unavailable: reported barometric altitude is missing "
|
|
"on a contributing segment; no geometric-altitude fallback is used.",
|
|
missing,
|
|
)
|
|
row_height = options.font_height * 1.8
|
|
terrain_attribution = terrain_source.startswith("PDOK")
|
|
legend_height = (len(tracks) + 5 + int(terrain_attribution)) * row_height
|
|
available_width = options.width - 2 * (options.margin + options.extension)
|
|
available_height = options.height - 2 * (options.margin + options.extension) - legend_height
|
|
if available_height < 25:
|
|
raise ValueError("Too many tracks for this page. Select a shorter interval or larger page.")
|
|
west, south, east, north = projection.rectangle.bounds
|
|
scale = min(available_width / (east - west), available_height / (north - south))
|
|
features = map_lines(data, projection, min_building_size=3 * options.pen_width / scale)
|
|
map_width, map_height = (east - west) * scale, (north - south) * scale
|
|
left = (options.width - map_width) / 2
|
|
top = options.margin + options.extension
|
|
matrix = [scale, 0, 0, -scale, left - west * scale, top + north * scale]
|
|
map_rectangle = box(left, top, left + map_width, top + map_height)
|
|
track_rectangle = box(
|
|
left - options.extension,
|
|
top - options.extension,
|
|
left + map_width + options.extension,
|
|
top + map_height + options.extension,
|
|
)
|
|
root = ET.Element(
|
|
tag("svg"),
|
|
{
|
|
"width": f"{options.width:g}mm",
|
|
"height": f"{options.height:g}mm",
|
|
"viewBox": f"0 0 {options.width:g} {options.height:g}",
|
|
"fill": "none",
|
|
"stroke": "black",
|
|
"stroke-width": f"{options.pen_width:g}",
|
|
"stroke-linecap": "round",
|
|
"stroke-linejoin": "round",
|
|
},
|
|
)
|
|
ET.SubElement(root, tag("title")).text = "Recorded flight map"
|
|
ET.SubElement(root, tag("desc")).text = (
|
|
f"Session {session.id}; bounds W,S,E,N={session.bounds.values}; "
|
|
f"{utc(start)} to {utc(end)} UTC. Local equirectangular projection. "
|
|
"Flight altitude is the time-weighted mean reported barometric pressure altitude "
|
|
"within the map, in metres, without terrain subtraction or vertical corrections. "
|
|
"It is not height above ground or terrain clearance. "
|
|
"Mean land elevation is a separate summary excluding mapped OSM water. "
|
|
f"Terrain source: {terrain_source}; "
|
|
f"Terrain reference: {terrain_reference}; "
|
|
f"valid land samples: {valid_land_samples}/{land_samples}; "
|
|
f"land coverage: {100 * land_coverage:.1f}%; excluded water samples: {water_samples}; "
|
|
f"mean ground elevation in metres: {ground}; "
|
|
"Map data: OpenStreetMap contributors, ODbL, https://www.openstreetmap.org/copyright. "
|
|
"Flight data: The OpenSky Network, https://opensky-network.org. "
|
|
"Schaefer et al., Bringing Up OpenSky: A Large-scale ADS-B Sensor Network "
|
|
"for Research, IPSN 2014, pp. 83-94."
|
|
)
|
|
map_group = ET.SubElement(root, tag("g"), {"id": "map"})
|
|
flight_group = ET.SubElement(root, tag("g"), {"id": "flights"})
|
|
text_group = ET.SubElement(root, tag("g"), {"id": "lettering"})
|
|
add_line(map_group, LineString(map_rectangle.exterior.coords))
|
|
seen = set()
|
|
for feature in features:
|
|
page_feature = affine_transform(feature, matrix).intersection(map_rectangle)
|
|
for part in lines(page_feature):
|
|
simplified = part.simplify(0.05)
|
|
key = simplified.normalize().wkb
|
|
if key not in seen:
|
|
for simplified_part in lines(simplified):
|
|
add_line(map_group, simplified_part)
|
|
seen.add(key)
|
|
occupied: list[BaseGeometry] = []
|
|
for index, track in enumerate(tracks, 1):
|
|
page_curves = [
|
|
part
|
|
for curve in track.curves
|
|
for part in curve.transform(matrix).clipped(track_rectangle.bounds)
|
|
]
|
|
group = ET.SubElement(
|
|
flight_group,
|
|
tag("g"),
|
|
{
|
|
"id": f"flight-{index}",
|
|
"data-icao24": track.observations[0].icao24,
|
|
"data-callsign": track.observations[0].callsign or "unavailable",
|
|
"data-start-utc": utc(track.start),
|
|
"data-end-utc": utc(track.end),
|
|
"data-altitude-source": "barometric",
|
|
"data-average-barometric-altitude-metres": (
|
|
f"{track.average_altitude:.3f}"
|
|
if track.average_altitude is not None
|
|
else "unavailable"
|
|
),
|
|
},
|
|
)
|
|
add_curves(group, page_curves)
|
|
short_label = str(index)
|
|
placed = False
|
|
for fraction in (0.9, 0.1, 0.7, 0.3, 0.5):
|
|
curve_index = min(int(fraction * len(page_curves)), len(page_curves) - 1)
|
|
anchor_x, anchor_y = page_curves[curve_index].point(
|
|
fraction * len(page_curves) - curve_index
|
|
)
|
|
for dx, dy in ((1, 1), (1, -3), (-4, 1), (-4, -3)):
|
|
x, y = anchor_x + dx, anchor_y + dy
|
|
extent = box(
|
|
x - 0.3,
|
|
y - 0.3,
|
|
x + len(short_label) * options.font_height,
|
|
y + options.font_height + 0.3,
|
|
)
|
|
if track_rectangle.covers(extent) and not any(
|
|
extent.intersects(previous) for previous in occupied
|
|
):
|
|
label(text_group, short_label, x, y, options.font_height)
|
|
occupied.append(extent)
|
|
placed = True
|
|
break
|
|
if placed:
|
|
break
|
|
if not placed:
|
|
raise ValueError("Cannot place track identifiers legibly; select fewer tracks.")
|
|
legend_y = top + map_height + options.extension + row_height
|
|
ground_label = "N/A" if ground is None else f"{ground:.1f} M"
|
|
legend = [
|
|
"FLIGHT TIMES UTC - REPORTED BAROMETRIC ALTITUDE, NOT HEIGHT ABOVE GROUND",
|
|
f"MAP AVERAGE LAND ELEVATION: {ground_label} - N/A MEANS UNAVAILABLE",
|
|
f"LAND ELEVATION COVERAGE: {100 * land_coverage:.1f} PERCENT - MAPPED WATER EXCLUDED",
|
|
]
|
|
for index, track in enumerate(tracks, 1):
|
|
first = track.observations[0]
|
|
altitude_label = (
|
|
"N/A" if track.average_altitude is None else f"{track.average_altitude:.0f} M"
|
|
)
|
|
legend.append(
|
|
f"{index}: {utc(track.start)} TO {utc(track.end)} "
|
|
f"{first.callsign or 'CALLSIGN N/A'} ({first.icao24.upper()}) BARO ALT {altitude_label}"
|
|
)
|
|
legend.extend(
|
|
[
|
|
"MAP: OPENSTREETMAP CONTRIBUTORS (ODBL) - OPENSTREETMAP.ORG/COPYRIGHT",
|
|
"FLIGHTS: THE OPENSKY NETWORK - OPENSKY-NETWORK.ORG",
|
|
]
|
|
)
|
|
if terrain_attribution:
|
|
legend.append("TERRAIN: AHN DTM VIA PDOK (CC0) - ELEVATIONS RELATIVE TO NAP")
|
|
for index, text in enumerate(legend):
|
|
if len(text) * options.font_height > options.width - 2 * options.margin:
|
|
raise ValueError("Legend is too wide; reduce --font-height or increase page width.")
|
|
label(text_group, text, options.margin, legend_y + index * row_height, options.font_height)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
temp_path: Path | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(dir=output.parent, suffix=".svg", delete=False) as stream:
|
|
temp_path = Path(stream.name)
|
|
ET.ElementTree(root).write(stream, encoding="utf-8", xml_declaration=True)
|
|
os.replace(temp_path, output)
|
|
finally:
|
|
if temp_path and temp_path.exists():
|
|
temp_path.unlink()
|
|
return len(tracks)
|