jschoubben
6eb693203d
The mesh builds the CDN from this repository and a commit (novox/hq ADR 0069); registry-api.novox.be, where HAL's image came from, is retired, so the running container can no longer be re-pulled. The HAL module.yml and service/docker-compose.yml retire with HAL. The Kometa metadata links every poster to this service's own address, which was the literal https://f1-circuits.zurag.be. It is now F1_CIRCUITS_BASE_URL, written by the mesh from the route's name (${bound:route:name}, mesh-controller #149) - falling back to the old address, so HAL's deployment behaves as before. The tile user agent follows it. The poster cache is a placed directory mounted at /app/cache, as HAL's /services/f1-circuits/cache was. The image was built from unpinned pip installs; service/requirements.txt now pins the exact set the running container has (pip freeze), and the base is python 3.12.13-alpine3.24 by digest (a later alpine 3.24 patch than the running build's 3.24.0 - the older digest is no longer tagged). The slug "f1" keeps the route login within 20 characters on any machine. Verified: the manifest parses on mesh-controller main and #149 and renders for a zurag.be node (F1_CIRCUITS_BASE_URL=https://f1-circuits. zurag.be). Built from this commit and run on an empty cache: /circuits, /circuits/<country> and a rendered Spa track are byte-identical to what ace serves today; the base URL comes from the variable, and without it falls back to the old address. Depends on mesh-controller #149.
788 lines
35 KiB
Python
788 lines
35 KiB
Python
"""Plex-style F1 race + season poster rendering.
|
|
|
|
Produces 1000x1500 PNGs (Plex's recommended TV-poster ratio) by compositing:
|
|
- Top + bottom checker-flag strips
|
|
- "FORMULA 1" wordmark (red)
|
|
- Round / year ribbon
|
|
- Grand Prix name (bold, large)
|
|
- Rasterised track outline (white) for race posters
|
|
- Race date
|
|
|
|
Output is cached under cache/posters/ keyed by (season, round, layout_slug)
|
|
so cold paths stay cheap.
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import urllib.request
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import cairosvg
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
from circuits_service import CircuitService
|
|
from ergast_service import ErgastService
|
|
|
|
# Where season background images live. Hosted by the dilbert files-api module —
|
|
# one landscape JPG per F1 season (key for the season-card grid in Dilbert's UI).
|
|
SEASON_BG_URL_TEMPLATE = "https://files-api.novox.be/dilbert/sports/formula1/cars/{season}.jpg"
|
|
|
|
|
|
class BackgroundStyle(str, Enum):
|
|
"""Selectable backdrop for race posters."""
|
|
AERIAL = "aerial" # Esri World Imagery satellite tiles (default)
|
|
OSM = "osm" # OpenStreetMap Carto street-map tiles
|
|
DARK = "dark" # CartoDB Dark Matter — minimalist black-and-grey map
|
|
LIGHT = "light" # CartoDB Positron — minimalist light/grey map
|
|
NONE = "none" # Plain black, no map fetch — fast fallback
|
|
|
|
|
|
# XYZ tile templates. None require API keys at our volume.
|
|
TILE_PROVIDERS = {
|
|
BackgroundStyle.AERIAL: {
|
|
"url": "https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}",
|
|
"attribution": "Esri, Maxar, Earthstar Geographics",
|
|
},
|
|
BackgroundStyle.OSM: {
|
|
"url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"attribution": "© OpenStreetMap contributors",
|
|
},
|
|
BackgroundStyle.DARK: {
|
|
"url": "https://basemaps.cartocdn.com/dark_all/{z}/{x}/{y}.png",
|
|
"attribution": "© CartoDB, © OpenStreetMap contributors",
|
|
},
|
|
BackgroundStyle.LIGHT: {
|
|
"url": "https://basemaps.cartocdn.com/light_all/{z}/{x}/{y}.png",
|
|
"attribution": "© CartoDB, © OpenStreetMap contributors",
|
|
},
|
|
}
|
|
|
|
# The address this service is reached at, which the Kometa metadata links every poster to. The
|
|
# mesh writes it from the name the route is served under (F1_CIRCUITS_BASE_URL); the fallback is
|
|
# the address HAL served it at, so a deployment that sets nothing behaves as before.
|
|
PUBLIC_BASE_URL = os.environ.get("F1_CIRCUITS_BASE_URL", "https://f1-circuits.zurag.be").rstrip("/")
|
|
|
|
# Polite UA — both providers expect one and Esri throttles aggressively without it.
|
|
TILE_USER_AGENT = f"f1-circuits-posters/1.0 (+{PUBLIC_BASE_URL})"
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
POSTER_W, POSTER_H = 1000, 1500
|
|
THUMB_W, THUMB_H = 1920, 1080 # Plex episode-thumb ratio
|
|
BG = (10, 10, 12)
|
|
RED = (220, 18, 32) # F1 red
|
|
WHITE = (245, 245, 245)
|
|
GREY = (160, 160, 170)
|
|
|
|
# Kometa F1 builder episode numbering — keep in sync with Kometa's modules/meta.py.
|
|
# Each value is (display_label, url_slug) used for thumb generation + filename.
|
|
_SESSION_STD = {
|
|
1: ("Free Practice 1", "fp1"),
|
|
2: ("Free Practice 2", "fp2"),
|
|
3: ("Free Practice 3", "fp3"),
|
|
4: ("Pre-Qualifying Buildup", "pre-qualifying"),
|
|
5: ("Qualifying", "qualifying"),
|
|
6: ("Post-Qualifying", "post-qualifying"),
|
|
7: ("Pre-Race Buildup", "pre-race"),
|
|
8: ("Race", "race"),
|
|
9: ("Post-Race", "post-race"),
|
|
10: ("Highlights", "highlights"),
|
|
}
|
|
_SESSION_SPRINT = {
|
|
1: ("Free Practice 1", "fp1"),
|
|
2: ("Pre-Sprint-Qualifying Buildup", "pre-sprint-qualifying"),
|
|
3: ("Sprint Qualifying", "sprint-qualifying"),
|
|
4: ("Post-Sprint-Qualifying", "post-sprint-qualifying"),
|
|
5: ("Pre-Sprint Buildup", "pre-sprint"),
|
|
6: ("Sprint", "sprint"),
|
|
7: ("Post-Sprint", "post-sprint"),
|
|
8: ("Pre-Qualifying Buildup", "pre-qualifying"),
|
|
9: ("Qualifying", "qualifying"),
|
|
10: ("Post-Qualifying", "post-qualifying"),
|
|
11: ("Pre-Race Buildup", "pre-race"),
|
|
12: ("Race", "race"),
|
|
13: ("Post-Race", "post-race"),
|
|
14: ("Highlights", "highlights"),
|
|
15: ("Bonus Content", "bonus"),
|
|
}
|
|
# Display-label lookup for the thumb renderer (slug → label, deduped across formats).
|
|
SESSION_LABEL_BY_SLUG = {slug: label for d in (_SESSION_STD, _SESSION_SPRINT) for label, slug in d.values()}
|
|
|
|
CHECKER_H = 60
|
|
CHECKER_SQ = 30
|
|
|
|
# Vertical pixel where we want the circuit's geographic centre to land. Picked so
|
|
# the track ends up visually centred *in the visible aerial band* — between the
|
|
# heading block at the top and the date strip at the bottom, not in the full
|
|
# 1000x1500 poster middle (which would look high once the heading dim is over it).
|
|
GEO_CENTER_Y = 935
|
|
|
|
# Slim-image-friendly font candidates. The Dockerfile installs fonts-dejavu-core
|
|
# so DejaVuSans-Bold is always present in production; the others cover common
|
|
# local-dev paths (Arch Linux ships Liberation Sans at /usr/share/fonts/liberation).
|
|
FONT_CANDIDATES = [
|
|
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", # Debian/Ubuntu
|
|
"/usr/share/fonts/dejavu/DejaVuSans-Bold.ttf", # Fedora
|
|
"/usr/share/fonts/liberation/LiberationSans-Bold.ttf", # Arch
|
|
"/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
|
|
]
|
|
|
|
|
|
class PostersService:
|
|
def __init__(self, circuits: CircuitService, ergast: ErgastService):
|
|
self.circuits = circuits
|
|
self.ergast = ergast
|
|
self.font_path = next((p for p in FONT_CANDIDATES if Path(p).exists()), None)
|
|
if not self.font_path:
|
|
logger.warning("No bundled font found — poster text will use PIL's default bitmap font")
|
|
self.cache_dir = Path("cache/posters")
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
# Pre-tuned zooms per circuit, keyed by slugified circuit name.
|
|
# Falls back to DEFAULT_ZOOM when not found.
|
|
self._zoom_by_circuit = self._load_zoom_lookup()
|
|
|
|
DEFAULT_ZOOM = 15
|
|
|
|
@staticmethod
|
|
def _load_zoom_lookup() -> dict:
|
|
"""Index f1-locations.json by slugified circuit name for fast zoom lookup."""
|
|
from ergast_service import slugify
|
|
path = Path("f1-locations.json")
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
except Exception as e:
|
|
logger.warning(f"Could not load f1-locations.json: {e}")
|
|
return {}
|
|
return {slugify(entry["name"]): entry for entry in data if "name" in entry}
|
|
|
|
# ------------------------------------------------------------------ utils
|
|
|
|
def _font(self, size: int) -> ImageFont.FreeTypeFont:
|
|
if self.font_path:
|
|
return ImageFont.truetype(self.font_path, size)
|
|
return ImageFont.load_default()
|
|
|
|
def _centered(self, draw: ImageDraw.ImageDraw, text: str, y: int, size: int, fill=WHITE) -> int:
|
|
"""Draw text centered horizontally; returns the drawn height."""
|
|
font = self._font(size)
|
|
# Shrink-to-fit if too wide
|
|
while size > 18 and draw.textlength(text, font) > POSTER_W - 80:
|
|
size -= 4
|
|
font = self._font(size)
|
|
w = draw.textlength(text, font)
|
|
draw.text(((POSTER_W - w) / 2, y), text, font=font, fill=fill)
|
|
bbox = draw.textbbox((0, 0), text, font=font)
|
|
return bbox[3] - bbox[1]
|
|
|
|
def _checker_strip(self, draw: ImageDraw.ImageDraw, y: int, height: int = CHECKER_H, sq: int = CHECKER_SQ):
|
|
for col, x in enumerate(range(0, POSTER_W + sq, sq)):
|
|
for row, yy in enumerate(range(y, y + height, sq)):
|
|
fill = WHITE if (col + row) % 2 == 0 else BG
|
|
draw.rectangle([x, yy, x + sq, yy + sq], fill=fill)
|
|
|
|
# ---------------------------------------------------------- map tiles
|
|
|
|
@staticmethod
|
|
def _latlon_to_pixel(lat: float, lon: float, zoom: int) -> tuple[float, float]:
|
|
"""Standard Web-Mercator pixel coordinates (slippy-map convention, 256-px tiles)."""
|
|
n = 2.0 ** zoom * 256
|
|
x = (lon + 180.0) / 360.0 * n
|
|
lat_rad = math.radians(lat)
|
|
y = (1.0 - math.log(math.tan(lat_rad) + 1.0 / math.cos(lat_rad)) / math.pi) / 2.0 * n
|
|
return x, y
|
|
|
|
def _fetch_map_background(self, lat: float, lon: float, zoom: int, style: BackgroundStyle) -> Optional[Image.Image]:
|
|
"""Stitch enough 256-px tiles around (lat, lon) to fill the full POSTER_W x POSTER_H."""
|
|
provider = TILE_PROVIDERS.get(style)
|
|
if not provider:
|
|
return None
|
|
|
|
# Centre pixel and the slab we need to cover. Anchor the geographic
|
|
# centre at GEO_CENTER_Y rather than POSTER_H/2 so the track lands in
|
|
# the middle of the *visible* aerial band.
|
|
cpx, cpy = self._latlon_to_pixel(lat, lon, zoom)
|
|
left, top = int(cpx - POSTER_W / 2), int(cpy - GEO_CENTER_Y)
|
|
tile_x_lo, tile_y_lo = left // 256, top // 256
|
|
tile_x_hi = (left + POSTER_W + 255) // 256
|
|
tile_y_hi = (top + POSTER_H + 255) // 256
|
|
|
|
canvas_w = (tile_x_hi - tile_x_lo) * 256
|
|
canvas_h = (tile_y_hi - tile_y_lo) * 256
|
|
canvas = Image.new("RGB", (canvas_w, canvas_h), BG)
|
|
|
|
any_ok = False
|
|
for tx in range(tile_x_lo, tile_x_hi):
|
|
for ty in range(tile_y_lo, tile_y_hi):
|
|
url = provider["url"].format(z=zoom, x=tx, y=ty)
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": TILE_USER_AGENT})
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
tile = Image.open(io.BytesIO(resp.read())).convert("RGB")
|
|
canvas.paste(tile, ((tx - tile_x_lo) * 256, (ty - tile_y_lo) * 256))
|
|
any_ok = True
|
|
except Exception as e:
|
|
logger.warning(f"Tile fetch failed {style.value} z={zoom} x={tx} y={ty}: {e}")
|
|
|
|
if not any_ok:
|
|
return None
|
|
|
|
ox = left - tile_x_lo * 256
|
|
oy = top - tile_y_lo * 256
|
|
return canvas.crop((ox, oy, ox + POSTER_W, oy + POSTER_H))
|
|
|
|
def _draw_track_projected(
|
|
self,
|
|
img: Image.Image,
|
|
coords: list,
|
|
lat0: float,
|
|
lon0: float,
|
|
zoom: int,
|
|
canvas_w: int = POSTER_W,
|
|
geo_center_y: int = GEO_CENTER_Y,
|
|
) -> bool:
|
|
"""Draw the track onto `img` by projecting each GeoJSON (lon, lat) point
|
|
into pixel space using the same web-mercator math as the map background.
|
|
This guarantees the white outline overlays the actual circuit in the
|
|
underlying aerial / OSM tiles.
|
|
|
|
canvas_w / geo_center_y must match the dimensions/offset used to fetch
|
|
the map background so the polyline aligns with the actual track in the
|
|
rendered tiles. Defaults match the race poster; pass THUMB_W / THUMB_H/2
|
|
for landscape session thumbs.
|
|
|
|
Returns True on success, False if there's nothing usable to draw."""
|
|
if not coords:
|
|
return False
|
|
|
|
cpx, cpy = self._latlon_to_pixel(lat0, lon0, zoom)
|
|
win_left = cpx - canvas_w / 2
|
|
win_top = cpy - geo_center_y
|
|
|
|
pts: list[tuple[float, float]] = []
|
|
for entry in coords:
|
|
# GeoJSON LineString is [[lon, lat], ...]; tolerate (lon, lat) tuples too.
|
|
if len(entry) < 2:
|
|
continue
|
|
lon, lat = float(entry[0]), float(entry[1])
|
|
px, py = self._latlon_to_pixel(lat, lon, zoom)
|
|
pts.append((px - win_left, py - win_top))
|
|
|
|
if len(pts) < 2:
|
|
return False
|
|
|
|
d = ImageDraw.Draw(img)
|
|
# Subtle dark stroke under, then the bright white line, so the outline
|
|
# stays legible whether the bg is the bright Mediterranean or a dark
|
|
# CartoDB tile.
|
|
d.line(pts + [pts[0]], fill=(0, 0, 0), width=10, joint="curve")
|
|
d.line(pts + [pts[0]], fill=WHITE, width=6, joint="curve")
|
|
return True
|
|
|
|
def _render_world_locator(self, lat: float, lon: float, size: int = 220) -> Optional[Image.Image]:
|
|
"""A small picture-in-picture map showing where on Earth this race is.
|
|
Uses CartoDB Dark Matter at zoom 5 — tight enough that the host country
|
|
fills most of the inset, while neighbours and coastlines still give the
|
|
viewer a "here's the region" anchor. A marker (red dot + white ring)
|
|
is drawn over the exact spot."""
|
|
zoom = 5
|
|
# Use the same stitcher with a custom canvas target.
|
|
provider = TILE_PROVIDERS[BackgroundStyle.DARK]
|
|
cpx, cpy = self._latlon_to_pixel(lat, lon, zoom)
|
|
# Frame 700x700 px window centred on the location, then downscale to `size`.
|
|
win = 700
|
|
left, top = int(cpx - win / 2), int(cpy - win / 2)
|
|
tx0, ty0 = left // 256, top // 256
|
|
tx1, ty1 = (left + win + 255) // 256, (top + win + 255) // 256
|
|
canvas = Image.new("RGB", ((tx1 - tx0) * 256, (ty1 - ty0) * 256), BG)
|
|
any_ok = False
|
|
for tx in range(tx0, tx1):
|
|
for ty in range(ty0, ty1):
|
|
# Wrap longitude tiles (world is cyclic at zoom 2 = 4 tiles wide)
|
|
wrapped_x = tx % (2 ** zoom)
|
|
if wrapped_x < 0:
|
|
wrapped_x += 2 ** zoom
|
|
if not (0 <= ty < 2 ** zoom):
|
|
continue
|
|
url = provider["url"].format(z=zoom, x=wrapped_x, y=ty)
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": TILE_USER_AGENT})
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
tile = Image.open(io.BytesIO(resp.read())).convert("RGB")
|
|
canvas.paste(tile, ((tx - tx0) * 256, (ty - ty0) * 256))
|
|
any_ok = True
|
|
except Exception as e:
|
|
logger.warning(f"Locator tile fail {tx},{ty}: {e}")
|
|
if not any_ok:
|
|
return None
|
|
ox, oy = left - tx0 * 256, top - ty0 * 256
|
|
cropped = canvas.crop((ox, oy, ox + win, oy + win)).resize((size, size), Image.LANCZOS)
|
|
|
|
# Marker: red filled circle + white ring at the exact spot (centre of the frame).
|
|
d = ImageDraw.Draw(cropped, "RGBA")
|
|
cx, cy = size // 2, size // 2
|
|
d.ellipse([cx - 9, cy - 9, cx + 9, cy + 9], outline=(255, 255, 255, 255), width=2)
|
|
d.ellipse([cx - 5, cy - 5, cx + 5, cy + 5], fill=RED + (255,))
|
|
|
|
# Thin white frame around the inset itself.
|
|
d.rectangle([0, 0, size - 1, size - 1], outline=(255, 255, 255, 200), width=2)
|
|
return cropped
|
|
|
|
# Visible aerial slot (between heading band and date strip) — target the
|
|
# bbox to ~85% of it so there's some breathing room around the track.
|
|
_BBOX_TARGET_W = int(POSTER_W * 0.85)
|
|
_BBOX_TARGET_H = int((1300 - 540) * 0.85) # visible aerial slot height, scaled
|
|
|
|
def _zoom_from_bbox(self, coords: list, max_zoom: int = 17, min_zoom: int = 10) -> int:
|
|
"""Pick the tightest zoom level at which the track's GeoJSON bbox still
|
|
fits inside the visible aerial area. Walks from high → low zoom and
|
|
returns the first zoom that fits — i.e. the most zoomed-in view that
|
|
still shows the whole lap."""
|
|
if len(coords) < 2:
|
|
return self.DEFAULT_ZOOM
|
|
lons = [float(c[0]) for c in coords if len(c) >= 2]
|
|
lats = [float(c[1]) for c in coords if len(c) >= 2]
|
|
if not lons or not lats:
|
|
return self.DEFAULT_ZOOM
|
|
lat_min, lat_max = min(lats), max(lats)
|
|
lon_min, lon_max = min(lons), max(lons)
|
|
for zoom in range(max_zoom, min_zoom - 1, -1):
|
|
x_lo, y_hi = self._latlon_to_pixel(lat_min, lon_min, zoom) # SW (small x, large y)
|
|
x_hi, y_lo = self._latlon_to_pixel(lat_max, lon_max, zoom) # NE (large x, small y)
|
|
if (x_hi - x_lo) <= self._BBOX_TARGET_W and (y_hi - y_lo) <= self._BBOX_TARGET_H:
|
|
return zoom
|
|
return min_zoom
|
|
|
|
def _resolve_circuit_geo(self, grand_prix_name: str, season: int) -> Optional[tuple[float, float, int]]:
|
|
"""Pull (lat, lon, zoom) for the circuit hosting this GP in this season.
|
|
|
|
Centre comes from the GeoJSON bbox midpoint (geometrically accurate),
|
|
zoom is computed from the bbox so a tight street circuit gets zoomed in
|
|
and a sprawling road course is zoomed out — automatically, no hand-tuning.
|
|
|
|
Falls back to Jolpica's circuit coordinate + f1-locations.json's hand-
|
|
tuned zoom when GeoJSON coords aren't available for this layout."""
|
|
races = self.ergast._fetch_season_schedule(season)
|
|
race = next((r for r in races if r["raceName"] == grand_prix_name), None)
|
|
if not race:
|
|
return None
|
|
|
|
layout = self.circuits.get_circuit_layout_by_ergast_data(grand_prix_name, season)
|
|
if layout:
|
|
try:
|
|
layout.load_geo_json_data()
|
|
coords = layout.coordinates
|
|
except Exception:
|
|
coords = None
|
|
if coords and len(coords) >= 2:
|
|
lons = [float(c[0]) for c in coords if len(c) >= 2]
|
|
lats = [float(c[1]) for c in coords if len(c) >= 2]
|
|
lat_c = (min(lats) + max(lats)) / 2
|
|
lon_c = (min(lons) + max(lons)) / 2
|
|
return lat_c, lon_c, self._zoom_from_bbox(coords)
|
|
|
|
# Fallback path: Jolpica coordinate + hand-tuned zoom
|
|
loc = race.get("Circuit", {}).get("Location", {})
|
|
if "lat" not in loc or "long" not in loc:
|
|
return None
|
|
lat, lon = float(loc["lat"]), float(loc["long"])
|
|
from ergast_service import slugify
|
|
entry = self._zoom_by_circuit.get(slugify(race["Circuit"].get("circuitName", "")))
|
|
zoom = entry["zoom"] if entry and "zoom" in entry else self.DEFAULT_ZOOM
|
|
return lat, lon, zoom
|
|
|
|
# ---------------------------------------------------------- track raster
|
|
|
|
def _rasterise_track(self, svg_path: Path, target_w: int = 700) -> Optional[Image.Image]:
|
|
try:
|
|
png_bytes = cairosvg.svg2png(
|
|
bytestring=svg_path.read_bytes(),
|
|
output_width=target_w,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to rasterise {svg_path}: {e}")
|
|
return None
|
|
track = Image.open(io.BytesIO(png_bytes)).convert("RGBA")
|
|
# Recolour all opaque pixels (black stroke in source) → white
|
|
px = track.load()
|
|
for y in range(track.height):
|
|
for x in range(track.width):
|
|
r, g, b, a = px[x, y]
|
|
if a > 0:
|
|
px[x, y] = (WHITE[0], WHITE[1], WHITE[2], a)
|
|
return track
|
|
|
|
# ------------------------------------------------------------------ race poster
|
|
|
|
def render_race_poster(
|
|
self,
|
|
grand_prix_name: str,
|
|
season: int,
|
|
bg: BackgroundStyle = BackgroundStyle.AERIAL,
|
|
) -> Optional[bytes]:
|
|
races = self.ergast._fetch_season_schedule(season)
|
|
race = next((r for r in races if r["raceName"] == grand_prix_name), None)
|
|
if not race:
|
|
logger.error(f"Race not found: {grand_prix_name} {season}")
|
|
return None
|
|
|
|
round_num = int(race["round"])
|
|
date = race.get("date", "")
|
|
|
|
layout = self.circuits.get_circuit_layout_by_ergast_data(grand_prix_name, season)
|
|
if not layout:
|
|
logger.error(f"Layout not resolved for {grand_prix_name} {season}")
|
|
return None
|
|
|
|
cache_key = f"{season}-{round_num:02d}-{layout.slug}-{bg.value}.png"
|
|
cache_path = self.cache_dir / cache_key
|
|
if cache_path.exists():
|
|
return cache_path.read_bytes()
|
|
|
|
# Fetch map background if requested; fall back to flat black if anything fails.
|
|
map_bg: Optional[Image.Image] = None
|
|
if bg is not BackgroundStyle.NONE:
|
|
geo = self._resolve_circuit_geo(grand_prix_name, season)
|
|
if geo:
|
|
lat, lon, zoom = geo
|
|
map_bg = self._fetch_map_background(lat, lon, zoom, bg)
|
|
|
|
if map_bg is not None:
|
|
# Dim the underlying map so the white track + heading text stay legible.
|
|
img = Image.alpha_composite(map_bg.convert("RGBA"), self._race_dark_overlay()).convert("RGB")
|
|
else:
|
|
img = Image.new("RGB", (POSTER_W, POSTER_H), BG)
|
|
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# Top checker
|
|
self._checker_strip(draw, 0)
|
|
|
|
# FORMULA 1 wordmark
|
|
self._centered(draw, "FORMULA 1", y=110, size=88, fill=RED)
|
|
|
|
# Round / Year ribbon
|
|
self._centered(draw, f"ROUND {round_num:02d} / {season}", y=220, size=34, fill=GREY)
|
|
|
|
# GP name (drop "Grand Prix" suffix; render "GRAND PRIX" smaller below)
|
|
name_short = grand_prix_name.replace("Grand Prix", "").strip().upper()
|
|
self._centered(draw, name_short, y=300, size=110, fill=WHITE)
|
|
self._centered(draw, "GRAND PRIX", y=440, size=44, fill=GREY)
|
|
|
|
# Track outline.
|
|
# When a map background is present we project the GeoJSON lat/lon points
|
|
# using the same zoom/centre we used for the tiles — that way the white
|
|
# outline is geographically aligned with the actual track in the photo.
|
|
# Without a map (bg=NONE) there's nothing to align with so we fall back
|
|
# to the hand-drawn SVG centred in the slot.
|
|
projected = False
|
|
if map_bg is not None:
|
|
try:
|
|
# GeoJSON is lazy-loaded — trigger the read before pulling coords.
|
|
layout.load_geo_json_data()
|
|
projected = self._draw_track_projected(
|
|
img, layout.coordinates, *self._resolve_circuit_geo(grand_prix_name, season)
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"GeoJSON track projection failed; falling back to SVG: {e}")
|
|
|
|
if not projected:
|
|
svg_path = Path("circuits") / layout.relative_svg_filepath
|
|
if svg_path.exists():
|
|
track = self._rasterise_track(svg_path, target_w=720)
|
|
if track:
|
|
slot_top, slot_bottom = 560, 1250
|
|
tx = (POSTER_W - track.width) // 2
|
|
ty = slot_top + ((slot_bottom - slot_top) - track.height) // 2
|
|
img.paste(track, (tx, ty), track)
|
|
|
|
# World locator inset (top-right) — picture-in-picture pointing at the
|
|
# exact race location on the globe. Only meaningful when a map bg is on.
|
|
# Sized + positioned so it tucks under the top checker and sits to the
|
|
# right of the centred FORMULA 1 wordmark without overlapping.
|
|
if bg is not BackgroundStyle.NONE:
|
|
geo = self._resolve_circuit_geo(grand_prix_name, season)
|
|
if geo:
|
|
lat, lon, _ = geo
|
|
locator = self._render_world_locator(lat, lon, size=140)
|
|
if locator:
|
|
margin = 25
|
|
img.paste(locator, (POSTER_W - locator.width - margin, CHECKER_H + margin))
|
|
|
|
# Date
|
|
if date:
|
|
self._centered(draw, date, y=POSTER_H - 170, size=32, fill=WHITE)
|
|
|
|
# Bottom checker
|
|
self._checker_strip(draw, POSTER_H - CHECKER_H)
|
|
|
|
buf = io.BytesIO()
|
|
img.save(buf, "PNG", optimize=True)
|
|
data = buf.getvalue()
|
|
cache_path.write_bytes(data)
|
|
return data
|
|
|
|
# ------------------------------------------------------------------ session thumb (16:9)
|
|
|
|
def _fetch_map_landscape(self, lat: float, lon: float, zoom: int, style: BackgroundStyle) -> Optional[Image.Image]:
|
|
"""Like _fetch_map_background but for THUMB_W x THUMB_H (1920x1080).
|
|
Centred so the geographic point lands at the visual midpoint."""
|
|
provider = TILE_PROVIDERS.get(style)
|
|
if not provider:
|
|
return None
|
|
cpx, cpy = self._latlon_to_pixel(lat, lon, zoom)
|
|
left, top = int(cpx - THUMB_W / 2), int(cpy - THUMB_H / 2)
|
|
tx0, ty0 = left // 256, top // 256
|
|
tx1, ty1 = (left + THUMB_W + 255) // 256, (top + THUMB_H + 255) // 256
|
|
canvas = Image.new("RGB", ((tx1 - tx0) * 256, (ty1 - ty0) * 256), BG)
|
|
any_ok = False
|
|
for tx in range(tx0, tx1):
|
|
for ty in range(ty0, ty1):
|
|
url = provider["url"].format(z=zoom, x=tx, y=ty)
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": TILE_USER_AGENT})
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
tile = Image.open(io.BytesIO(resp.read())).convert("RGB")
|
|
canvas.paste(tile, ((tx - tx0) * 256, (ty - ty0) * 256))
|
|
any_ok = True
|
|
except Exception as e:
|
|
logger.warning(f"Landscape tile fail {style.value} z={zoom} x={tx} y={ty}: {e}")
|
|
if not any_ok:
|
|
return None
|
|
ox, oy = left - tx0 * 256, top - ty0 * 256
|
|
return canvas.crop((ox, oy, ox + THUMB_W, oy + THUMB_H))
|
|
|
|
def render_session_thumb(self, grand_prix_name: str, season: int, session_slug: str) -> Optional[bytes]:
|
|
"""1920x1080 Plex episode-thumb. Aerial of the circuit + GP name + session label."""
|
|
label = SESSION_LABEL_BY_SLUG.get(session_slug)
|
|
if not label:
|
|
return None
|
|
|
|
races = self.ergast._fetch_season_schedule(season)
|
|
race = next((r for r in races if r["raceName"] == grand_prix_name), None)
|
|
if not race:
|
|
return None
|
|
round_num = int(race["round"])
|
|
layout = self.circuits.get_circuit_layout_by_ergast_data(grand_prix_name, season)
|
|
if not layout:
|
|
return None
|
|
|
|
cache_key = f"thumb-{season}-{round_num:02d}-{layout.slug}-{session_slug}.png"
|
|
cache_path = self.cache_dir / cache_key
|
|
if cache_path.exists():
|
|
return cache_path.read_bytes()
|
|
|
|
# Same per-circuit geo (lat, lon, bbox-derived zoom) as the race poster,
|
|
# so the thumb shows the same place at the same zoom level.
|
|
geo = self._resolve_circuit_geo(grand_prix_name, season)
|
|
map_bg = self._fetch_map_landscape(*geo, BackgroundStyle.AERIAL) if geo else None
|
|
|
|
img = Image.new("RGB", (THUMB_W, THUMB_H), BG)
|
|
if map_bg is not None:
|
|
# Heavier dim than the poster — 16:9 fills the player thumb at small
|
|
# sizes, so we need solid contrast for the overlaid label text.
|
|
overlay = Image.new("RGBA", (THUMB_W, THUMB_H), (0, 0, 0, 140))
|
|
img = Image.alpha_composite(map_bg.convert("RGBA"), overlay).convert("RGB")
|
|
|
|
# Project the track polyline onto the landscape canvas — same math
|
|
# as the race poster but with THUMB_W / THUMB_H/2 so it overlays the
|
|
# actual circuit visible in the aerial.
|
|
try:
|
|
layout.load_geo_json_data()
|
|
self._draw_track_projected(
|
|
img, layout.coordinates, *geo,
|
|
canvas_w=THUMB_W, geo_center_y=THUMB_H // 2,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Session-thumb track projection failed: {e}")
|
|
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# Two left-aligned text columns: GP name top-left, session label bottom-left.
|
|
# Round/year accent in red between them.
|
|
x_margin = 80
|
|
name_short = grand_prix_name.replace("Grand Prix", "").strip().upper()
|
|
|
|
# FORMULA 1 small wordmark top-left
|
|
font_wm = self._font(40)
|
|
draw.text((x_margin, 60), "FORMULA 1", fill=RED, font=font_wm)
|
|
|
|
# Round/year ribbon
|
|
font_ribbon = self._font(28)
|
|
draw.text((x_margin, 120), f"ROUND {round_num:02d} / {season}", fill=GREY, font=font_ribbon)
|
|
|
|
# GP name — fit to 1300px max width
|
|
gp_size = 120
|
|
font_gp = self._font(gp_size)
|
|
while gp_size > 60 and draw.textlength(name_short, font_gp) > 1300:
|
|
gp_size -= 10
|
|
font_gp = self._font(gp_size)
|
|
draw.text((x_margin, 180), name_short, fill=WHITE, font=font_gp)
|
|
draw.text((x_margin, 180 + gp_size + 5), "GRAND PRIX", fill=GREY, font=self._font(36))
|
|
|
|
# Session label — bottom-left, very prominent
|
|
sess_size = 90
|
|
font_sess = self._font(sess_size)
|
|
while sess_size > 50 and draw.textlength(label.upper(), font_sess) > 1300:
|
|
sess_size -= 6
|
|
font_sess = self._font(sess_size)
|
|
sess_label = label.upper()
|
|
sess_y = THUMB_H - 130 - sess_size
|
|
draw.text((x_margin, sess_y), sess_label, fill=WHITE, font=font_sess)
|
|
|
|
# Thin red accent bar above the session label
|
|
draw.rectangle([x_margin, sess_y - 16, x_margin + 120, sess_y - 8], fill=RED)
|
|
|
|
buf = io.BytesIO()
|
|
img.save(buf, "PNG", optimize=True)
|
|
data = buf.getvalue()
|
|
cache_path.write_bytes(data)
|
|
return data
|
|
|
|
# ------------------------------------------------------------------ kometa metadata
|
|
|
|
BASE_URL = PUBLIC_BASE_URL
|
|
|
|
def render_kometa_metadata(self, first_season: int = 1950, last_offset: int = 1) -> str:
|
|
"""Emit the Kometa metadata YAML for the Formula 1 library, covering
|
|
every F1 season from `first_season` through current_year + last_offset.
|
|
|
|
Per show: f1_season builder + url_poster pointing at the season hero.
|
|
Per Plex season (== round): url_poster pointing at the race poster.
|
|
Kometa silently skips shows / seasons that don't exist in the library."""
|
|
from datetime import datetime
|
|
from urllib.parse import quote
|
|
|
|
last_season = datetime.utcnow().year + last_offset
|
|
lines: list[str] = [
|
|
"# Auto-generated by f1-circuits (/kometa/formula-1-metadata.yml).",
|
|
"# Do not commit — fetch via metadata_files.url in Kometa config.",
|
|
"metadata:",
|
|
]
|
|
for season in range(first_season, last_season + 1):
|
|
races = self.ergast._fetch_season_schedule(season)
|
|
lines.append(f' "Season {season}":')
|
|
lines.append(f" f1_season: {season}")
|
|
lines.append(" round_prefix: true")
|
|
lines.append(" shorten_gp: true")
|
|
lines.append(f" url_poster: {self.BASE_URL}/posters/season/{season}.png")
|
|
if not races:
|
|
continue
|
|
lines.append(" seasons:")
|
|
for race in races:
|
|
try:
|
|
round_num = int(race["round"])
|
|
name = race["raceName"]
|
|
except (KeyError, ValueError):
|
|
continue
|
|
race_url = f"{self.BASE_URL}/posters/race/{quote(name)}/{season}.png"
|
|
lines.append(f" {round_num}:")
|
|
lines.append(f" url_poster: {race_url}")
|
|
|
|
# Per-episode session thumbs. Detect sprint format from the
|
|
# presence of a Sprint section in the Jolpica race entry; the
|
|
# F1 builder maps each format to its own episode numbering and
|
|
# we follow the same map here so each Plex episode lands on a
|
|
# correctly-labelled session thumb.
|
|
is_sprint = any(k.lower().startswith("sprint") for k in race.keys())
|
|
session_map = _SESSION_SPRINT if is_sprint else _SESSION_STD
|
|
lines.append(" episodes:")
|
|
for ep_num, (_label, slug) in session_map.items():
|
|
thumb_url = f"{self.BASE_URL}/posters/session/{quote(name)}/{season}/{slug}.png"
|
|
lines.append(f" {ep_num}:")
|
|
lines.append(f" url_poster: {thumb_url}")
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
# ------------------------------------------------------------------ season poster
|
|
|
|
def _fetch_season_background(self, season: int) -> Optional[Image.Image]:
|
|
"""Pull the season's hero photo from the dilbert files-api CDN."""
|
|
url = SEASON_BG_URL_TEMPLATE.format(season=season)
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=10) as resp:
|
|
bg = Image.open(io.BytesIO(resp.read())).convert("RGB")
|
|
except Exception as e:
|
|
logger.warning(f"No season background for {season} ({url}): {e}")
|
|
return None
|
|
# Cover-fit into 1000x1500: scale so both dims meet/exceed target, centre-crop.
|
|
sw, sh = bg.size
|
|
scale = max(POSTER_W / sw, POSTER_H / sh)
|
|
bg = bg.resize((int(sw * scale), int(sh * scale)), Image.LANCZOS)
|
|
nw, nh = bg.size
|
|
left = (nw - POSTER_W) // 2
|
|
top = (nh - POSTER_H) // 2
|
|
return bg.crop((left, top, left + POSTER_W, top + POSTER_H))
|
|
|
|
def _race_dark_overlay(self) -> Image.Image:
|
|
"""Dimming overlay for race posters when a map background is present.
|
|
Stronger at top (heading area) and middle (track), lighter elsewhere so
|
|
the underlying map detail still reads through."""
|
|
overlay = Image.new("RGBA", (POSTER_W, POSTER_H), (0, 0, 0, 0))
|
|
ovr = overlay.load()
|
|
for y in range(POSTER_H):
|
|
if y < 540:
|
|
# Strong dim under the FORMULA 1 / Round / GP-name block
|
|
alpha = 200
|
|
elif y < 1300:
|
|
# Mid-section behind the track — light dim
|
|
alpha = 110
|
|
else:
|
|
# Bottom (date strip + checker)
|
|
alpha = 200
|
|
for x in range(POSTER_W):
|
|
ovr[x, y] = (0, 0, 0, alpha)
|
|
return overlay
|
|
|
|
def _vertical_dark_gradient(self) -> Image.Image:
|
|
"""A vertical alpha gradient that's darkest top + bottom — keeps the photo's
|
|
midsection visible while making wordmark / year / footer text legible."""
|
|
overlay = Image.new("RGBA", (POSTER_W, POSTER_H), (0, 0, 0, 0))
|
|
ovr = overlay.load()
|
|
for y in range(POSTER_H):
|
|
if y < 500:
|
|
# Strong dim at top for the wordmark + year, fading to nothing at y=500
|
|
alpha = 220 - (y * 220 // 500)
|
|
elif y > 1100:
|
|
# Strong dim at bottom for "WORLD CHAMPIONSHIP" footer
|
|
alpha = (y - 1100) * 220 // (POSTER_H - 1100)
|
|
alpha = min(alpha, 220)
|
|
else:
|
|
# Mild dim through the middle to anchor text contrast
|
|
alpha = 60
|
|
for x in range(POSTER_W):
|
|
ovr[x, y] = (0, 0, 0, alpha)
|
|
return overlay
|
|
|
|
def render_season_poster(self, season: int) -> bytes:
|
|
cache_path = self.cache_dir / f"season-{season}.png"
|
|
if cache_path.exists():
|
|
return cache_path.read_bytes()
|
|
|
|
bg = self._fetch_season_background(season)
|
|
if bg is None:
|
|
# Fallback to the all-black layout if the background isn't available.
|
|
img = Image.new("RGB", (POSTER_W, POSTER_H), BG)
|
|
else:
|
|
# Composite the dimming gradient over the hero photo for text legibility.
|
|
img = Image.alpha_composite(bg.convert("RGBA"), self._vertical_dark_gradient()).convert("RGB")
|
|
|
|
draw = ImageDraw.Draw(img)
|
|
self._checker_strip(draw, 0)
|
|
self._centered(draw, "FORMULA 1", y=110, size=88, fill=RED)
|
|
self._centered(draw, str(season), y=240, size=240, fill=WHITE)
|
|
self._centered(draw, "WORLD CHAMPIONSHIP", y=POSTER_H - 200, size=48, fill=WHITE)
|
|
self._checker_strip(draw, POSTER_H - CHECKER_H)
|
|
|
|
buf = io.BytesIO()
|
|
img.save(buf, "PNG", optimize=True)
|
|
data = buf.getvalue()
|
|
cache_path.write_bytes(data)
|
|
return data
|