Started this year with a weekend mapping project: locate and map public traffic camera feeds from city’s website. Day 2 and I’m halfway done, about 130 out of ca 260 have live feed links now. Roughly 5 of them weren’t previously mapped at all.

Purpose of this diary entry is to save the checker scipt in publicly accessible place for unlikely reuse. Script scrapes all possible camera IDs and cross references them against already mapped feeds via overpass.

EDIT: Project was mostly completed as of 2026-01-02, changeset/176761784 outlines up to 10 missing cameras yet to be mapped.

Overpass query to get already mapped ones:

[out:json][timeout:25];
area(id:3600079510)->.searchArea;
node["contact:webcam"~ristm](area.searchArea);
out geom;

"""
Download latest Tallinn traffic camera images from ristmikud.tallinn.ee.
Images are saved next to this script (or into a fixed subfolder if the directory
isn't clean) and overwritten on each run. 
Also each image gets a small text label in the corner - similar label is on website,
but there label is added at html, not server side.
The script walks cam001 upward and stops after 10 sequential 404 responses.
Optionally checks Overpass (OSM) and can skip cameras already mapped there.

To install dependencies:
# Remember to create and activate a virtual environment first
python -m pip install -U pip
python -m pip install requests beautifulsoup4 pillow
"""

from __future__ import annotations

from pathlib import Path
import re
import time
from io import BytesIO

import requests
from bs4 import BeautifulSoup
from PIL import Image, ImageDraw, ImageFont

BASE = "https://ristmikud.tallinn.ee"
PAGE_URL = f"{BASE}/index.php/cams"

OVERPASS_URL = "https://overpass-api.de/api/interpreter"
OVERPASS_QUERY = """[out:json][timeout:25];
area(id:3600079510)->.searchArea;
nwr["contact:webcam"~ristm](area.searchArea);
out geom;"""

MAX_404_IN_A_ROW = 10
SLEEP = 0.1
TIMEOUT = 15

# Images are saved into a subdirectory of the script directory
SUBDIR_NAME = "tallinna-veeb"
SCRIPT_DIR = Path(__file__).resolve().parent

# If True: skip cams already present in OSM contact:webcam from Overpass
SKIP_EXISTING_IN_OSM = True


def make_unique_subdir(base_dir: Path, name: str) -> Path:
    cand = base_dir / name
    if not cand.exists():
        cand.mkdir(parents=True, exist_ok=True)
        return cand

    i = 2
    while True:
        cand2 = base_dir / f"{name}-{i}"
        if not cand2.exists():
            cand2.mkdir(parents=True, exist_ok=True)
            return cand2
        i += 1


def choose_output_dir(base_dir: Path, subdir_name: str = SUBDIR_NAME) -> Path:
    """
    Use base_dir if it contains exactly one .py file and all other entries are .jpg files,
    with no subdirectories. Otherwise use base_dir/subdir_name (fixed; no -2 etc.).
    """
    entries = list(base_dir.iterdir())
    if not entries:
        return base_dir

    # Not "clean" if any subdirectories exist
    if any(p.is_dir() for p in entries):
        return base_dir / subdir_name

    files = [p for p in entries if p.is_file()]
    py_files = [p for p in files if p.suffix.lower() == ".py"]
    other = [p for p in files if p.suffix.lower() not in (".py", ".jpg")]

    clean = (len(py_files) == 1) and (len(other) == 0)
    return base_dir if clean else (base_dir / subdir_name)



def fetch_cam_labels(session: requests.Session) -> dict[str, str]:
    cam_to_label: dict[str, str] = {}
    try:
        html = session.get(PAGE_URL, timeout=TIMEOUT).text
        soup = BeautifulSoup(html, "html.parser")
        for img in soup.find_all("img", id=re.compile(r"^cam\d{3}$")):
            cam_id = img.get("id")  # e.g. cam107
            a = img.find_parent("a")
            if not a:
                continue
            label = " ".join(a.get_text(" ", strip=True).split()).replace("*", "").strip()
            if label:
                cam_to_label[cam_id] = label
    except requests.RequestException:
        pass
    return cam_to_label


def overpass_existing_cam_ids(session: requests.Session) -> set[str]:
    """
    Returns a set like {"cam050", "cam049", ...} from OSM elements' contact:webcam tag.
    """
    existing: set[str] = set()
    try:
        r = session.post(
            OVERPASS_URL,
            data=OVERPASS_QUERY.encode("utf-8"),
            # This weird construct is caused by osm website's software
            headers = {"Content-Type": "text/plain; charset" + "=" + "utf-8"},
            timeout=TIMEOUT,
        )
        r.raise_for_status()
        data = r.json()
    except Exception as err:
        print(err)
        return existing

    # Find cam### in contact:webcam URLs
    cam_re = re.compile(r"/(cam\d{3})\.jpg\b", re.IGNORECASE)
    for el in data.get("elements", []):
        tags = el.get("tags") or {}
        url = tags.get("contact:webcam")
        if not url:
            continue
        m = cam_re.search(url)
        if m:
            existing.add(m.group(1).lower())
    return existing


def pick_font(font_size):
    for fp in (
        r"C:\Windows\Fonts\segoeui.ttf",
        r"C:\Windows\Fonts\calibri.ttf",
        r"C:\Windows\Fonts\calibri.ttf",
        r"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
        r"DejaVuSans.ttf"
    ):
        try:
            return ImageFont.truetype(fp, font_size)
        except OSError:
            pass
    return ImageFont.load_default()


def annotate(im: Image.Image, text: str) -> Image.Image:
    """
    Minimal, font-agnostic annotation:
    - uses PIL default font
    - draws a black box bottom-left, white text on top
    """
    im = im.convert("RGB")
    W, H = im.size
    draw = ImageDraw.Draw(im)

    font_size = max(18, int(H * 0.035))  # scales: ~25px@720p, ~38px@1080p
    pad = max(8, int(H * 0.012))
    font = pick_font(font_size)

    x0, y0, x1, y1 = draw.textbbox((0, 0), text, font=font)
    tw, th = x1 - x0, y1 - y0

    x = pad
    y = H - th - pad
    draw.rectangle((x - pad, y - pad, x + tw + pad, y + th + pad), fill=(0, 0, 0))
    draw.text((x, y), text, font=font, fill=(255, 255, 255))
    return im


def main():
    out_dir = choose_output_dir(SCRIPT_DIR, SUBDIR_NAME)
    out_dir.mkdir(parents=True, exist_ok=True)

    s = requests.Session()
    s.headers.update({"User-Agent": "Mozilla/5.0"})

    labels = fetch_cam_labels(s)

    existing_in_osm = overpass_existing_cam_ids(s)
    if existing_in_osm:
        print(f"Overpass: found {len(existing_in_osm)} existing cams in OSM.")
    else:
        print("Overpass: found 0 cams (or request failed).")

    n = 1
    streak_404 = 0

    while streak_404 < MAX_404_IN_A_ROW:
        cam_id = f"cam{n:03d}"          # e.g. cam050
        cam_id_l = cam_id.lower()

        if SKIP_EXISTING_IN_OSM and cam_id_l in existing_in_osm:
            print("skip (already in OSM):", cam_id)
            n += 1
            continue

        url = f"{BASE}/last/{cam_id}.jpg"
        out = out_dir / f"{cam_id}.jpg"

        try:
            r = s.get(url, timeout=TIMEOUT)
        except requests.RequestException:
            n += 1
            time.sleep(SLEEP)
            continue

        if r.status_code == 404:
            streak_404 += 1
            n += 1
            time.sleep(SLEEP)
            continue

        streak_404 = 0  # reset on any non-404 response

        ctype = (r.headers.get("Content-Type") or "").lower()
        if r.status_code == 200 and ctype.startswith("image/"):
            try:
                im = Image.open(BytesIO(r.content))
                text = labels.get(cam_id, cam_id)
                im = annotate(im, text)
                im.save(out, format="JPEG", quality=92, optimize=True)
                print("saved", cam_id, "-", text, "->", out)
            except Exception:
                pass

        n += 1
        time.sleep(SLEEP)

    print("stopped after 10 sequential 404s; last tried:", n - 1)
    print("output directory:", out_dir)


if __name__ == "__main__":
    main()