maps-saas/infrastructure/scripts/normalize_oneway.py

#!/usr/bin/env python3
# --------------------------------------------------------------------------
# normalize_oneway.py — تصحيح وسوم الاتجاه الواحد قبل بناء رسم GraphHopper
#
# لماذا هذا السكربت أصلاً:
#   GraphHopper يثبّت اتجاه كل حافة أثناء الاستيراد اعتماداً على وسوم OSM
#   (oneway, oneway:motorcar, junction=roundabout). لا يوجد أي إعداد في
#   config.yml — ولا أي custom_model — يستطيع فتح اتجاه مُغلق في البيانات.
#   فإن كان طريق رئيسي ثنائي الاتجاه في الواقع (مثل طريق السخنة) موسوماً
#   oneway=yes في OSM، فالحل الوحيد هو تعديل البيانات قبل البناء.
#
# وضعان:
#   audit  : لا يعدّل شيئاً. يُخرج CSV بكل الطرق أحادية الاتجاه المرشحة
#            (way_id, highway, name, الطول التقريبي) لمراجعتها بشرياً.
#   apply  : يكتب ملف PBF جديداً بعد إزالة/تصحيح وسوم الاتجاه، إما
#            لقائمة معتمدة (--allowlist) أو لأصناف طرق كاملة (--classes).
#
# تحذير مهم:
#   الطرق المزدوجة (dual carriageway) هي جسمان منفصلان كل منهما أحادي الاتجاه
#   بشكل صحيح تماماً. إزالة oneway عنها تُنتج مسارات غير قانونية (سير بعكس
#   السير). لذلك الافتراضي هو --allowlist. استخدام --classes عملية شاملة
#   وخطرة، وتتطلب --force، ويجب أن تسبقها مراجعة تقرير audit.
# --------------------------------------------------------------------------
import argparse
import csv
import math
import sys

try:
    import osmium
except ImportError:
    sys.exit(
        "pyosmium غير مثبت.\n"
        "  python3 -m venv .venv-osm && ./.venv-osm/bin/pip install osmium\n"
        "ثم شغّل السكربت عبر ./.venv-osm/bin/python"
    )

ONEWAY_TRUE = {"yes", "true", "1", "-1"}

# الوسوم التي تحدد الاتجاه لسيارة. جميعها يجب أن يُزال معاً، وإلا بقي القيد.
DIRECTIONAL_TAGS = ("oneway", "oneway:motorcar", "oneway:motor_vehicle", "oneway:vehicle")

# أصناف لا تُمسّ أبداً: المسارات المنفصلة والروابط والدوارات أحادية بطبيعتها.
NEVER_TOUCH_HIGHWAY = {"motorway", "motorway_link"}

DEFAULT_CLASSES = ["trunk", "primary", "secondary", "tertiary"]


def haversine_len(nodes):
    """طول تقريبي بالأمتار من عقد الطريق (يتطلب مواقع العقد)."""
    total = 0.0
    prev = None
    for n in nodes:
        if not n.location.valid():
            prev = None
            continue
        if prev is not None:
            lat1, lon1, lat2, lon2 = map(
                math.radians, (prev.lat, prev.lon, n.location.lat, n.location.lon)
            )
            a = (
                math.sin((lat2 - lat1) / 2) ** 2
                + math.cos(lat1) * math.cos(lat2) * math.sin((lon2 - lon1) / 2) ** 2
            )
            total += 2 * 6371000 * math.asin(math.sqrt(a))
        prev = n.location
    return total


def is_oneway(tags):
    for t in DIRECTIONAL_TAGS:
        if tags.get(t) in ONEWAY_TRUE:
            return True
    return False


def is_protected(tags):
    """أحادي الاتجاه لسبب هيكلي صحيح — لا يُلمس."""
    if tags.get("highway") in NEVER_TOUCH_HIGHWAY:
        return True
    if (tags.get("highway") or "").endswith("_link"):
        return True
    if tags.get("junction") in ("roundabout", "circular"):
        return True
    return False


class Auditor(osmium.SimpleHandler):
    def __init__(self, classes, writer):
        super().__init__()
        self.classes = set(classes)
        self.writer = writer
        self.count = 0
        self.meters = 0.0

    def way(self, w):
        tags = dict(w.tags)
        hw = tags.get("highway")
        if hw not in self.classes or not is_oneway(tags) or is_protected(tags):
            return
        length = haversine_len(w.nodes)
        self.count += 1
        self.meters += length
        self.writer.writerow(
            [
                w.id,
                hw,
                tags.get("name", ""),
                tags.get("name:ar", ""),
                round(length),
                tags.get("lanes", ""),
                tags.get("oneway", ""),
            ]
        )


class Rewriter(osmium.SimpleHandler):
    """ينسخ الملف كاملاً ويحذف وسوم الاتجاه عن الطرق المستهدفة فقط."""

    def __init__(self, writer, classes, allowlist, names):
        super().__init__()
        self.writer = writer
        self.classes = set(classes) if classes else set()
        self.allowlist = allowlist or set()
        self.names = {n.strip() for n in (names or []) if n.strip()}
        self.changed = 0

    def node(self, n):
        self.writer.add_node(n)

    def relation(self, r):
        self.writer.add_relation(r)

    def _targeted(self, w, tags):
        if is_protected(tags):
            return False
        if w.id in self.allowlist:
            return True
        if self.names and (
            tags.get("name") in self.names or tags.get("name:ar") in self.names
        ):
            return True
        return bool(self.classes) and tags.get("highway") in self.classes

    def way(self, w):
        tags = dict(w.tags)
        if is_oneway(tags) and self._targeted(w, tags):
            for t in DIRECTIONAL_TAGS:
                tags.pop(t, None)
            # أثر مقصود: علامة تسمح بتتبع ما عدّلناه لاحقاً في PostGIS/التقارير
            tags["oneway"] = "no"
            tags["gh:oneway_normalized"] = "yes"
            self.changed += 1
            self.writer.add_way(w.replace(tags=tags))
        else:
            self.writer.add_way(w)


def main():
    p = argparse.ArgumentParser(description="تصحيح وسوم الاتجاه الواحد في ملف PBF")
    p.add_argument("mode", choices=["audit", "apply"])
    p.add_argument("--input", required=True)
    p.add_argument("--output", help="ملف PBF الناتج (مطلوب في apply)")
    p.add_argument("--report", default="oneway-audit.csv")
    p.add_argument(
        "--classes",
        default=",".join(DEFAULT_CLASSES),
        help="أصناف highway المستهدفة (audit) أو المشمولة بالتعديل الشامل (apply + --force)",
    )
    p.add_argument("--allowlist", help="ملف نصي: way_id في كل سطر")
    p.add_argument("--names", help="أسماء طرق مفصولة بفاصلة، مثل: طريق السخنة")
    p.add_argument(
        "--force",
        action="store_true",
        help="مطلوب لتطبيق التعديل على أصناف كاملة بدل قائمة معتمدة",
    )
    args = p.parse_args()

    classes = [c.strip() for c in args.classes.split(",") if c.strip()]

    if args.mode == "audit":
        with open(args.report, "w", newline="", encoding="utf-8") as fh:
            wr = csv.writer(fh)
            wr.writerow(
                ["way_id", "highway", "name", "name_ar", "length_m", "lanes", "oneway"]
            )
            h = Auditor(classes, wr)
            h.apply_file(args.input, locations=True)
        print(f"✅ audit: {h.count} طريق أحادي الاتجاه، ~{h.meters/1000:.1f} كم")
        print(f"   التقرير: {args.report}")
        print("   راجعه ثم مرّر way_ids المعتمدة عبر --allowlist في وضع apply.")
        return

    if not args.output:
        sys.exit("apply يتطلب --output")

    allow = set()
    if args.allowlist:
        with open(args.allowlist, encoding="utf-8") as fh:
            for line in fh:
                line = line.split("#")[0].strip()
                if line.isdigit():
                    allow.add(int(line))

    names = args.names.split(",") if args.names else []
    blanket = not allow and not names
    if blanket and not args.force:
        sys.exit(
            "لا allowlist ولا names → هذا تعديل شامل على أصناف كاملة.\n"
            "يكسر الطرق المزدوجة (dual carriageway) الموسومة بشكل صحيح.\n"
            "أعد التشغيل مع --force إن كان هذا مقصوداً بعد مراجعة تقرير audit."
        )

    writer = osmium.SimpleWriter(args.output)
    try:
        h = Rewriter(writer, classes if blanket else [], allow, names)
        h.apply_file(args.input, locations=False)
    finally:
        writer.close()
    print(f"✅ apply: عُدّل {h.changed} طريق → {args.output}")
    print("   الخطوة التالية: احذف graph-cache وأعد بناء GraphHopper.")


if __name__ == "__main__":
    main()