#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
topo_viewer.py
Version: 1.01
Reads a lat/lon from latest.txt, queries the USGS National Map API for
US Topo PDF maps covering that area, downloads any new ones, and prints
an HTML page displaying all available maps.
Called by topo_viewer.php via shell_exec().
Changes in 1.01:
- After fetching API results, group maps by geographic name and keep only
the most recent publicationDate per name. This eliminates the 2011,
2014, 2017, 2020 duplicates and shows one current map per quad sheet.
"""
# ===========================================================================
# USER CONFIGURATION
# ===========================================================================
# Full path to latest.txt (single line: "T17S R3E Section 21, 44.0776, -122.4553")
LATEST_TXT = "/var/www/html/weather/tools/trs2latlon/latest.txt"
# Directory where downloaded PDF maps are saved (must be writable by www-data)
MAPS_DIR = "/var/www/html/weather/tools/topo_maps/maps"
# Web-accessible URL path to the maps directory
MAPS_URL = "/weather/tools/topo_maps/maps"
# Search radius around the lat/lon in decimal degrees (0.1 ~ 7 miles)
BBOX_HALF = 0.0001
# USGS National Map API endpoint
API_URL = "https://tnmaccess.nationalmap.gov/api/v1/products"
# Dataset name to query
DATASET = "US Topo"
# ===========================================================================
# END USER CONFIGURATION
# ===========================================================================
import sys
import os
import re
import requests
# ---------------------------------------------------------------------------
# Helper: emit a self-contained error page and exit
# ---------------------------------------------------------------------------
def die(message):
print("Content-Type: text/html\n")
print(f"""
Topo Viewer - Error
Topo Viewer Error
{message}
""")
sys.exit(1)
# ---------------------------------------------------------------------------
# Helper: extract a normalized map name from a USGS title string
#
# Input examples:
# "US Topo 7.5-minute map for Blue River, OR"
# "USGS US Topo 7.5-minute map for Blue River, OR 2011"
# "USGS US Topo 7.5-minute map for Saddleblanket Mountain, OR 2020"
#
# Output: "Blue River, OR" / "Saddleblanket Mountain, OR"
# ---------------------------------------------------------------------------
def normalize_map_name(title):
# Strip leading "USGS " if present
t = re.sub(r'^USGS\s+', '', title, flags=re.IGNORECASE)
# Strip the standard prefix up to "for "
t = re.sub(r'^US Topo\s+[\d.]+-minute map for\s+', '', t, flags=re.IGNORECASE)
# Strip a trailing 4-digit year (e.g. " 2011", " 2020")
t = re.sub(r'\s+\d{4}\s*$', '', t)
return t.strip()
# ---------------------------------------------------------------------------
# Step 1: Read latest.txt and parse lat/lon
# ---------------------------------------------------------------------------
if not os.path.isfile(LATEST_TXT):
die(f"Cannot find latest.txt at: {LATEST_TXT}")
try:
with open(LATEST_TXT, "r", encoding="utf-8") as f:
line = f.readline().strip()
except Exception as e:
die(f"Error reading latest.txt: {e}")
if not line:
die("latest.txt appears to be empty.")
parts = [p.strip() for p in line.split(",")]
if len(parts) < 3:
die(f"Expected at least 3 comma-separated fields in latest.txt, got: {line}")
try:
location_label = parts[0]
lat = float(parts[1])
lon = float(parts[2])
except ValueError as e:
die(f"Could not parse lat/lon from latest.txt: {e}
Line was: {line}")
# ---------------------------------------------------------------------------
# Step 2: Query USGS National Map API
# ---------------------------------------------------------------------------
bbox = f"{lon - BBOX_HALF},{lat - BBOX_HALF},{lon + BBOX_HALF},{lat + BBOX_HALF}"
params = {
"datasets": DATASET,
"bbox": bbox,
"outputFormat": "JSON",
"max": 50,
}
try:
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
except requests.exceptions.RequestException as e:
die(f"USGS API request failed: {e}")
except ValueError:
die("USGS API returned invalid JSON.")
items = data.get("items", [])
# ---------------------------------------------------------------------------
# Step 3: Deduplicate -- keep only the most recent map per named quad sheet
#
# Group by normalized name, compare publicationDate strings (YYYY-MM-DD
# sorts correctly as plain strings), keep the winner.
# ---------------------------------------------------------------------------
best = {} # normalized_name -> item dict
for item in items:
title = item.get("title", "")
download_url = item.get("downloadURL", "")
pub_date = item.get("publicationDate", "")
if not download_url or not download_url.lower().endswith(".pdf"):
continue
name_key = normalize_map_name(title)
if name_key not in best:
best[name_key] = item
else:
# Compare publication dates; later date wins
if pub_date > best[name_key].get("publicationDate", ""):
best[name_key] = item
# Sort the winners alphabetically by map name for a tidy display
deduped_items = [best[k] for k in sorted(best.keys())]
# ---------------------------------------------------------------------------
# Step 4: Ensure maps directory exists
# ---------------------------------------------------------------------------
os.makedirs(MAPS_DIR, exist_ok=True)
# ---------------------------------------------------------------------------
# Step 5: Download any new PDFs
# ---------------------------------------------------------------------------
downloaded_maps = [] # list of dicts: {title, filename, is_new, pub_date}
errors = []
for item in deduped_items:
title = item.get("title", "Untitled")
download_url = item.get("downloadURL", "")
product_id = item.get("sourceId") or item.get("id") or ""
pub_date = item.get("publicationDate", "")
map_name = normalize_map_name(title)
# Build a safe filename from the product_id (or fall back to URL basename)
if product_id:
safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in str(product_id))
filename = f"{safe_id}.pdf"
else:
filename = os.path.basename(download_url.split("?")[0])
if not filename.lower().endswith(".pdf"):
filename += ".pdf"
filepath = os.path.join(MAPS_DIR, filename)
is_new = False
if not os.path.isfile(filepath):
try:
dl = requests.get(download_url, timeout=120, stream=True)
dl.raise_for_status()
with open(filepath, "wb") as pf:
for chunk in dl.iter_content(chunk_size=65536):
pf.write(chunk)
is_new = True
except Exception as e:
errors.append(f"Failed to download '{title}': {e}")
continue
downloaded_maps.append({
"title": map_name,
"pub_date": pub_date,
"filename": filename,
"url": f"{MAPS_URL}/{filename}",
"is_new": is_new,
})
# ---------------------------------------------------------------------------
# Step 6: Emit HTML page
# ---------------------------------------------------------------------------
map_count = len(downloaded_maps)
new_count = sum(1 for m in downloaded_maps if m["is_new"])
print("""
Topo Maps - """ + location_label + """
Quad sheets: {map_count}
Newly downloaded: {new_count}
Showing most recent edition per quad sheet
""")
if errors:
print('Download Errors
')
for err in errors:
print(f" - {err}
")
print("
No topo maps found
No US Topo PDFs were returned for this location and search radius.
Lat: {lat:.4f}, Lon: {lon:.4f}, bbox +/- {BBOX_HALF} deg
""")
else:
print('')
for m in downloaded_maps:
new_badge = '
New' if m["is_new"] else ""
pub = f'
Published: {m["pub_date"]}' if m["pub_date"] else ""
print(f"""
""")
print("