vespucci-data-analysis/script.py

#!/usr/bin/env python3

import sys
import json
import time
import concurrent.futures
import requests
from collections import defaultdict
from ipo import ipo, opi, p


OSM_USERNAME = ""

if not OSM_USERNAME:
	print("Fill in your OSM username in the script first.", file=sys.stderr)
	sys.exit(1)


session = requests.Session()


changes = defaultdict(dict)
for oneway_bicycle in ("yes", "no", "-1", "other"):
	for cycleway_both in ("yes", "no", "other"):
		changes \
			[f"removed oneway:bicycle={oneway_bicycle}"] \
			[f"added cycleway:both={cycleway_both}"] \
			= []
	changes \
		[f"removed oneway:bicycle={oneway_bicycle}"] \
		["didn't modify cycleway:both"] \
		= []

oneway_added_after_cycleway_both = []
nothing_to_report = []
preconditions_not_met = []


def normalized_oneway(value: str) -> str:
	if value in ("yes", "1", "true"):
		return "yes"
	elif value in ("no", "0", "false"):
		return "no"
	elif value in ("-1", "reverse"):
		return "-1"
	else:
		return "other"

def normalized_cycleway(value: str) -> str:
	if value in ("yes", "no"):
		return value
	else:
		return "other"


def get(url):
	r = session.get(
		url, timeout=30,
		headers={
			"User-Agent": "M!dgard's script to detect data corruption by a Vespucci bug, run by "
			"`{OSM_USERNAME}`",
			"X-Note": "See https://community.openstreetmap.org/t/98353. This script is used on "
			"output of Overpass to check the tag history to determine if a road was corrupted "
			f"by a Vespucci bug or UX issue. Run by OSM user `{OSM_USERNAME}`, contact them in "
			"case of problems."
		}
	)
	if r.status_code == 503:
		time.sleep(60)
		return get(url)
	r.raise_for_status()
	return r


def handle_way(i, total, way_id):
	print(f"\r{i}/{len(ways)}", flush=True, end="", file=sys.stderr)

	history_url = f"https://api.openstreetmap.org/api/0.6/way/{way_id}/history.json"
	r = get(history_url)

	history = (ipo(r.json()["elements"]) | reversed | list | opi)
	# history = [{"version": 6, …}, {"version": 5, …}, …]

	sort_in = None

	if (
		history[0].get("tags", {}).get("cycleway:both") != "no" or
		history[0].get("tags", {}).get("oneway") not in ("yes", "-1") or
		"oneway:bicycle" in history[0].get("tags", {})
	):
		print(
			f"w{way_id}: preconditions for tags not met: wrong query used, "
			"or OSM element modified after executing query",
			file=sys.stderr, flush=True
		)
		sort_in = preconditions_not_met
		sort_in.append(way_id)
		return

	for previous_version, version in zip(history[1:], history):
		assert sort_in is None

		cs = version["changeset"]
		v = version["version"]
		version_prefix = f"w{way_id}: v{v} (#{cs})"


		if not previous_version.get("tags", []):
			sort_in = nothing_to_report
			break


		assert "oneway:bicycle" not in version["tags"]
		oneway_bicycle_removed = "oneway:bicycle" in previous_version["tags"]
		cycleway_both_modified = (
			"cycleway:both" in version["tags"] and
			previous_version["tags"].get("cycleway:both") != version["tags"]["cycleway:both"]
		)


		# oneway changed somewhere after the purported StreetComplete change
		if previous_version["tags"].get("oneway") != version["tags"].get("oneway"):
			print(
				f"\n{version_prefix} +oneway={version['tags']['oneway']}",
				flush=True, file=sys.stderr
			)
			sort_in = oneway_added_after_cycleway_both
			break


		oneway_bicycle = previous_version["tags"].get("oneway:bicycle")
		cycleway_both = version["tags"].get("cycleway:both")

		if cycleway_both_modified:

			if oneway_bicycle_removed:

				print(
					f"\n{version_prefix} "
					f"+cycleway:both={cycleway_both} -oneway:bicycle={oneway_bicycle}",
					flush=True, file=sys.stderr
				)
				sort_in = changes \
					[f"removed oneway:bicycle={normalized_oneway(oneway_bicycle)}"] \
					[f"added cycleway:both={normalized_cycleway(cycleway_both)}"]
				break

			else:
				sort_in = nothing_to_report
				break

		if oneway_bicycle_removed:
			# oneway:bicycle=* was removed after cycleway:both was added
			print(
				f"\n{version_prefix} "
				f"-oneway:bicycle={oneway_bicycle}",
				flush=True, file=sys.stderr
			)
			sort_in = changes \
				[f"removed oneway:bicycle={normalized_oneway(oneway_bicycle)}"] \
				["didn't modify cycleway:both"]
			break

	if sort_in is None:
		sort_in = nothing_to_report

	sort_in.append(way_id)


print(
	"Feed JSON data from this Overpass wizard query on stdin:\n"
	"(cycleway:both=no and (oneway=yes or oneway=1 or oneway=true or oneway=-1 or oneway=reverse) "
	"and oneway:bicycle!=*) and type:way global\n",
	file=sys.stderr
)

ways = json.load(sys.stdin)["elements"]
for way in ways:
	assert way["type"] == "way"

try:
	with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
		_ = list(executor.map(
			lambda i_way: handle_way(i_way[0], len(ways), i_way[1]),
			enumerate(map(lambda x: x["id"], ways))
		))

finally:
	print(json.dumps(
		{
			"changes": changes,
			"oneway_added_after_cycleway_both": oneway_added_after_cycleway_both,
			"nothing_to_report": nothing_to_report,
			"preconditions_not_met": preconditions_not_met
		},
		indent="\t"
	))