Last active
September 3, 2026 06:29
-
-
Save dannyob/89dad7823de1050dd393664c4f79e5b0 to your computer and use it in GitHub Desktop.
Live Fortnite map position from the minimap: SIFT-matches your minimap against the island, serves position + nearest loot with bearings on a local web page. Passive screen capture only.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # /// script | |
| # requires-python = ">=3.10" | |
| # dependencies = ["opencv-python>=4.9", "numpy>=1.26", "mss>=9.0"] | |
| # /// | |
| """ | |
| Live Fortnite map position from the minimap, served as a web page. | |
| Reads your own screen, works out where you are on the island, and serves a map | |
| tile plus the nearest loot spawns with distance and compass bearing - so you can | |
| have it on a phone or second monitor while you play. | |
| It is passive: it screenshots your own display and never touches the game | |
| process, reads its memory, or sends it input. That is the same thing OBS does. | |
| uv run fortnite_position.py pick # 1. drag a box around your minimap | |
| uv run fortnite_position.py fetch # 2. download map + spawn data | |
| uv run fortnite_position.py # 3. serve on http://localhost:9999 | |
| Needs Fortnite in *Windowed Fullscreen* - exclusive fullscreen returns black | |
| frames to screen capture. | |
| -------------------------------------------------------------------------- | |
| HOW IT WORKS (and why it is built this way) | |
| The minimap is north-locked, fixed-zoom and player-centred, so locating it on a | |
| reference map of the island is a pure 2-D translation problem. Template matching | |
| seems the obvious tool and does not work - it finds no genuine correlation peak | |
| at any scale. SIFT features + RANSAC do, and they also self-report scale and | |
| rotation, which turn out to be the most useful signal in the whole pipeline: | |
| measured over hundreds of real frames the scale is constant to +-0.001 and the | |
| rotation is 0.0 +- 0.1 deg. | |
| That gives a much better confidence test than counting matched features. A match | |
| landing on the established scale with no rotation is almost certainly right even | |
| with very few inliers, so the acceptance logic leans on geometry and positional | |
| continuity rather than raw counts. Three refinements earned their place, each | |
| after a measured failure: | |
| * CLAHE before feature detection. The storm dims the minimap; without local | |
| contrast equalisation matching collapses entirely once the overlay passes | |
| ~0.6 strength. | |
| * A second, looser ratio test. In low-texture terrain (desert canyon, uniform | |
| forest) the true match beats its look-alikes only narrowly, and a strict | |
| ratio test discards it before RANSAC ever sees it. Over 20 real failures the | |
| strict pass recovered 0 and a 0.92 pass recovered all 20. | |
| * A spatially gated retry. Restricting candidates to where you could plausibly | |
| be removes distant look-alikes entirely; in uniform forest this turned | |
| 3-inlier scatter into 7-8 inlier consensus. | |
| The last two are fallbacks that only run after the strict pass fails, so they | |
| cannot degrade ordinary tracking - and every geometry and continuity gate still | |
| applies to whatever they return. | |
| COORDINATE TRANSFORM | |
| Positions come out in real Fortnite world units (uu; 100 uu = 1 m), not | |
| arbitrary pixels. The chain is exact, not fitted by hand: | |
| * fortnite.gg draws its map with Leaflet using L.CRS.Simple centred on | |
| [-128, 128], so tile-pyramid pixels are px = (2^z * gg_y, -2^z * gg_x). | |
| * Its coordinate frame maps to world units by a fit over 13 named POIs whose | |
| world positions come from fortnite-api.com. Residual is 4.5 uu - about 4 cm | |
| across a 2.5 km island - and the error is pure rounding of gg's 2-decimal | |
| coordinates, so the transform is exact rather than approximate. | |
| Composing the two gives the axis-aligned transform in world_from_px() below. | |
| -------------------------------------------------------------------------- | |
| """ | |
| import argparse | |
| import collections | |
| import concurrent.futures as cf | |
| import hashlib | |
| import http.server | |
| import io | |
| import json | |
| import math | |
| import os | |
| import re | |
| import socket | |
| import socketserver | |
| import threading | |
| import time | |
| import urllib.request | |
| import cv2 | |
| import numpy as np | |
| # ========================================================================== | |
| # CONFIGURATION - the bits you may need to change | |
| # ========================================================================== | |
| # Fortnite map build. Changes every season, and the map tiles are versioned by | |
| # it. Leave as None and `fetch` tries the newest builds in turn, keeping the | |
| # first that the live minimap actually matches - fortnite.gg also publishes | |
| # builds that are not the map in play, so newest is not always right. Have | |
| # Fortnite on screen when you fetch so the check can run. Pin a string like | |
| # "42.00" to skip all that; the live value is in the devtools console on | |
| # https://fortnite.gg as: Data.map | |
| MAP_BUILD = None | |
| # Reference zoom. Leave as None and `fetch` picks it from your minimap size: | |
| # the reference has to be at least as detailed as the minimap or the features do | |
| # not correspond and nothing matches. Roughly, zoom 4 (4096px, 0.73 m/px) suits a | |
| # 4K/5K screen and zoom 3 (2048px) a 1080p one. Set an integer to override. | |
| ZOOM = None | |
| # Spawn categories to download. None means "every non-empty category | |
| # fortnite.gg publishes" (~30 of them) - you then pick which ones to show from | |
| # the checkboxes at the bottom of the page, so there is no need to decide here. | |
| # Set an explicit list only if you want a smaller download. | |
| ITEM_LAYERS = None | |
| # Web server. BIND accepts "local" (localhost only), "tailscale" (also reachable | |
| # from your other tailnet devices - handy for viewing on a phone), "all", or an | |
| # explicit comma-separated list of addresses. | |
| PORT = 9999 | |
| BIND = "local" | |
| # How many nearby items to list. | |
| TOP_N = 8 | |
| # Seconds between fixes. A fix takes ~0.15 s, so 0.25 is achievable if you want | |
| # a smoother trace; 1.0 is plenty for knowing where you are. | |
| INTERVAL = 1.0 | |
| # ========================================================================== | |
| # Transform constants - derived, do not tune by hand. See the docstring. | |
| # ========================================================================== | |
| GG_SCALE_X, GG_SCALE_Y = 1171.571, 1171.572 | |
| ORIGIN_X, ORIGIN_Y = -141584.8, -149564.3 | |
| HERE = os.path.dirname(os.path.abspath(__file__)) | |
| REF_PNG = os.path.join(HERE, "reference.png") | |
| REF_META = os.path.join(HERE, "reference.json") | |
| REF_FEAT = os.path.join(HERE, "reference_feat.npz") | |
| REGION_JSON = os.path.join(HERE, "region.json") | |
| SPAWNS_JSON = os.path.join(HERE, "spawns.json") | |
| STATE_JSON = os.path.join(HERE, "state.json") | |
| UA = {"User-Agent": "Mozilla/5.0 (personal map tool)"} | |
| TAILSCALE_EXE = "C:/Program Files/Tailscale/tailscale.exe" | |
| COMPASS = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE", | |
| "S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW"] | |
| PALETTE = [(255, 190, 60), (80, 220, 255), (140, 255, 140), | |
| (120, 120, 255), (200, 160, 255), (160, 255, 220)] | |
| RESCUABLE = {"low_inliers", "few_candidates", "no_transform", "rotated", "bad_scale"} | |
| def load(path, default=None): | |
| if not os.path.exists(path): | |
| return default | |
| with open(path, encoding="utf-8-sig") as fh: # -sig: tolerate a BOM | |
| return json.load(fh) | |
| def save(path, obj): | |
| with open(path, "w", encoding="utf-8") as fh: | |
| json.dump(obj, fh, indent=2) | |
| def get(url, timeout=30): | |
| return urllib.request.urlopen( | |
| urllib.request.Request(url, headers=UA), timeout=timeout).read() | |
| # ========================================================================== | |
| # Step 1 - choose the minimap region | |
| # ========================================================================== | |
| def cmd_pick(_args): | |
| """Drag a box around the minimap. Fortnite must be visible on screen.""" | |
| import mss | |
| with getattr(mss, "MSS", mss.mss)() as sct: | |
| mon = sct.monitors[0] | |
| shot = cv2.cvtColor(np.asarray(sct.grab(mon)), cv2.COLOR_BGRA2BGR) | |
| # Fit the (possibly huge) desktop into a window that fits on the desktop. | |
| scale = min(1.0, 1600.0 / shot.shape[1]) | |
| view = cv2.resize(shot, None, fx=scale, fy=scale, interpolation=cv2.INTER_AREA) | |
| print("Drag a box around the minimap, then press ENTER (or c to cancel).") | |
| x, y, w, h = cv2.selectROI("select the minimap", view, showCrosshair=False) | |
| cv2.destroyAllWindows() | |
| if w < 8 or h < 8: | |
| print("cancelled") | |
| return | |
| region = {"X": int(x / scale) + mon["left"], "Y": int(y / scale) + mon["top"], | |
| "Width": int(w / scale), "Height": int(h / scale)} | |
| save(REGION_JSON, region) | |
| print(f"saved {REGION_JSON}: {region}") | |
| preview = shot[int(y / scale):int((y + h) / scale), | |
| int(x / scale):int((x + w) / scale)] | |
| cv2.imwrite(os.path.join(HERE, "region_preview.png"), preview) | |
| print("wrote region_preview.png - check it shows the minimap and nothing else") | |
| # ========================================================================== | |
| # Step 2 - fetch the reference map and the spawn data | |
| # ========================================================================== | |
| def tile(z, x, y, build, retries=3): | |
| for attempt in range(retries): | |
| try: | |
| data = get(f"https://fortnite.gg/maps/{build}/{z}/{x}/{y}.webp") | |
| return x, y, cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_COLOR) | |
| except Exception: | |
| if attempt == retries - 1: | |
| return x, y, None | |
| return x, y, None | |
| def discover_build(hint=None): | |
| """Find the newest map build that actually serves tiles. | |
| The build is only published inside the page HTML, which sits behind | |
| Cloudflare, so probe the tile CDN instead: ask for one tile per candidate | |
| and keep the highest that answers. | |
| """ | |
| import urllib.error | |
| def exists(build): | |
| try: | |
| urllib.request.urlopen(urllib.request.Request( | |
| f"https://fortnite.gg/maps/{build}/1/0/0.webp", headers=UA), | |
| timeout=12).read(1) | |
| return True | |
| except Exception: | |
| return False | |
| # Minor versions are consecutive hotfixes, not steps of ten: season 41 ran | |
| # to 41.20 and season 42 was already on 42.02 within hours. So scan every | |
| # minor from 0 up, stopping after a run of misses. | |
| base = int(hint.split(".")[0]) if hint else 41 | |
| found = [] | |
| with cf.ThreadPoolExecutor(max_workers=12) as ex: | |
| for maj in range(base, base + 4): | |
| cands = [f"{maj}.{m:02d}" for m in range(0, 40)] | |
| hits = [b for b, ok in zip(cands, ex.map(exists, cands)) if ok] | |
| found += hits | |
| if not hits and maj > base: | |
| break # no such season yet; stop climbing | |
| if not found: | |
| raise SystemExit("could not find any map build - is fortnite.gg reachable?") | |
| ranked = sorted(found, key=lambda b: tuple(int(x) for x in b.split(".")), | |
| reverse=True) | |
| print(f"map builds available: {', '.join(ranked)}") | |
| return ranked | |
| def fetch_map(build, zoom): | |
| n = 2 ** zoom | |
| _, _, probe = tile(zoom, 0, 0, build) | |
| if probe is None: | |
| raise SystemExit( | |
| f"could not fetch tiles for build {build!r}.\n" | |
| f"The build is almost certainly out of date - see MAP_BUILD at the " | |
| f"top of this file for how to find the current one.") | |
| ts = probe.shape[0] | |
| side = ts * n | |
| print(f"map {build} zoom {zoom}: {n}x{n} tiles of {ts}px -> {side}x{side}") | |
| canvas = np.zeros((side, side, 3), np.uint8) | |
| missing = done = 0 | |
| with cf.ThreadPoolExecutor(max_workers=8) as ex: | |
| futs = [ex.submit(tile, zoom, x, y, build) | |
| for x in range(n) for y in range(n)] | |
| for fut in cf.as_completed(futs): | |
| x, y, img = fut.result() | |
| done += 1 | |
| if img is None: | |
| missing += 1 | |
| else: | |
| canvas[y * ts:(y + 1) * ts, x * ts:(x + 1) * ts] = img | |
| if done % 32 == 0 or done == len(futs): | |
| print(f" {done}/{len(futs)} tiles", end="\r", flush=True) | |
| print() | |
| if missing: | |
| print(f"WARNING: {missing} tiles failed and are blank") | |
| cv2.imwrite(REF_PNG, canvas) | |
| save(REF_META, {"map": build, "zoom": zoom, "size": side, | |
| "uu_per_px_x": GG_SCALE_X / (2 ** zoom), | |
| "uu_per_px_y": GG_SCALE_Y / (2 ** zoom), | |
| "origin_x": ORIGIN_X, "origin_y": ORIGIN_Y}) | |
| if os.path.exists(REF_FEAT): | |
| os.remove(REF_FEAT) # stale cache for the old map | |
| print(f"wrote {REF_PNG} ({GG_SCALE_X / (2 ** zoom) / 100:.2f} m/px)") | |
| def as_point(c): | |
| """Most categories store [x, y]; a few (ziplines) store two endpoints.""" | |
| if len(c) == 2 and all(isinstance(v, (int, float)) for v in c): | |
| return c | |
| if len(c) == 2 and all(isinstance(v, (list, tuple)) and len(v) == 2 for v in c): | |
| return [(c[0][0] + c[1][0]) / 2.0, (c[0][1] + c[1][1]) / 2.0] # midpoint | |
| return None | |
| def fetch_spawns(layers): | |
| """Pull spawn points from fortnite.gg and convert them to world units. | |
| Fetched at runtime rather than shipped: it is their data, and this way it is | |
| always current. The file is public and needs no login. | |
| """ | |
| # Cache-bust: the bare URL is served from a Cloudflare cache that can be | |
| # weeks stale (observed 20 days old, a whole season behind). The site | |
| # itself always requests this with a ?v= parameter for the same reason. | |
| url = f"https://fortnite.gg/data/spawns.js?v={int(time.time())}" | |
| txt = get(url).decode("utf-8") | |
| m = re.match(r"\s*window\.Spawns\s*=\s*(\{.*\})\s*;?\s*$", txt, re.S) | |
| if not m: | |
| raise SystemExit("spawns.js is not in the expected form - site changed?") | |
| spawns = json.loads(m.group(1)) | |
| if not layers: # everything with points in it | |
| layers = sorted(k for k, v in spawns.items() | |
| if isinstance(v, list) | |
| and any(g.get("coords") for g in v)) | |
| print(f" taking all {len(layers)} non-empty categories") | |
| out = {} | |
| for name in layers: | |
| groups = spawns.get(name) | |
| if not isinstance(groups, list): | |
| print(f" '{name}' not found - check the key against spawns.js") | |
| continue | |
| pts = [] | |
| for g in groups: | |
| for c in g.get("coords", []): | |
| pt = as_point(c) | |
| if pt is not None: | |
| gx, gy = pt | |
| pts.append([round(GG_SCALE_X * gy + ORIGIN_X), | |
| round(GG_SCALE_Y * -gx + ORIGIN_Y)]) | |
| if pts: | |
| out[name] = pts | |
| print(f" {name}: {len(pts)}") | |
| # Season-specific layers (sprite chests, vaults, extraction sites) are NOT | |
| # in this file - they live in the page HTML, behind Cloudflare. To add them, | |
| # open https://fortnite.gg, and run this in the devtools console: | |
| # | |
| # copy(JSON.stringify(Object.fromEntries(Object.entries( | |
| # Data.data.spawns.sub).map(([k,v])=>[k,(v.markers||[]).flatMap( | |
| # m=>m.coords||[]).map(([x,y])=>[Math.round(1171.571*y-141584.8), | |
| # Math.round(-1171.572*x-149564.3)])])))) | |
| # | |
| # then paste into spawns.json, merging with what is written here. | |
| if not out: | |
| raise SystemExit("no layers resolved - nothing to show") | |
| save(SPAWNS_JSON, out) | |
| print(f"wrote {SPAWNS_JSON}") | |
| # The minimap always covers about the same slice of the island regardless of | |
| # screen size - measured at ~35,000 uu (350 m) across. So the pixel width of your | |
| # capture region tells us its resolution, and hence which reference zoom matches. | |
| MINIMAP_SPAN_UU = 34955 | |
| def recommend_zoom(region): | |
| z = math.log2(GG_SCALE_X * region["Width"] / MINIMAP_SPAN_UU) | |
| return max(2, min(5, round(z))) | |
| def verify_build(zoom): | |
| """Score how well the current on-screen minimap matches reference.png. | |
| fortnite.gg publishes builds that are not the live Battle Royale map, so a | |
| higher build number does not mean a better one: 42.02 and 42.03 scored 4-5 | |
| junk inliers while 42.00 - the map actually in play - scored 384. The only | |
| reliable test is to match the real minimap against the candidate. | |
| """ | |
| region = load(REGION_JSON) | |
| if not region: | |
| return None | |
| try: | |
| import mss | |
| box = {"left": int(region["X"]), "top": int(region["Y"]), | |
| "width": int(region["Width"]), "height": int(region["Height"])} | |
| with getattr(mss, "MSS", mss.mss)() as sct: | |
| live = cv2.cvtColor(np.asarray(sct.grab(box)), cv2.COLOR_BGRA2BGR) | |
| ref = cv2.imread(REF_PNG, cv2.IMREAD_GRAYSCALE) | |
| if ref is None: | |
| return None | |
| sift = cv2.SIFT_create(nfeatures=0) | |
| clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8)) | |
| kp, d1 = sift.detectAndCompute( | |
| clahe.apply(cv2.cvtColor(live, cv2.COLOR_BGR2GRAY)), None) | |
| k2, d2 = sift.detectAndCompute(ref, None) | |
| if d1 is None or d2 is None or len(kp) < 50: | |
| return None | |
| pts = np.float32([k.pt for k in k2]) | |
| fl = cv2.FlannBasedMatcher(dict(algorithm=1, trees=5), dict(checks=64)) | |
| fl.add([d2.astype(np.float32)]) | |
| fl.train() | |
| raw = fl.knnMatch(d1.astype(np.float32), k=2) | |
| good = [a for a, b in (q for q in raw if len(q) == 2) | |
| if a.distance < 0.92 * b.distance] | |
| if len(good) < 6: | |
| return 0 | |
| s = np.float32([kp[g.queryIdx].pt for g in good]).reshape(-1, 1, 2) | |
| t = np.float32([pts[g.trainIdx] for g in good]).reshape(-1, 1, 2) | |
| M, inl = cv2.estimateAffinePartial2D(s, t, method=cv2.RANSAC, | |
| ransacReprojThreshold=4.0, maxIters=20000) | |
| if M is None: | |
| return 0 | |
| rot = abs(math.degrees(math.atan2(M[1, 0], M[0, 0]))) | |
| return int(inl.sum()) if rot < 5 or rot > 355 else 0 | |
| except Exception: | |
| return None | |
| def cmd_fetch(args): | |
| zoom = args.zoom | |
| if zoom is None: | |
| region = load(REGION_JSON) | |
| if not region: | |
| raise SystemExit("run `pick` first, or set ZOOM explicitly") | |
| zoom = recommend_zoom(region) | |
| print(f"minimap region is {region['Width']}px wide -> using zoom {zoom} " | |
| f"({256 * 2 ** zoom}px reference)") | |
| if args.map: | |
| fetch_map(args.map, zoom) | |
| else: | |
| # Try the newest builds in turn and keep the first that the live minimap | |
| # actually matches. Needs Fortnite on screen; without it, take the newest | |
| # and say so, because the check cannot be made. | |
| cands = discover_build() | |
| for i, build in enumerate(cands[:3]): | |
| fetch_map(build, zoom) | |
| score = verify_build(zoom) | |
| if score is None: | |
| print(" (cannot verify - Fortnite not on screen; keeping this build)") | |
| break | |
| if score >= 15: | |
| print(f" verified against the live minimap: {score} inliers") | |
| break | |
| print(f" build {build} does not match what is on screen " | |
| f"({score} inliers) - trying the next one") | |
| else: | |
| print(" WARNING: no candidate matched the live minimap; " | |
| "pin one with --map once you know which is right") | |
| fetch_spawns(ITEM_LAYERS) | |
| # ========================================================================== | |
| # Step 3 - track and serve | |
| # ========================================================================== | |
| # ========================================================================== | |
| # Compass heading | |
| # ========================================================================== | |
| # Fortnite prints the exact bearing as a number above the compass caret, centred | |
| # on screen centre. The font is fixed, so ten digit templates classify it by | |
| # normalised correlation - no OCR dependency. The templates below were measured | |
| # from real frames and are stored as a zlib+base64 blob. | |
| # | |
| # Geometry is expressed as fractions of the screen so it survives a change of | |
| # resolution; the templates themselves are size-normalised before matching. | |
| DIGIT_B64 = ( | |
| "eNrVmX9MVVUcwE8N3338fFIMxwOizdd07cXKpfPNpa1a4+lSy1FYEQzZGzmvZQ23tuLNcM3QmFBkTmu1tLEny7QmOenH" | |
| "HA+KIINkMRjIIhfGD6lAUnOne++759e938PzreHk+9e5n3fuud/zPd9zzvf7fQgZ4sdEPIhJH+akj+ITWJTaCK7FVlF1" | |
| "rGK7+DW+E+Z+DIkHFbC2B0Nt4V3GLTp4ouk2I5fNca5zd1T7ROXGmqrAeonrIl1fkPtn8B/pHGX+afFnfII5Oo+RICr/" | |
| "RauEzB/zefjE9PS/Jr9ymNKJiSnuG6EIPDA8LGqq8zWDg1PWiYVQT895wA4hlHsSg5zqdr08txLmKD9GnlOotwq/x7b5" | |
| "4vdKS8XvR/gWD0IQF2z8f3mlhD8o4XdIeK6Eb5tl/W8oP74A5tzuiJU3G4/HhP2zXehaQnCJxavWGnSF3d1WaHgR5M+L" | |
| "JBynpeGYePQxZTrcAM7pnMabVcaFuTAZAXmzRQdTjtnWRZc3r2Mdt0egZS7vAycwxs3gyTyC4BObX8e5wulESobsfJ6i" | |
| "vEtsrDho16SjfNd+l2JQZ3qHZfA9Tp1vtn/1Ze2NxN2AOs/GxRWBet6ZwS7Tn89cYnwxaV3ryEo5RK+xDZRf0g/1/fQF" | |
| "yrvdMN+JBO6uNyVP4Md5x00p+obqY6LsgCav0lE+vx1Z41tdTt1Fhljcz+jZqvvZ2IcZP7OK++bb3DCnl8McN1Wl0vih" | |
| "oVuwJ+Ferz8QYBHMJx8a8o7P+DHzC8syXG6t9+o/FEzBMU/qD+SR7s8O3c5b6d5+ONq6XOsIh8PDdm4RZSuIH0XrIFwY" | |
| "h+J32fGueE2HhNeseE+CMXuHq5+n47cpxGgORcHAvrCuL4330MLLMPfA8SF/FGpSIzltaP+TEt6LwXH2TsP9yRbbKPLN" | |
| "pjbFHjjP8ojc3wtzEsHuTvbw+nhbaGQr9CeWPJIj8LQG5m/8OKTduhQJ/au5Ns/N5sA6JI5DToctmuwwH77V2niOSCcn" | |
| "owT+1tnJO2mQeHShmEZ4bH4o8hoUW/+ZeBaovU/K37pZeYs4R8KHUMx849lIs5rf6UM+dnR8ebcs3vDFyHFWdC7J0Yz5" | |
| "ttDH4EGYi7OMzon8uQnm1vGjcov+8DrigxJDZGX9DnZPQT7YzmAeGsnHl1u0a6HXdWiM0V+F5B1t2BGKfE+QWxSlzui9" | |
| "TeHvnaT7+HgsgeD5A8J3m9LjDZz6NXh3z//Ipv/r2lBJb8RQH1MV50v0oSPcDdXHfsoE62OyulnnEoSWHCWS8YKlxmK7" | |
| "GXWeWaTJ41b+VCTOuRAIeHm+9g/abshh/LNf+HVPlewXT4z8APuWqqr7MFzfy66PsR7I+PO83aLXx7CwLstYwCBZxyel" | |
| "dTPQT8oc8rpZ0gc2XGEkDElNGPRblODCoJ8jpChm2YPui6BkAWLlN5sITt8I8wfg/g9xcTyH8xie5PBjOCof588rhi/w" | |
| "unRRPBTkMKuiizci4wN89+dY9wEUE+8JwpWHHh6XzjL/TsJZyt0lUbMLzQY3NkjwtI3fqiinzBf2KjTbQclCbRIvMDee" | |
| "65zFNxsTDWyvBNdqbyTXAd78ooP3BE688a+Q5t/t7e1X6fnjJa2rddq3vmIvkMaY+D9CrDy7BeSNVM9JvRBAT1kvWvqj" | |
| "2ezVMv5JcmRpSe+aftus2vQzGRVdtOC+vIhBy0U8/jSCeeUjxr6tso6DW+/V7so2wJyNGW56qPJ64iPUDuK8mH02ifah" | |
| "FneLvEzCvcyCMJeto7ju2FkO+sl62H80v0qoBXiB9q3EdJvfpjtn8nNjXyjn7ftCr4+ZfRdy4ZjDxdfH2lxO846y1hn2" | |
| "GeM7ysCwKw76W0N1SvOj1cw92ntpfazYQfFflWhZ4z/2OF+Pn1f12Xm1EFcX0rKNCsfbkVP80BUrH1ut85UjcNyOhmz8" | |
| "nB4b5FPXZiFSZ3l5+WDUBCZlv+TedH8suU89KomHP1XVi/y96QtE5B5OT/EQZfMiZcI6iR2GJPnOtMTOeKICoYoJOt31" | |
| "9IfRcHiUy1PAWRU75j0jyVN8sF/J/FDz2/R9kN/qFV9XG+DnkQJZjQFX0jTlPxDNY1g=" | |
| ) | |
| TPL_W, TPL_H = 24, 34 | |
| HEAD_BOX_W, HEAD_BOX_Y0, HEAD_BOX_H = 0.04297, 0.00833, 0.04352 | |
| HEAD_BAND = (0.213, 0.660) # digit band inside the crop | |
| HEAD_DW = (0.00234, 0.00664) # plausible digit width (fraction of screen) | |
| HEAD_DH = (0.01296, 0.01852) # plausible digit height (fraction of screen) | |
| class HeadingReader: | |
| """Reads the numeric bearing off the compass. Returns None when unsure. | |
| A wrong heading is worse than no heading - it would send you the wrong way - | |
| so this declines rather than guesses. Two checks catch the common failure, | |
| which is a HUD marker icon merging with a digit and leaving a partial read: | |
| the number must stay centred, and the digits must sit shoulder to shoulder. | |
| Measured over 42 labelled frames: 34 correct, 0 wrong, 8 declined. | |
| """ | |
| def __init__(self, screen_w, screen_h): | |
| import base64, zlib | |
| raw = zlib.decompress(base64.b64decode(DIGIT_B64)) | |
| n = TPL_W * TPL_H | |
| self.tpl = {d: np.frombuffer(raw[i * n:(i + 1) * n], np.uint8) | |
| .reshape(TPL_H, TPL_W).astype(np.float32) / 255.0 | |
| for i, d in enumerate("0123456789")} | |
| w = int(screen_w * HEAD_BOX_W) | |
| self.box = {"left": int(screen_w / 2 - w / 2), "top": int(screen_h * HEAD_BOX_Y0), | |
| "width": w, "height": int(screen_h * HEAD_BOX_H)} | |
| self.band = (int(self.box["height"] * HEAD_BAND[0]), | |
| int(self.box["height"] * HEAD_BAND[1])) | |
| self.dw = (screen_w * HEAD_DW[0], screen_w * HEAD_DW[1]) | |
| self.dh = (screen_h * HEAD_DH[0], screen_h * HEAD_DH[1]) | |
| self.centre = self.box["width"] / 2.0 | |
| def _candidate_masks(self, bgr): | |
| """No single threshold isolates the glyphs on every backdrop. | |
| Against dark scenery plain white works. Against a HUD marker icon the | |
| glyph outline separates them. Against bright cloud or snow there is no | |
| outline at all (measured: 0% of the band below V=115, 99% above V=205) | |
| and only a strict cut picks the glyphs out. Each is tried in turn and the | |
| result is validated the same way, so a bad mask is simply rejected. | |
| """ | |
| hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV) | |
| v, s = hsv[:, :, 2], hsv[:, :, 1] | |
| white = ((v > 205) & (s < 60)).astype(np.uint8) * 255 | |
| cut = cv2.dilate((v < 115).astype(np.uint8) * 255, | |
| np.ones((3, 3), np.uint8), iterations=1) | |
| return [white, | |
| cv2.bitwise_and(white, cv2.bitwise_not(cut)), | |
| ((v > 238) & (s < 25)).astype(np.uint8) * 255] | |
| def _boxes(self, m_full): | |
| m = m_full[self.band[0]:self.band[1]] | |
| n, _, st, _ = cv2.connectedComponentsWithStats(m, 8) | |
| b = [(st[j, 0], st[j, 1], st[j, 2], st[j, 3]) for j in range(1, n) | |
| if self.dw[0] <= st[j, 2] <= self.dw[1] | |
| and self.dh[0] <= st[j, 3] <= self.dh[1]] | |
| return sorted(b), m | |
| def read(self, bgr): | |
| for mask in self._candidate_masks(bgr): | |
| val = self._read_mask(mask) | |
| if val is not None: | |
| return val | |
| return None | |
| def _read_mask(self, mask): | |
| b, m = self._boxes(mask) | |
| if not (1 <= len(b) <= 3): | |
| return None | |
| span_c = (b[0][0] + b[-1][0] + b[-1][2]) / 2.0 | |
| if abs(span_c - self.centre) > max(4.0, self.box["width"] * 0.02): | |
| return None # off-centre: digits are missing | |
| gap = max(2.0, self.box["width"] * 0.055) | |
| for p, q in zip(b, b[1:]): | |
| if q[0] - (p[0] + p[2]) > gap: | |
| return None # a digit was lost to an overlay | |
| out, worst = "", 1.0 | |
| for (x, y, w, h) in b: | |
| v = cv2.resize(m[y:y + h, x:x + w], (TPL_W, TPL_H), | |
| interpolation=cv2.INTER_AREA).astype(np.float32) / 255.0 | |
| best, score = None, -1.0 | |
| for d, t in self.tpl.items(): | |
| s = float(cv2.matchTemplate(v, t, cv2.TM_CCOEFF_NORMED)[0][0]) | |
| if s > score: | |
| best, score = d, s | |
| out += best | |
| worst = min(worst, score) | |
| if worst < 0.60: | |
| return None | |
| val = int(out) | |
| return val if 0 <= val <= 359 else None | |
| def relative_bearing(bearing, heading): | |
| """Signed turn from where you are facing to `bearing`: +right, -left.""" | |
| return ((bearing - heading + 540.0) % 360.0) - 180.0 | |
| class Tracker: | |
| def __init__(self, args): | |
| self.args = args | |
| self.region = load(REGION_JSON) | |
| if not self.region: | |
| raise SystemExit("no region.json - run: pick") | |
| self.meta = load(REF_META) | |
| self.loot = load(SPAWNS_JSON) | |
| if not self.meta or self.loot is None: | |
| raise SystemExit("no reference/spawn data - run: fetch") | |
| self.ref_bgr = cv2.imread(REF_PNG, cv2.IMREAD_COLOR) | |
| if self.ref_bgr is None: | |
| raise SystemExit("reference.png missing - run: fetch") | |
| self.sift = cv2.SIFT_create(nfeatures=0) | |
| self.clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8)) | |
| self.ref_pts, desc = self._reference_features() | |
| self.flann = cv2.FlannBasedMatcher(dict(algorithm=1, trees=5), | |
| dict(checks=64)) | |
| self.flann.add([desc.astype(np.float32)]) | |
| self.flann.train() | |
| # A fixed palette runs out at ~30 layers, so spread hues evenly instead; | |
| # alternating lightness keeps neighbouring hues apart. | |
| names = sorted(self.loot) | |
| self.colors = {} | |
| for i, k in enumerate(names): | |
| hue = int(179 * i / max(1, len(names))) | |
| val = 255 if i % 2 == 0 else 205 | |
| bgr = cv2.cvtColor(np.uint8([[[hue, 180, val]]]), cv2.COLOR_HSV2BGR)[0][0] | |
| self.colors[k] = (int(bgr[0]), int(bgr[1]), int(bgr[2])) | |
| self.heading_reader = None | |
| if not args.no_heading: | |
| try: | |
| import mss | |
| with getattr(mss, "MSS", mss.mss)() as sct: | |
| mon = sct.monitors[0] | |
| self.heading_reader = HeadingReader(mon["width"], mon["height"]) | |
| print(f"heading readout enabled ({mon['width']}x{mon['height']})") | |
| except Exception as exc: | |
| print(f"heading readout unavailable: {exc}") | |
| self.state = {"ok": False, "msg": "starting"} | |
| self.lock = threading.Lock() | |
| self.last_good = None | |
| self._last_epoch = None | |
| self.scale_hist = collections.deque(maxlen=60) | |
| # The learned scale is tied to this reference: a different ZOOM changes | |
| # it, so discard a remembered value that belongs to another build. | |
| st = load(STATE_JSON) or {} | |
| self._seed_scale = (st.get("expect_scale") | |
| if st.get("zoom") == self.meta["zoom"] | |
| and st.get("map") == self.meta["map"] else None) | |
| self._last_miss = 0.0 | |
| def _reference_features(self): | |
| """~6 s over a 4096px map, so cache it next to the image.""" | |
| stamp = int(os.path.getmtime(REF_PNG)) | |
| if os.path.exists(REF_FEAT): | |
| try: | |
| z = np.load(REF_FEAT) | |
| if int(z["stamp"]) == stamp: | |
| print(f"loaded {len(z['pts'])} cached reference features") | |
| return z["pts"], z["desc"] | |
| except Exception: | |
| pass | |
| print("computing reference features (one-off, ~6s)…") | |
| kp, desc = self.sift.detectAndCompute( | |
| cv2.cvtColor(self.ref_bgr, cv2.COLOR_BGR2GRAY), None) | |
| pts = np.float32([k.pt for k in kp]) | |
| np.savez_compressed(REF_FEAT, pts=pts, desc=desc, stamp=stamp) | |
| print(f"computed {len(pts)} reference features") | |
| return pts, desc | |
| # --- capture --------------------------------------------------------- | |
| def grab(self): | |
| import mss | |
| r = self.region | |
| box = {"left": int(r["X"]), "top": int(r["Y"]), | |
| "width": int(r["Width"]), "height": int(r["Height"])} | |
| with getattr(mss, "MSS", mss.mss)() as sct: | |
| return cv2.cvtColor(np.asarray(sct.grab(box)), cv2.COLOR_BGRA2BGR) | |
| def grab_heading(self): | |
| if self.heading_reader is None: | |
| return None | |
| import mss | |
| with getattr(mss, "MSS", mss.mss)() as sct: | |
| raw = np.asarray(sct.grab(self.heading_reader.box)) | |
| return self.heading_reader.read(cv2.cvtColor(raw, cv2.COLOR_BGRA2BGR)) | |
| # --- geometry -------------------------------------------------------- | |
| def world_from_px(self, px, py): | |
| m = self.meta | |
| return (m["uu_per_px_x"] * px + m["origin_x"], | |
| m["uu_per_px_y"] * py + m["origin_y"]) | |
| def px_from_world(self, wx, wy): | |
| m = self.meta | |
| return ((wx - m["origin_x"]) / m["uu_per_px_x"], | |
| (wy - m["origin_y"]) / m["uu_per_px_y"]) | |
| def expected_scale(self): | |
| if len(self.scale_hist) >= 3: | |
| return float(np.median(self.scale_hist)) | |
| return self._seed_scale | |
| def continuity_ok(self, world): | |
| """Reject a low-count match that teleports away from the last fix.""" | |
| if not self.last_good or self._last_epoch is None: | |
| return False | |
| dt = max(0.0, time.time() - self._last_epoch) | |
| if dt > 300: | |
| return False | |
| lx, ly = self.last_good["world"] | |
| return math.hypot(world[0] - lx, world[1] - ly) <= 4000.0 * dt + 10000.0 | |
| def gated_matches(self, desc, k=8): | |
| """Ratio test restricted to reference features near the last fix.""" | |
| if not self.last_good or self._last_epoch is None: | |
| return [] | |
| dt = max(0.0, time.time() - self._last_epoch) | |
| if dt > 20.0: | |
| return [] | |
| radius = (4000.0 * dt + 15000.0) / self.meta["uu_per_px_x"] | |
| cx, cy = self.last_good["px"] | |
| out = [] | |
| for nb in self.flann.knnMatch(desc.astype(np.float32), k=k): | |
| local = [n for n in nb | |
| if abs(self.ref_pts[n.trainIdx][0] - cx) <= radius | |
| and abs(self.ref_pts[n.trainIdx][1] - cy) <= radius] | |
| if not local: | |
| continue | |
| if len(local) >= 2: | |
| if local[0].distance < 0.92 * local[1].distance: | |
| out.append(local[0]) | |
| elif len(nb) < 2 or local[0].distance < 0.92 * nb[1].distance: | |
| out.append(local[0]) | |
| return out | |
| # --- matching -------------------------------------------------------- | |
| def _solve(self, kp, good, shape, diag): | |
| if len(good) < self.args.min_inliers_relaxed: | |
| diag["category"] = "few_candidates" | |
| return None, f"only {len(good)} candidate matches" | |
| src = np.float32([kp[g.queryIdx].pt for g in good]).reshape(-1, 1, 2) | |
| dst = np.float32([self.ref_pts[g.trainIdx] for g in good]).reshape(-1, 1, 2) | |
| M, inl = cv2.estimateAffinePartial2D(src, dst, method=cv2.RANSAC, | |
| ransacReprojThreshold=4.0, | |
| maxIters=8000) | |
| if M is None or inl is None: | |
| diag["category"] = "no_transform" | |
| return None, "no consistent transform" | |
| n = int(inl.sum()) | |
| scale = float(np.hypot(M[0, 0], M[1, 0])) | |
| rot = float(np.degrees(np.arctan2(M[1, 0], M[0, 0]))) | |
| diag.update(inliers=n, scale=round(scale, 3), rot=round(rot, 2)) | |
| if abs(rot) > 5.0: | |
| diag["category"] = "rotated" | |
| return None, f"rotation {rot:.1f} deg - rejected" | |
| # Scale gates are RELATIVE to the learned minimap scale, never absolute. | |
| # That scale depends on your screen resolution and on ZOOM, so a value | |
| # hard-coded for one setup rejects everything on another: at ZOOM 4 it is | |
| # ~0.95 on a 5120px-wide screen but ~2.5 on a 1080p one, and halving ZOOM | |
| # halves it again. The first strict fix establishes it and it is then | |
| # remembered in state.json. | |
| exp = self.expected_scale() | |
| if exp: | |
| if scale > 2.5 * exp and n >= self.args.min_inliers: | |
| # Opening the full map (M) hides the HUD minimap and puts the | |
| # whole island under the capture region, matching confidently at | |
| # ~8x the minimap scale. Real, but its centre is the map view | |
| # rather than you - so hold the last fix instead of using it. | |
| diag["category"] = "map_open" | |
| return None, "full map screen open" | |
| if not (0.75 * exp < scale < 1.33 * exp): | |
| diag["category"] = "bad_scale" | |
| return None, f"scale {scale:.2f} vs expected {exp:.2f} - rejected" | |
| else: | |
| # Bootstrap, before any scale is known. Only sanity limits apply, so | |
| # the first fix needs a full-strength inlier count to be trusted. | |
| if scale > 3.0 and n >= self.args.min_inliers: | |
| diag["category"] = "map_open" | |
| return None, "full map screen open (assumed - scale not yet learned)" | |
| if not (0.15 < scale < 3.0): | |
| diag["category"] = "bad_scale" | |
| return None, f"scale {scale:.2f} - rejected" | |
| h, w = shape | |
| p = cv2.transform(np.array([[[w / 2, h / 2]]], np.float32), M)[0][0] | |
| px, py = float(p[0]), float(p[1]) | |
| # Geometry substitutes for inlier count: the minimap is fixed-zoom and | |
| # north-locked, so a match on the established scale with no rotation is | |
| # strong evidence even with few inliers - provided it is also continuous | |
| # with where we just were. | |
| relaxed = False | |
| if n < self.args.min_inliers: | |
| exp = self.expected_scale() | |
| geo_ok = (exp is not None and abs(rot) <= 1.0 | |
| and abs(scale - exp) / exp <= 0.04) | |
| if not (n >= self.args.min_inliers_relaxed and geo_ok | |
| and self.continuity_ok(self.world_from_px(px, py))): | |
| diag["category"] = "low_inliers" | |
| return None, f"only {n} inliers" | |
| relaxed = True | |
| diag.pop("category", None) | |
| return ({"px": (px, py), "inliers": n, "cand": len(good), "scale": scale, | |
| "rot": rot, "relaxed": relaxed, "gated": diag["gated"]}, None) | |
| def fix(self, bgr): | |
| gray = self.clahe.apply(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)) | |
| diag = {"kp": 0, "cand": 0, "inliers": 0, "scale": None, "rot": None, | |
| "gated": False} | |
| kp, desc = self.sift.detectAndCompute(gray, None) | |
| diag["kp"] = 0 if desc is None else len(kp) | |
| if desc is None or len(kp) < 8: | |
| diag["category"] = "no_features" | |
| return None, "too few features", diag | |
| raw = self.flann.knnMatch(desc.astype(np.float32), k=2) | |
| last_err = None | |
| for ratio in (self.args.ratio, self.args.ratio_loose): | |
| good = [a for a, b in (p for p in raw if len(p) == 2) | |
| if a.distance < ratio * b.distance] | |
| diag["cand"], diag["ratio"] = len(good), ratio | |
| res, err = self._solve(kp, good, gray.shape, diag) | |
| if res is not None: | |
| return res, None, diag | |
| last_err = err | |
| if diag.get("category") not in RESCUABLE: | |
| break # e.g. map_open - retrying is futile | |
| if diag.get("category") in RESCUABLE: | |
| rescue = self.gated_matches(desc) | |
| if len(rescue) >= self.args.min_inliers_relaxed: | |
| gd = dict(diag, gated=True, cand=len(rescue)) | |
| res2, _ = self._solve(kp, rescue, gray.shape, gd) | |
| if res2 is not None: | |
| diag.clear(), diag.update(gd) | |
| return res2, None, diag | |
| return None, last_err, diag | |
| # --- loot ------------------------------------------------------------ | |
| def nearest(self, wx, wy, top, layers=None): | |
| """Nearest spawns, but capped per layer. | |
| Layer sizes differ by two orders of magnitude - season 42 has 1125 cheat | |
| codes against 8 vaults - so a straight global sort returns nothing but | |
| the dense layer and the rare things you actually care about never | |
| appear. Take the closest few of each, then rank those. | |
| """ | |
| per_layer = max(1, self.args.per_layer) | |
| picked = [] | |
| for name, pts in self.loot.items(): | |
| if layers is not None and name not in layers: | |
| continue | |
| rows = [] | |
| for x, y in pts: | |
| dx, dy = x - wx, y - wy | |
| b = math.degrees(math.atan2(dx, -dy)) % 360 # +X east, +Y south | |
| rows.append({"layer": name, | |
| "dist_m": round(math.hypot(dx, dy) / 100.0, 1), | |
| "bearing": round(b, 1), | |
| "point": COMPASS[round(b / 22.5) % 16], | |
| "world": [x, y]}) | |
| rows.sort(key=lambda r: r["dist_m"]) | |
| picked += rows[:per_layer] | |
| picked.sort(key=lambda r: r["dist_m"]) | |
| return picked[:top] | |
| def radar(self, wx, wy, heading, rng, cap, layers=None): | |
| """Everything within `rng` metres, for the heading-up radar. | |
| Separate from nearest(): the table wants a handful of the most relevant | |
| things, the radar wants the whole local picture. | |
| """ | |
| out = [] | |
| limit = rng * 100.0 # metres -> world units | |
| for name, pts in self.loot.items(): | |
| if layers is not None and name not in layers: | |
| continue | |
| for x, y in pts: | |
| dx, dy = x - wx, y - wy | |
| if abs(dx) > limit or abs(dy) > limit: | |
| continue # cheap reject before hypot | |
| d = math.hypot(dx, dy) | |
| if d > limit: | |
| continue | |
| b = math.degrees(math.atan2(dx, -dy)) % 360 | |
| e = {"layer": name, "dist_m": round(d / 100.0, 1), "bearing": round(b, 1)} | |
| if heading is not None: | |
| e["rel"] = round(relative_bearing(b, heading), 1) | |
| out.append(e) | |
| out.sort(key=lambda r: r["dist_m"]) | |
| return out[:cap] | |
| # --- loop ------------------------------------------------------------ | |
| def step(self): | |
| bgr = self.grab() | |
| # Read the compass regardless of the map fix: they are independent, and | |
| # a heading is still worth showing while the position is lost. | |
| heading = self.grab_heading() | |
| res, err, diag = self.fix(bgr) | |
| if res is None: | |
| self.save_miss(bgr, err, diag) | |
| st = {"ok": False, "msg": err, "heading": heading, | |
| "category": diag.get("category", "none")} | |
| # Keep showing the last position, but expire it: at the start of a | |
| # new match the previous fix belongs to the previous game. | |
| if (self.last_good and self._last_epoch is not None | |
| and time.time() - self._last_epoch <= self.args.hold_seconds): | |
| st["hold"] = self.last_good | |
| return st | |
| px, py = res["px"] | |
| wx, wy = self.world_from_px(px, py) | |
| return {"ok": True, "heading": heading, | |
| "inliers": res["inliers"], "cand": res["cand"], | |
| "scale": round(res["scale"], 3), "rot": round(res["rot"], 2), | |
| "relaxed": res["relaxed"], "gated": res["gated"], | |
| "px": [round(px, 1), round(py, 1)], "world": [round(wx), round(wy)], | |
| "t": time.strftime("%H:%M:%S")} | |
| def save_miss(self, bgr, err, diag): | |
| """Optional: keep unexplained failures for offline study (--save-misses). | |
| Skips lobby and loading screens, which have no minimap: real minimap | |
| crops yield well over a thousand keypoints, those yield a few hundred. | |
| """ | |
| a = self.args | |
| if not a.save_misses or diag.get("category") == "map_open": | |
| return | |
| if diag["kp"] < 800 or diag["cand"] < 2: | |
| return | |
| if time.time() - self._last_miss < 5.0: | |
| return | |
| self._last_miss = time.time() | |
| d = os.path.join(HERE, "misses") | |
| os.makedirs(d, exist_ok=True) | |
| stamp = time.strftime("%Y%m%d_%H%M%S") | |
| cv2.imwrite(os.path.join(d, f"miss_{stamp}.png"), bgr) | |
| with open(os.path.join(d, "log.jsonl"), "a", encoding="utf-8") as fh: | |
| fh.write(json.dumps({"file": f"miss_{stamp}.png", "reason": err, | |
| **{k: v for k, v in diag.items()}}) + "\n") | |
| def run(self): | |
| while True: | |
| t0 = time.time() | |
| try: | |
| st = self.step() | |
| except Exception as exc: # never let the server die | |
| st = {"ok": False, "msg": f"{type(exc).__name__}: {exc}"} | |
| st["epoch"] = time.time() | |
| st["clock"] = time.strftime("%H:%M:%S") | |
| if st.get("ok"): | |
| self.last_good = {k: st[k] for k in ("world", "px", "t")} | |
| self._last_epoch = st["epoch"] | |
| self.scale_hist.append(st["scale"]) | |
| if len(self.scale_hist) >= 5: | |
| val = float(np.median(self.scale_hist)) | |
| if not self._seed_scale or abs(val - self._seed_scale) / self._seed_scale >= 0.005: | |
| self._seed_scale = val | |
| save(STATE_JSON, {"expect_scale": round(val, 4), | |
| "zoom": self.meta["zoom"], | |
| "map": self.meta["map"]}) | |
| with self.lock: | |
| self.state = st | |
| time.sleep(max(0.0, self.args.interval - (time.time() - t0))) | |
| def view(self, layers=None): | |
| """State plus the lists for one layer selection. | |
| Built per request rather than in the capture loop, so the checkboxes can | |
| change what is shown without restarting anything - and so the per-layer | |
| caps apply to what you actually asked for. | |
| """ | |
| st = self.snapshot() | |
| src = st if st.get("ok") else st.get("hold") | |
| st["layers"] = [{"layer": k, "count": len(v), | |
| "color": "#%02x%02x%02x" % (self.colors[k][2], | |
| self.colors[k][1], | |
| self.colors[k][0])} | |
| for k, v in sorted(self.loot.items())] | |
| st["radar_range"] = self.args.radar_range | |
| if not src: | |
| return st | |
| wx, wy = src["world"] | |
| heading = st.get("heading") | |
| items = self.nearest(wx, wy, self.args.top, layers) | |
| if heading is not None: | |
| # A compass bearing is only half the answer while playing - what you | |
| # want is which way to turn from where you are already facing. | |
| for it in items: | |
| it["rel"] = round(relative_bearing(it["bearing"], heading), 1) | |
| radar = self.radar(wx, wy, heading, self.args.radar_range, | |
| self.args.radar_max, layers) | |
| if st.get("ok"): | |
| st["items"], st["radar"] = items, radar | |
| else: | |
| st["hold"] = dict(src, items=items) | |
| st["radar"] = radar | |
| return st | |
| def snapshot(self): | |
| with self.lock: | |
| return dict(self.state) | |
| # --- tile ------------------------------------------------------------ | |
| def tile_png(self, layers=None, half=300, out=620): | |
| st = self.view(layers) | |
| held = False | |
| if not st.get("ok"): | |
| if st.get("hold"): | |
| st, held = st["hold"], True | |
| else: | |
| blank = np.full((out, out, 3), 24, np.uint8) | |
| cv2.putText(blank, "no fix", (out // 2 - 70, out // 2), | |
| cv2.FONT_HERSHEY_SIMPLEX, 1.2, (90, 90, 90), 2) | |
| return cv2.imencode(".png", blank)[1].tobytes() | |
| px, py = st["px"] | |
| H, W = self.ref_bgr.shape[:2] | |
| x0 = max(0, min(W - 2 * half, int(px - half))) | |
| y0 = max(0, min(H - 2 * half, int(py - half))) | |
| crop = self.ref_bgr[y0:y0 + 2 * half, x0:x0 + 2 * half].copy() | |
| for it in st["items"]: | |
| ix, iy = self.px_from_world(*it["world"]) | |
| ix, iy = ix - x0, iy - y0 | |
| if 0 <= ix < crop.shape[1] and 0 <= iy < crop.shape[0]: | |
| c = self.colors.get(it["layer"], (255, 255, 255)) | |
| cv2.circle(crop, (int(ix), int(iy)), 9, (0, 0, 0), -1) | |
| cv2.circle(crop, (int(ix), int(iy)), 7, c, -1) | |
| cx, cy = int(px - x0), int(py - y0) | |
| cv2.circle(crop, (cx, cy), 13, (0, 0, 0), 3) | |
| cv2.circle(crop, (cx, cy), 11, (255, 255, 255), 3) | |
| cv2.line(crop, (cx, cy - 22), (cx, cy - 42), (255, 255, 255), 3) # north | |
| crop = cv2.resize(crop, (out, out), interpolation=cv2.INTER_LANCZOS4) | |
| if held: # desaturate so stale is obvious | |
| g = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) | |
| crop = cv2.addWeighted(crop, 0.35, | |
| cv2.cvtColor(g, cv2.COLOR_GRAY2BGR), 0.65, -18) | |
| return cv2.imencode(".png", crop)[1].tobytes() | |
| def legend(self): | |
| return [{"layer": k, "color": "#%02x%02x%02x" % (c[2], c[1], c[0])} | |
| for k, c in self.colors.items()] | |
| PAGE = """<!doctype html><html><head><meta charset="utf-8"> | |
| <meta name="viewport" content="width=device-width, initial-scale=1"> | |
| <title>Fortnite position</title><style> | |
| :root{color-scheme:dark} | |
| body{background:#12141a;color:#e8e8ef;font:16px/1.55 "Segoe UI",system-ui,sans-serif;margin:0;padding:20px} | |
| .wrap{display:flex;gap:22px;flex-wrap:wrap;align-items:flex-start;max-width:1200px;margin:0 auto} | |
| h1{font-size:19px;margin:0 0 12px;font-weight:600} | |
| img{border-radius:10px;display:block;background:#181b22;width:620px;max-width:100%} | |
| .panel{flex:1;min-width:330px} | |
| table{border-collapse:collapse;width:100%;font-variant-numeric:tabular-nums} | |
| th{text-align:left;font-weight:600;color:#8b90a0;font-size:12px;text-transform:uppercase; | |
| letter-spacing:.6px;padding:0 10px 7px 0;border-bottom:1px solid #262a35} | |
| td{padding:9px 10px 9px 0;border-bottom:1px solid #1c1f28;font-size:16px} | |
| .dot{display:inline-block;width:10px;height:10px;border-radius:50%;margin-right:9px;vertical-align:-1px} | |
| .deg{font-weight:700;font-size:17px} | |
| .meta{color:#8b90a0;font-size:13px;margin-top:14px} | |
| .clock{float:right;font-variant-numeric:tabular-nums;color:#8b90a0;font-size:15px} | |
| .clock b{color:#e8e8ef;font-weight:600} | |
| .stale{color:#ffcc66}.dead{color:#ff8f6b}.bad{color:#ff8f6b}.ok{color:#7ee081} | |
| .dim{color:#5f6472} | |
| .opts{max-width:1200px;margin:22px auto 0;padding:14px 18px;background:#171a22; | |
| border:1px solid #242835;border-radius:12px} | |
| .optshead{font-size:12px;text-transform:uppercase;letter-spacing:.6px;color:#8b90a0; | |
| font-weight:600;margin-bottom:10px;display:flex;align-items:center;gap:8px} | |
| .optshead button{background:#232734;color:#c9cede;border:1px solid #333947;border-radius:6px; | |
| padding:3px 9px;font:inherit;font-size:11px;text-transform:none;letter-spacing:0;cursor:pointer} | |
| .optgrid{display:grid;grid-template-columns:repeat(auto-fill,minmax(190px,1fr));gap:7px 14px} | |
| .optgrid label{display:flex;align-items:center;gap:8px;font-size:14px;color:#c9cede;cursor:pointer} | |
| .optgrid input{accent-color:#4a9eff;width:16px;height:16px;flex:none} | |
| .optgrid i{width:10px;height:10px;border-radius:50%;flex:none} | |
| .optgrid .n{color:#5f6472;font-size:12px;margin-left:auto;font-variant-numeric:tabular-nums} | |
| @media (max-width:760px){.optgrid{grid-template-columns:repeat(auto-fill,minmax(150px,1fr))} | |
| .optgrid label{font-size:16px}.optgrid input{width:20px;height:20px}} | |
| .hdrbar{max-width:1200px;margin:0 auto 18px;padding:14px 18px;background:#171a22; | |
| border:1px solid #242835;border-radius:12px;display:flex;align-items:baseline;gap:16px;flex-wrap:wrap} | |
| .hdgbox{display:flex;align-items:baseline;font-variant-numeric:tabular-nums;line-height:1} | |
| .hdgval{font-size:52px;font-weight:700;letter-spacing:-1px;color:#e8e8ef;min-width:2.2ch;text-align:right} | |
| .hdgdeg{font-size:30px;font-weight:600;color:#8b90a0;margin-left:2px} | |
| .hdgpt{font-size:26px;font-weight:700;color:#9fd0ff;margin-left:14px} | |
| .hdgnote{font-size:13px;color:#8b90a0} | |
| .hdgval.stale,.hdgdeg.stale{color:#5f6472} | |
| @media (max-width:760px){.hdgval{font-size:44px}.hdgpt{font-size:22px}} | |
| svg#radar{width:100%;max-width:340px;display:block;background:#0e1016;border-radius:12px} | |
| .sub{font-size:12px;font-weight:500;color:#8b90a0;letter-spacing:0} | |
| .legend{display:flex;flex-wrap:wrap;gap:10px 16px;margin:10px 0 0;font-size:12px;color:#9aa0b0} | |
| .legend span.k{display:inline-flex;align-items:center;gap:6px} | |
| .legend i{width:9px;height:9px;border-radius:50%;display:inline-block} | |
| .turn{font-weight:700;color:#9fd0ff;font-variant-numeric:tabular-nums} | |
| .ahead{font-weight:700;color:#7ee081} | |
| @media (max-width:760px){ | |
| body{font-size:18px;padding:14px} | |
| h1{font-size:21px} td{font-size:18px;padding:11px 8px 11px 0} .deg{font-size:19px} | |
| .clock{font-size:16px;float:none;display:block;margin-top:3px} | |
| img{width:100%} | |
| } | |
| </style></head><body> | |
| <div class="hdrbar"> | |
| <div class="hdgbox"><span class="hdgval" id="hdg">--</span><span class="hdgdeg">°</span> | |
| <span class="hdgpt" id="hdgpt"></span></div> | |
| <div class="hdgnote" id="hdgnote">compass reading</div> | |
| </div> | |
| <div class="wrap"> | |
| <div><h1>Live position <span class="clock" id="clock">—</span></h1><img id="tile" src="/tile.png"></div> | |
| <div class="panel"><h1>Radar <span class="sub" id="radarsub"></span></h1> | |
| <svg id="radar" viewBox="0 0 320 320" role="img" aria-label="heading-up radar"></svg> | |
| <div class="legend" id="legend"></div> | |
| <h1 style="margin-top:18px">Nearest items</h1> | |
| <table><thead><tr><th>Item</th><th>Dist</th><th>Bearing</th><th>Turn</th></tr></thead> | |
| <tbody id="rows"><tr><td colspan="4">waiting…</td></tr></tbody></table> | |
| <div class="meta" id="meta"></div></div></div> | |
| <div class="opts"> | |
| <div class="optshead">Show on radar | |
| <button id="allon">all</button><button id="alloff">none</button> | |
| <span class="dim" id="optcount"></span></div> | |
| <div class="optgrid" id="optgrid"></div> | |
| </div> | |
| <script> | |
| let lastEpoch=null,lastSeen=null,lastFix=null,reachable=false,COL={}; | |
| // Relative turn: negative is left, positive is right. An arrow reads faster | |
| // than a signed number when you are glancing at this mid-game. | |
| // North-up radar. The world stays put and a white needle shows which way you | |
| // are facing - a rotating map is harder to read than a moving needle. Radius | |
| // uses a square-root scale, which spreads the near cluster out where the | |
| // decisions actually get made. | |
| const R=140, CX=160, CY=160; | |
| function blipXY(relDeg, dist, range){ | |
| const t=relDeg*Math.PI/180, f=Math.sqrt(Math.min(dist,range)/range); | |
| return [CX+R*f*Math.sin(t), CY-R*f*Math.cos(t)]; | |
| } | |
| function drawRadar(s){ | |
| const el=document.getElementById('radar'), sub=document.getElementById('radarsub'); | |
| const range=s.radar_range||250, blips=(s.ok&&s.radar)?s.radar:[]; | |
| const headUp=(s.heading!==undefined&&s.heading!==null); | |
| sub.textContent = 'north up · '+range+' m'+(headUp?'':' · no heading'); | |
| let g=''; | |
| // range rings, labelled at real distances | |
| [range*0.04, range*0.16, range*0.36, range].forEach(d=>{ | |
| const r=R*Math.sqrt(d/range); | |
| g+='<circle cx="'+CX+'" cy="'+CY+'" r="'+r.toFixed(1)+'" fill="none" stroke="#242835"/>'; | |
| g+='<text x="'+(CX+3)+'" y="'+(CY-r+11)+'" fill="#4d5361" font-size="9">'+Math.round(d)+'m</text>'; | |
| }); | |
| g+='<line x1="'+CX+'" y1="'+(CY-R)+'" x2="'+CX+'" y2="'+(CY+R)+'" stroke="#1e2230"/>' | |
| + '<line x1="'+(CX-R)+'" y1="'+CY+'" x2="'+(CX+R)+'" y2="'+CY+'" stroke="#1e2230"/>'; | |
| // cardinals are fixed now that the dial no longer turns | |
| [['N',0],['E',90],['S',180],['W',270]].forEach(function(c){ | |
| const t=c[1]*Math.PI/180, x=CX+(R-7)*Math.sin(t), y=CY-(R-7)*Math.cos(t); | |
| g+='<text x="'+x.toFixed(1)+'" y="'+(y+4).toFixed(1)+'" fill="'+(c[0]==='N'?'#aab3c6':'#6b7285') | |
| + '" font-size="11" font-weight="700" text-anchor="middle">'+c[0]+'</text>'; | |
| }); | |
| // blips, far ones first so near ones land on top | |
| blips.slice().sort((a,b)=>b.dist_m-a.dist_m).forEach(b=>{ | |
| const [x,y]=blipXY(b.bearing,b.dist_m,range), c=COL[b.layer]||'#fff'; | |
| const near=b.dist_m<range*0.25; | |
| g+='<line x1="'+CX+'" y1="'+CY+'" x2="'+x.toFixed(1)+'" y2="'+y.toFixed(1)+'" stroke="'+c | |
| + '" stroke-width="'+(near?1.4:0.8)+'" opacity="'+(near?0.5:0.22)+'"/>'; | |
| g+='<circle cx="'+x.toFixed(1)+'" cy="'+y.toFixed(1)+'" r="'+(near?4.6:3.4) | |
| + '" fill="'+c+'" stroke="#0e1016" stroke-width="1.2"><title>'+pretty(b.layer)+' ' | |
| + b.dist_m.toFixed(0)+'m</title></circle>'; | |
| }); | |
| // the needle: where you are facing, drawn over everything so it stays findable | |
| if(headUp){ | |
| const t=s.heading*Math.PI/180; | |
| const hx=CX+(R-2)*Math.sin(t), hy=CY-(R-2)*Math.cos(t); | |
| g+='<line x1="'+CX+'" y1="'+CY+'" x2="'+hx.toFixed(1)+'" y2="'+hy.toFixed(1) | |
| + '" stroke="#ffffff" stroke-width="2" opacity="0.95"/>'; | |
| // arrowhead at the rim | |
| const ax=CX+(R-2)*Math.sin(t), ay=CY-(R-2)*Math.cos(t); | |
| const l=t+2.62, r2=t-2.62, s2=11; | |
| g+='<polygon points="'+ax.toFixed(1)+','+ay.toFixed(1)+' ' | |
| + (ax+s2*Math.sin(l)).toFixed(1)+','+(ay-s2*Math.cos(l)).toFixed(1)+' ' | |
| + (ax+s2*Math.sin(r2)).toFixed(1)+','+(ay-s2*Math.cos(r2)).toFixed(1) | |
| + '" fill="#ffffff"/>'; | |
| } | |
| g+='<circle cx="'+CX+'" cy="'+CY+'" r="4.5" fill="#ffffff" stroke="#0e1016" stroke-width="1.5"/>'; | |
| el.innerHTML=g; | |
| const seen=[...new Set(blips.map(b=>b.layer))]; | |
| document.getElementById('legend').innerHTML = seen.length | |
| ? seen.map(k=>'<span class="k"><i style="background:'+(COL[k]||'#fff')+'"></i>'+pretty(k)+'</span>').join('') | |
| : '<span class="dim">no spawns in range</span>'; | |
| } | |
| // Layer picker. The list is whatever the spawn data actually contains, so it | |
| // follows the season without any hardcoded names. The choice lives in | |
| // localStorage and rides along on every request as ?layers=, which keeps the | |
| // per-layer caps applied to what you asked for rather than filtering after. | |
| let SEL=null, LAYERS=[]; | |
| function loadSel(){ | |
| try{ const v=localStorage.getItem('fnp.layers'); if(v) SEL=new Set(JSON.parse(v)); } | |
| catch(e){} | |
| } | |
| function saveSel(){ | |
| try{ localStorage.setItem('fnp.layers', JSON.stringify([...SEL])); }catch(e){} | |
| } | |
| function qs(){ return SEL===null ? '' : '?layers='+encodeURIComponent([...SEL].join(',')); } | |
| function buildOptions(layers){ | |
| const sig=layers.map(l=>l.layer).join(','); | |
| if(sig===buildOptions._sig) { syncCount(); return; } | |
| buildOptions._sig=sig; LAYERS=layers; | |
| if(SEL===null){ | |
| // First visit: start with the sparse, high-value layers rather than all | |
| // thirty at once, which would be unreadable. | |
| SEL=new Set(layers.filter(l=>l.count<=60).map(l=>l.layer)); | |
| if(!SEL.size) SEL=new Set(layers.map(l=>l.layer)); | |
| saveSel(); | |
| } | |
| const g=document.getElementById('optgrid'); | |
| g.innerHTML=layers.map(l=> | |
| '<label><input type="checkbox" data-k="'+l.layer+'"'+(SEL.has(l.layer)?' checked':'')+'>'+ | |
| '<i style="background:'+l.color+'"></i>'+pretty(l.layer)+ | |
| '<span class="n">'+l.count+'</span></label>').join(''); | |
| g.querySelectorAll('input').forEach(cb=>cb.onchange=()=>{ | |
| cb.checked ? SEL.add(cb.dataset.k) : SEL.delete(cb.dataset.k); | |
| saveSel(); syncCount(); tick(); | |
| }); | |
| syncCount(); | |
| } | |
| function syncCount(){ | |
| const el=document.getElementById('optcount'); | |
| if(el&&SEL) el.textContent=SEL.size+' of '+LAYERS.length+' shown'; | |
| } | |
| document.getElementById('allon').onclick=()=>{ | |
| SEL=new Set(LAYERS.map(l=>l.layer)); saveSel(); buildOptions._sig=null; | |
| buildOptions(LAYERS); tick(); | |
| }; | |
| document.getElementById('alloff').onclick=()=>{ | |
| SEL=new Set(); saveSel(); buildOptions._sig=null; buildOptions(LAYERS); tick(); | |
| }; | |
| loadSel(); | |
| const PTS=["N","NNE","NE","ENE","E","ESE","SE","SSE","S","SSW","SW","WSW","W","WNW","NW","NNW"]; | |
| let lastHdg=null, lastHdgAt=null; | |
| // Shown big at the top so it can be checked against the number the game itself | |
| // prints on the compass - if the two ever disagree, the reader is at fault. | |
| function paintHeading(s){ | |
| const v=document.getElementById('hdg'), pt=document.getElementById('hdgpt'), | |
| note=document.getElementById('hdgnote'), degEl=document.querySelector('.hdgdeg'); | |
| const live=(s && s.heading!==undefined && s.heading!==null); | |
| if(live){ lastHdg=s.heading; lastHdgAt=Date.now(); } | |
| if(live){ | |
| v.textContent=s.heading; v.classList.remove('stale'); degEl.classList.remove('stale'); | |
| pt.textContent=PTS[Math.round(s.heading/22.5)%16]; | |
| note.textContent='compass reading'; | |
| } else if(lastHdg!==null){ | |
| const age=Math.round((Date.now()-lastHdgAt)/1000); | |
| v.textContent=lastHdg; v.classList.add('stale'); degEl.classList.add('stale'); | |
| pt.textContent=PTS[Math.round(lastHdg/22.5)%16]; | |
| note.textContent='last read '+age+'s ago (compass not readable right now)'; | |
| } else { | |
| v.textContent='--'; v.classList.add('stale'); degEl.classList.add('stale'); | |
| pt.textContent=''; note.textContent='compass not readable'; | |
| } | |
| } | |
| function turnCell(i){ | |
| if(i.rel===undefined||i.rel===null) return '<span class="dim">-</span>'; | |
| const a=Math.abs(i.rel); | |
| if(a<8) return '<span class="ahead">▲ ahead</span>'; | |
| return '<span class="turn">'+(i.rel<0?'←':'→')+' '+a.toFixed(0)+'°</span>'; | |
| } | |
| const NOTE={map_open:'full map open — holding last fix'}; | |
| const pretty=s=>s.replace(/_/g,' ').replace(/\\b\\w/g,c=>c.toUpperCase()); | |
| function paintClock(){ | |
| const el=document.getElementById('clock'); | |
| if(lastSeen===null){el.textContent='—';return;} | |
| const age=(Date.now()-lastSeen)/1000; | |
| el.className='clock '+(age>10?'dead':(age>3?'stale':'')); | |
| el.innerHTML=(reachable?'':'offline · ')+'<b>'+(lastFix||'no fix')+'</b> · '+age.toFixed(0)+'s ago'; | |
| } | |
| setInterval(paintClock,250); | |
| async function tick(){ | |
| let s; | |
| try{ s=await (await fetch('/state.json'+qs())).json(); reachable=true; } | |
| catch(e){ reachable=false; paintClock(); paintHeading(null); return; } | |
| if(!Object.keys(COL).length && s.legend) s.legend.forEach(l=>COL[l.layer]=l.color); | |
| if(s.epoch!==lastEpoch){lastEpoch=s.epoch;lastSeen=Date.now();} | |
| if(s.ok&&s.t) lastFix=s.t; | |
| paintClock(); | |
| const rows=document.getElementById('rows'),meta=document.getElementById('meta'); | |
| if(!s.ok&&!s.hold){ | |
| rows.innerHTML='<tr><td colspan="4" class="bad">'+(s.msg||'no fix')+'</td></tr>'; | |
| meta.textContent=''; | |
| }else{ | |
| const held=!s.ok,d=held?s.hold:s; | |
| rows.innerHTML=(held?'<tr><td colspan="4" class="stale">'+(NOTE[s.category]||s.msg||'holding last fix')+'</td></tr>':'')+ | |
| d.items.map(i=>'<tr'+(held?' style="opacity:.55"':'')+'>'+ | |
| '<td><span class="dot" style="background:'+(COL[i.layer]||'#fff')+'"></span>'+pretty(i.layer)+'</td>'+ | |
| '<td>'+i.dist_m.toFixed(0)+' m</td>'+ | |
| '<td><span class="deg">'+i.bearing.toFixed(0)+'°</span> '+i.point+'</td>'+ | |
| '<td>'+turnCell(i)+'</td></tr>').join(''); | |
| const hd=(s.heading!==undefined&&s.heading!==null) | |
| ? ' · facing <b>'+s.heading+'°</b>' | |
| : ' · <span class="dim">facing ?</span>'; | |
| meta.innerHTML='world <b>'+d.world[0]+', '+d.world[1]+'</b>'+hd+' · '+(held | |
| ?'<span class="stale">held since '+d.t+'</span>' | |
| :'<span class="'+(s.inliers>=10?'ok':'bad')+'">'+s.inliers+' inliers</span>/'+s.cand+ | |
| ' · scale '+s.scale+' · rot '+s.rot+'°'+(s.gated?' · gated':'')+' · '+s.t); | |
| } | |
| paintHeading(s); | |
| drawRadar(s); | |
| if(s.layers) buildOptions(s.layers); | |
| document.getElementById('tile').src='/tile.png'+(qs()?qs()+'&':'?')+'t='+Date.now(); | |
| } | |
| tick(); setInterval(tick,1000); | |
| </script></body></html>""" | |
| def tailscale_ips(): | |
| """Tailscale's own CLI is authoritative; fall back to reading the interface | |
| in case the CLI is unavailable but the address is still assigned.""" | |
| import shutil, subprocess | |
| exe = shutil.which("tailscale") or TAILSCALE_EXE | |
| ips, why = [], "" | |
| for flag in ("-4", "-6"): | |
| try: | |
| out = subprocess.run([exe, "ip", flag], capture_output=True, | |
| text=True, timeout=10) | |
| ips += [l.strip() for l in out.stdout.split() if l.strip()] | |
| if out.returncode and out.stderr.strip(): | |
| why = out.stderr.strip().splitlines()[0] | |
| except Exception as exc: | |
| why = f"{type(exc).__name__}: {exc}" | |
| if not ips: | |
| for fam in (socket.AF_INET, socket.AF_INET6): | |
| try: | |
| for info in socket.getaddrinfo(socket.gethostname(), None, fam): | |
| a = info[4][0] | |
| if a.startswith("100.") or a.lower().startswith("fd7a:"): | |
| ips.append(a) | |
| except Exception: | |
| pass | |
| return list(dict.fromkeys(ips)), why | |
| def resolve_binds(spec): | |
| if spec == "local": | |
| return ["127.0.0.1"] | |
| if spec == "all": | |
| return ["0.0.0.0"] | |
| if spec == "tailscale": | |
| ips, why = tailscale_ips() | |
| if not ips: | |
| # Loud: otherwise the only symptom is "my phone cannot connect", | |
| # with no hint that Tailscale is logged out rather than the tool | |
| # being broken. | |
| print("=" * 68) | |
| print(" NO TAILSCALE ADDRESS - serving on localhost only.") | |
| if why: | |
| print(f" tailscale said: {why}") | |
| print(" Other devices will NOT be able to reach this.") | |
| print(" Fix with: tailscale up then restart this") | |
| print("=" * 68) | |
| return ["127.0.0.1"] | |
| return ips + ["127.0.0.1"] | |
| return [s.strip() for s in spec.split(",") if s.strip()] | |
| def want_layers(path): | |
| """Parse ?layers=a,b,c - absent means every layer.""" | |
| import urllib.parse | |
| q = urllib.parse.urlparse(path).query | |
| v = urllib.parse.parse_qs(q).get("layers", [None])[0] | |
| if v is None: | |
| return None | |
| picked = {s for s in v.split(",") if s} | |
| return picked # an empty selection legitimately shows nothing | |
| def serve(tracker, port, binds): | |
| class Handler(http.server.BaseHTTPRequestHandler): | |
| def log_message(self, *a): | |
| pass | |
| def _send(self, body, ctype): | |
| self.send_response(200) | |
| self.send_header("Content-Type", ctype) | |
| self.send_header("Content-Length", str(len(body))) | |
| self.send_header("Cache-Control", "no-store") | |
| self.end_headers() | |
| self.wfile.write(body) | |
| def do_GET(self): | |
| path = self.path.split("?")[0] | |
| try: | |
| if path == "/": | |
| self._send(PAGE.encode("utf-8"), "text/html; charset=utf-8") | |
| elif path == "/state.json": | |
| st = tracker.view(want_layers(self.path)) | |
| st["legend"] = tracker.legend() | |
| self._send(json.dumps(st).encode("utf-8"), "application/json") | |
| elif path == "/tile.png": | |
| self._send(tracker.tile_png(want_layers(self.path)), "image/png") | |
| else: | |
| self.send_error(404) | |
| except (BrokenPipeError, ConnectionAbortedError): | |
| pass | |
| class Server(socketserver.ThreadingTCPServer): | |
| # Not SO_REUSEADDR on Windows: there it lets a second process silently | |
| # shadow a port already in use, and you get whichever answers first. | |
| allow_reuse_address = (os.name != "nt") | |
| daemon_threads = True | |
| class Server6(Server): | |
| address_family = socket.AF_INET6 | |
| servers = [] | |
| for addr in binds: | |
| cls = Server6 if ":" in addr else Server | |
| try: | |
| servers.append(cls((addr, port), Handler)) | |
| shown = f"[{addr}]" if ":" in addr else addr | |
| print(f"serving http://{shown}:{port}") | |
| except OSError as exc: | |
| print(f" could not bind {addr}:{port} - {exc}") | |
| if not servers: | |
| raise SystemExit("no address could be bound") | |
| for srv in servers: | |
| threading.Thread(target=srv.serve_forever, daemon=True).start() | |
| print("Ctrl+C to stop") | |
| try: | |
| while True: | |
| time.sleep(3600) | |
| except KeyboardInterrupt: | |
| for srv in servers: | |
| srv.shutdown() | |
| def cmd_run(args): | |
| tracker = Tracker(args) | |
| if args.once: | |
| print(json.dumps(tracker.step(), indent=2)) | |
| return | |
| threading.Thread(target=tracker.run, daemon=True).start() | |
| serve(tracker, args.port, resolve_binds(args.bind)) | |
| def main(): | |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[1]) | |
| sub = ap.add_subparsers(dest="cmd") | |
| sub.add_parser("pick", help="drag a box around the minimap") | |
| f = sub.add_parser("fetch", help="download the map and spawn data") | |
| f.add_argument("--map", default=MAP_BUILD) | |
| f.add_argument("--zoom", type=int, default=ZOOM) | |
| ap.add_argument("--port", type=int, default=PORT) | |
| ap.add_argument("--bind", default=BIND) | |
| ap.add_argument("--interval", type=float, default=INTERVAL) | |
| ap.add_argument("--top", type=int, default=TOP_N) | |
| ap.add_argument("--radar-range", type=float, default=250.0, | |
| help="radar radius in metres") | |
| ap.add_argument("--radar-max", type=int, default=60, | |
| help="max blips drawn on the radar") | |
| ap.add_argument("--no-heading", action="store_true", | |
| help="skip reading the compass bearing") | |
| ap.add_argument("--per-layer", type=int, default=3, | |
| help="max entries any one layer may contribute") | |
| ap.add_argument("--once", action="store_true", help="one fix, print, exit") | |
| ap.add_argument("--save-misses", action="store_true", | |
| help="keep frames that failed to match, for debugging") | |
| ap.add_argument("--min-inliers", type=int, default=8) | |
| ap.add_argument("--min-inliers-relaxed", type=int, default=4) | |
| ap.add_argument("--ratio", type=float, default=0.78) | |
| ap.add_argument("--ratio-loose", type=float, default=0.92) | |
| ap.add_argument("--hold-seconds", type=float, default=120.0) | |
| args = ap.parse_args() | |
| {"pick": cmd_pick, "fetch": cmd_fetch}.get(args.cmd, cmd_run)(args) | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment