OSM changeset counter

OSM Changeset counter.py

from time import sleep
import io
import gzip
import requests


SITE = "https://planet.openstreetmap.org"
STATE_PATH = "/replication/changesets/state.yaml"
CHANGES_PATH = "/replication/changesets/"
LAST_RUN_KEY = "last_run: "
SEQUENCE_KEY = "sequence: "


# https://stackoverflow.com/a/46591040/8946790
def wrap(s, w):
    return [s[i : i + w] for i in range(0, len(s), w)]


# This is some awful code:
def commaify(n: int, c: str = ",") -> str:
    return c.join(wrap(str(n)[::-1], 3))[::-1]


def value_of(text: str, key: str, end: str = "\n") -> str:
    a = text.find(key) + len(key)
    b = text[a:].find(end) + a
    return text[a:b]


def main() -> int:

    prev_biggest_id = None
    changeset_100_found = False

    running = True
    while running:
        res = requests.get(SITE + STATE_PATH)

        last_run = value_of(res.text, LAST_RUN_KEY)
        seq = int(value_of(res.text, SEQUENCE_KEY))

        seq_path = commaify(f"{seq:0>9}", "/")
        res = requests.get(SITE + CHANGES_PATH + seq_path + ".osm.gz")
        text = gzip.decompress(res.content).decode()

        # Bad code that finds biggest changeset id:
        biggest_id = 0
        for line in text.split("\n"):
            line = line.strip()
            if line.startswith("<changeset"):
                id = int(value_of(line, 'id="', '"'))
                if id > biggest_id:
                    biggest_id = id
                if id == 100_000_000:
                    changeset_100_found = True

        if (prev_biggest_id == None) or (biggest_id - prev_biggest_id) != 0:
            print(last_run)
            # print(f"{seq_path=!r}")
            print(f"Latest CS: {commaify(biggest_id)}")
            print(f"CSs from 100M: {commaify(100_000_000 - biggest_id)}")
            if prev_biggest_id:
                print(f"CSs since last: {commaify(biggest_id - prev_biggest_id)}")
                print(f"Minites to 100M: {(100_000_000 - biggest_id)/(biggest_id - prev_biggest_id)}")
            if changeset_100_found:
                print(
                    "\u001B[7m\u001B[6m\u001B]8;;"
                    + "https://www.osm.org/changeset/100000000"
                    + "\u001B\\"
                    + "CHANGESET 100M FOUND!!! CLICK HERE!!!"
                    + "\u001B]8;;\u001B\\\u001B[0m"
                )
        print()

        sleep(60)  # Wait a minuite

        prev_biggest_id = biggest_id

    return 0


if __name__ == "__main__":
    exit(main())