#!/usr/bin/env python3 """Oshkosh 10-second wallpaper — LOCAL composite for Hommer (Cinnamon). Fetches each camera's live frame DIRECTLY from YouTube's CDN (i.ytimg.com) to this machine, composites the 7-cam grid locally, writes a JPEG, and sets it as the desktop wallpaper. Loops every ~10s. Nothing proxies through dizydiz.com — same client-fetch model as the public Windows client. Stop via the flag file. """ import os, io, time, subprocess, urllib.request from datetime import datetime, timezone, timedelta from PIL import Image, ImageDraw, ImageFont W, H = 1920, 1080 CT = timezone(timedelta(hours=-5)) FLAG = "/tmp/oshkosh-10s.flag" OUTA = "/tmp/oshkosh-10s-a.jpg" OUTB = "/tmp/oshkosh-10s-b.jpg" FONT_B = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" FONT_R = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf" CAMS = [("Green Dot","CvBfCakKQOA"),("Warbirds","8XtERF62tGA"),("Boeing Plaza","rXH0zy6sLqo"), ("Vintage Tower","ZAsxWfxeaVY"),("Ultralights","frrnSfxoFkM"),("Seaplane Base","Kpr9ZJphux0"), ("Oshkosh LIVE","LIVEFEED")] # LIVEFEED (2026-07-20): EAA's produced broadcast from eaa.org/airventure/live (Brightcove # "FEATURED STREAM") during lineup.json show windows (show -30min → +4h, night → +3h), else # ADS-B radar. Schedule-gated because EAA keeps the stream live with an ambient plaza cam # after shows; the frame-test only covers off-air resolution. Re-evaluated every ~15 min. LIVEFEED = "LIVEFEED" LIVEFEED_EAA = "https://players.brightcove.net/627008079/default_default/index.html?videoId=6375380916112" LIVEFEED_FALLBACK = "-jSEC_rpBUo" # EAA Radio ADS-B radar (24/7) last = {} # id -> (Image, ts) freshest fetched frame (background threads write here) shown = {} # id -> (Image, ts) what's currently displayed; advanced ONE tile per tick hls = {} # id -> (url, resolved_ts) real HLS URL from yt-dlp, reused ~4h URL_TTL = 14400 def now(): return datetime.now(CT) def resolve(vid): u = hls.get(vid) ttl = 900 if vid == LIVEFEED else URL_TTL # re-evaluate the show window / feed every 15 min if u and time.time()-u[1] < ttl: return u[0] if vid == LIVEFEED: targets = ([LIVEFEED_EAA] if in_show_window() else []) + \ [f"https://www.youtube.com/watch?v={LIVEFEED_FALLBACK}"] else: targets = [f"https://www.youtube.com/watch?v={vid}"] for i, t in enumerate(targets): try: r = subprocess.run(["yt-dlp","-g","-f","best[height<=720]/best", t], capture_output=True, timeout=45) url = (r.stdout or b"").decode().strip().splitlines() if not url: continue if vid == LIVEFEED: # frame-test: live-event URLs can resolve while off-air t2 = subprocess.run(["ffmpeg","-loglevel","quiet","-i",url[0], "-frames:v","1","-f","null","-"], capture_output=True, timeout=25) if t2.returncode != 0: continue hls[vid] = (url[0], time.time(), i > 0); return url[0] except Exception: pass return None def fetch(vid): """Grab the REAL current video frame from YouTube's HLS stream (client-direct).""" url = resolve(vid) if url: try: r = subprocess.run(["ffmpeg","-y","-loglevel","error","-i",url, "-frames:v","1","-q:v","3",f"/tmp/o10-{vid}.jpg"], capture_output=True, timeout=15) if r.returncode == 0 and os.path.getsize(f"/tmp/o10-{vid}.jpg") > 15000: last[vid] = (Image.open(f"/tmp/o10-{vid}.jpg").convert("RGB"), time.time()); return except Exception: pass hls.pop(vid, None) # grab failed → re-resolve next pass # fallback: YouTube live thumbnail (poster card) so a tile is never blank try: tvid = LIVEFEED_FALLBACK if vid == LIVEFEED else vid req = urllib.request.Request(f"https://i.ytimg.com/vi/{tvid}/hqdefault_live.jpg", headers={"User-Agent":"Mozilla/5.0"}) data = urllib.request.urlopen(req, timeout=8).read() if len(data) > 5000: last[vid] = (Image.open(io.BytesIO(data)).convert("RGB"), time.time()) except Exception: pass import json as _json, re as _re def in_show_window(): """True while inside a lineup.json air-show window (show -30min → +4h, night → +3h).""" try: days = _json.load(open("/mnt/obsidian/Websites/DizyDiz/oshkosh/lineup.json"))["days"] except Exception: return True # no schedule readable -> let the frame-test decide n = now() d = days.get(n.strftime("%Y-%m-%d")) if not d: return False def win(t, hours): m = _re.search(r"(\d{1,2})(?::(\d{2}))?\s*(AM|PM)", t or "") if not m: return None hh = int(m.group(1)) % 12 + (12 if m.group(3) == "PM" else 0) s = n.replace(hour=hh, minute=int(m.group(2) or 0), second=0) - timedelta(minutes=30) return (s, s + timedelta(hours=hours)) ws = [w for w in (win(d.get("time"), 4), win(d.get("night"), 3)) if w] return any(a <= n <= b for a, b in ws) def todays_lineup(): """Today's air-show lineup from the vault lineup.json (local, no proxy).""" try: days = _json.load(open("/mnt/obsidian/Websites/DizyDiz/oshkosh/lineup.json"))["days"] except Exception: return None t = now().strftime("%Y-%m-%d") if t in days: return {**days[t], "_today": True} for k in sorted(days): if k > t: return {**days[k], "_today": False} return None _WX = {"t":0,"text":""} def kosh_metar(): if time.time()-_WX["t"] < 600: return _WX["text"] try: req = urllib.request.Request("https://aviationweather.gov/api/data/metar?ids=KOSH&format=raw", headers={"User-Agent":"dizydiz-oshkosh"}) raw = urllib.request.urlopen(req, timeout=10).read().decode().strip().splitlines() if raw: _WX.update(t=time.time(), text=_re.sub(r"^(METAR|SPECI)\s+","",raw[0]).split(" RMK")[0].strip()) except Exception: _WX["t"]=time.time()-480 return _WX["text"] _OVH = {"t":0,"n":None} def overhead_count(): if time.time()-_OVH["t"] < 90: return _OVH["n"] try: req = urllib.request.Request("https://api.airplanes.live/v2/point/43.9844/-88.5570/10", headers={"User-Agent":"dizydiz-oshkosh"}) d = _json.loads(urllib.request.urlopen(req, timeout=10).read()) _OVH.update(t=time.time(), n=len(d.get("ac") or [])) except Exception: _OVH["t"]=time.time()-60 return _OVH["n"] def fit(img, w, h): sw, sh = img.size; s = max(w/sw, h/sh) img = img.resize((int(sw*s)+1, int(sh*s)+1)) x=(img.width-w)//2; y=(img.height-h)//2 return img.crop((x,y,x+w,y+h)) def compose(): c = Image.new("RGB",(W,H),(8,8,12)); d = ImageDraw.Draw(c) cols, rows = 4, 2; tw, th = W//cols, (H-40)//rows lf = ImageFont.truetype(FONT_B,22); tf = ImageFont.truetype(FONT_R,16) for i,(name,vid) in enumerate(CAMS): x=(i%cols)*tw; y=(i//cols)*th if vid in shown: c.paste(fit(shown[vid][0], tw-2, th-2), (x+1,y+1)) ts = datetime.fromtimestamp(shown[vid][1],CT).strftime("%I:%M:%S %p") tc = (255,255,255) if time.time()-shown[vid][1] <= 120 else (255,195,70) else: ts, tc = "no signal yet", (255,120,120) wl=d.textlength(name,font=lf); wt=d.textlength(ts,font=tf) d.rectangle([x+8,y+8,x+34+wl+wt,y+42], fill=(0,0,0)) d.text((x+14,y+11),name,font=lf,fill=(255,255,255)) d.text((x+22+wl,y+15),ts,font=tf,fill=tc) # info tile — flips to the air-show lineup every 15s x=(7%cols)*tw; y=(7//cols)*th; cx=x+tw//2 d.rectangle([x+1,y+1,x+tw-1,y+th-1],fill=(14,20,38)) day = todays_lineup() if day and int(time.time()//15) % 2 == 1: hdr = "TODAY'S AIR SHOW" if day["_today"] else f"{day['label']} AIR SHOW" f=ImageFont.truetype(FONT_B,24); d.text((cx-d.textlength(hdr,font=f)/2,y+12),hdr,font=f,fill=(255,255,255)) f=ImageFont.truetype(FONT_R,17); d.text((cx-d.textlength(day['time'],font=f)/2,y+44),day['time'],font=f,fill=(120,200,255)) fl=ImageFont.truetype(FONT_R,16); ly=y+76 for p in day["performers"][:(th-108)//21]: d.text((x+22,ly),"• "+p,font=fl,fill=(215,220,230)); ly+=21 if day.get("night"): d.text((x+22,ly+2),"★ night show tonight",font=fl,fill=(255,210,90)) else: for txt,sz,col,fy in [("EAA AIRVENTURE",30,(255,255,255),0.16),("OSHKOSH 2026",30,(255,255,255),0.28), ("July 20 – 26",24,(120,200,255),0.44),("Air shows daily ~2:30 PM CT",18,(200,205,215),0.58), ("Green Dot cam runs 24/7",18,(200,205,215),0.68), ("Watch w/ audio:",17,(200,205,215),0.82),("eaa.org/airventure/live",18,(120,200,255),0.89)]: f=ImageFont.truetype(FONT_B if sz>23 else FONT_R,sz) d.text((cx-d.textlength(txt,font=f)/2,y+th*fy),txt,font=f,fill=col) d.rectangle([0,H-40,W,H],fill=(10,10,14)); ftr=ImageFont.truetype(FONT_R,19) right = f"{now().strftime('%b %d %I:%M:%S %p')} CT" if int(time.time()//15) % 2 == 0: left = "Oshkosh Camera Wallpaper — dizydiz.com/oshkosh" else: wx = kosh_metar() or "unavailable"; n = overhead_count() left = f"KOSH WX: {wx}" + (f" ✈ {n} aircraft within 10 nm of Wittman" if n is not None else "") d.text((14,H-31),left,font=ftr,fill=(230,230,235)) d.text((W-14-d.textlength(right,font=ftr),H-31),right,font=ftr,fill=(230,230,235)) return c def set_wallpaper(path): env = dict(os.environ, DISPLAY=":0", XDG_RUNTIME_DIR="/run/user/1000", DBUS_SESSION_BUS_ADDRESS="unix:path=/run/user/1000/bus") subprocess.run(["gsettings","set","org.cinnamon.desktop.background","picture-options","zoom"], env=env) subprocess.run(["gsettings","set","org.cinnamon.desktop.background","picture-uri",f"file://{path}"], env=env) def _bg_fetch(vid): """Background: keep this cam's freshest frame in `last`, refreshed ~every 8s.""" while os.path.exists(FLAG): try: fetch(vid) except Exception: pass time.sleep(8) def main(): open(FLAG,"w").close() import threading from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=len(CAMS)) as ex: # prime all tiles once list(ex.map(lambda cv: fetch(cv[1]), CAMS)) shown.update(last) # start fully populated for _,vid in CAMS: # background fetchers per cam threading.Thread(target=_bg_fetch, args=(vid,), daemon=True).start() tick, rr = 0, 0 STEP = 10.0 / len(CAMS) # reveal ONE freshly-fetched tile per ~1.4s -> smooth ripple, no flash while os.path.exists(FLAG): t0 = time.time() vid = CAMS[rr % len(CAMS)][1]; rr += 1 if vid in last: shown[vid] = last[vid] # advance just this one tile out = OUTA if tick % 2 == 0 else OUTB; tick += 1 try: compose().save(out, quality=82); set_wallpaper(out) except Exception as e: print("cycle error:", e, flush=True) time.sleep(max(0.3, STEP - (time.time()-t0))) if __name__ == "__main__": main()