fix(routing): allow bidirectional travel on two-way corridors
- u_turn_costs: 60 (was infinite -> U-turns forbidden -> huge detours) - distance_influence 70 -> 20, de-prioritise service/living_street/track - restore safe ignored_highways (was "" -> footways imported, bad snapping) - profiles_ch -> profiles_lm (edge-based CH too heavy with turn_costs) - server port 8989 -> 8080 to match compose mapping, healthcheck and API - add normalize-oneway + check-bidirectional tooling
This commit is contained in:
Hamza-Ayed
Executable
+225
@@ -0,0 +1,225 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user