organicmaps/subways
13
Fork 0
Code Issues Pull requests1 Projects Releases Wiki Activity Actions

process train_station_entrance similar to subway_entrance

Browse source
This commit is contained in:
Alexey Zakharenkov 2023-11-29 12:35:37 +03:00 • committed by Alexey Zakharenkov
parent 6c796ac8c1
commit 970b4a51ee
20 changed files with 474 additions and 111 deletions

31
process_subways.py
View file

@ -24,7 +24,7 @@ from subway_structure import (
City,
CriticalValidationError,
find_transfers,
get_unused_entrances_geojson,
get_unused_subway_entrances_geojson,
MODES_OVERGROUND,
MODES_RAPID,
)
@ -38,26 +38,37 @@ DEFAULT_CITIES_INFO_URL = (
Point = tuple[float, float]
def overpass_request(
overground: bool, overpass_api: str, bboxes: list[list[float]]
) -> list[dict]:
def compose_overpass_request(
overground: bool, bboxes: list[list[float]]
) -> str:
if not bboxes:
raise RuntimeError("No bboxes given for overpass request")
query = "[out:json][timeout:1000];("
modes = MODES_OVERGROUND if overground else MODES_RAPID
for bbox in bboxes:
bbox_part = "({})".format(",".join(str(coord) for coord in bbox))
bbox_part = f"({','.join(str(coord) for coord in bbox)})"
query += "("
for mode in modes:
query += 'rel[route="{}"]{};'.format(mode, bbox_part)
for mode in sorted(modes):
query += f'rel[route="{mode}"]{bbox_part};'
query += ");"
query += "rel(br)[type=route_master];"
if not overground:
query += "node[railway=subway_entrance]{};".format(bbox_part)
query += "rel[public_transport=stop_area]{};".format(bbox_part)
query += f"node[railway=subway_entrance]{bbox_part};"
query += f"node[railway=train_station_entrance]{bbox_part};"
query += f"rel[public_transport=stop_area]{bbox_part};"
query += (
"rel(br)[type=public_transport][public_transport=stop_area_group];"
)
query += ");(._;>>;);out body center qt;"
logging.debug("Query: %s", query)
return query
def overpass_request(
overground: bool, overpass_api: str, bboxes: list[list[float]]
) -> list[dict]:
query = compose_overpass_request(overground, bboxes)
url = "{}?data={}".format(overpass_api, urllib.parse.quote(query))
response = urllib.request.urlopen(url, timeout=1000)
if (r_code := response.getcode()) != 200:
@ -489,7 +500,7 @@ def main() -> None:
write_recovery_data(options.recovery_path, recovery_data, cities)
if options.entrances:
json.dump(get_unused_entrances_geojson(osm), options.entrances)
json.dump(get_unused_subway_entrances_geojson(osm), options.entrances)
if options.dump:
if os.path.isdir(options.dump):

6
scripts/download_all_subways.sh
View file

@ -1,6 +0,0 @@
#!/bin/bash
# Still times out, do not use unless you want to be blocked for some hours on Overpass API
TIMEOUT=2000
QUERY='[out:json][timeout:'$TIMEOUT'];(rel["route"="subway"];rel["route"="light_rail"];rel["public_transport"="stop_area"];rel["public_transport"="stop_area_group"];node["station"="subway"];node["station"="light_rail"];node["railway"="subway_entrance"];);(._;>;);out body center qt;'
http http://overpass-api.de/api/interpreter "data==$QUERY" --timeout $TIMEOUT > subways-$(date +%y%m%d).json
http https://overpass-api.de/api/status | grep available

6
scripts/filter_all_subways.sh
View file

@ -1,6 +0,0 @@
#!/bin/bash
[ $# -lt 1 ] && echo 'Usage: $0 <planet.o5m> [<path_to_osmfilter>] [<output_file>]' && exit 1
OSMFILTER=${2-./osmfilter}
QRELATIONS="route=subway =light_rail =monorail route_master=subway =light_rail =monorail public_transport=stop_area =stop_area_group"
QNODES="station=subway =light_rail =monorail railway=subway_entrance subway=yes light_rail=yes monorail=yes"
"$OSMFILTER" "$1" --keep= --keep-relations="$QRELATIONS" --keep-nodes="$QNODES" --drop-author -o="${3:-subways-$(date +%y%m%d).osm}"

2
scripts/process_subways.sh
View file

@ -217,7 +217,7 @@ if [ -n "${NEED_FILTER-}" ]; then
check_osmctools
mkdir -p $TMPDIR/osmfilter_temp/
QRELATIONS="route=subway =light_rail =monorail =train route_master=subway =light_rail =monorail =train public_transport=stop_area =stop_area_group"
QNODES="railway=station station=subway =light_rail =monorail railway=subway_entrance subway=yes light_rail=yes monorail=yes train=yes"
QNODES="railway=station =subway_entrance =train_station_entrance station=subway =light_rail =monorail subway=yes light_rail=yes monorail=yes train=yes"
"$OSMCTOOLS/osmfilter" "$PLANET_METRO" \
--keep= \
--keep-relations="$QRELATIONS" \

157
subway_structure.py
View file

@ -1,11 +1,12 @@
from __future__ import annotations
import math
import re
from collections import Counter, defaultdict
from itertools import islice
from itertools import chain, islice
from css_colours import normalize_colour
MAX_DISTANCE_TO_ENTRANCES = 300 # in meters
MAX_DISTANCE_STOP_TO_LINE = 50 # in meters
ALLOWED_STATIONS_MISMATCH = 0.02 # part of total station count
@ -283,13 +284,11 @@ def format_elid_list(ids):
class Station:
@staticmethod
def get_modes(el):
mode = el["tags"].get("station")
modes = [] if not mode else [mode]
for m in ALL_MODES:
if el["tags"].get(m) == "yes":
modes.append(m)
return set(modes)
def get_modes(el: dict) -> set[str]:
modes = {m for m in ALL_MODES if el["tags"].get(m) == "yes"}
if mode := el["tags"].get("station"):
modes.add(mode)
return modes
@staticmethod
def is_station(el, modes):
@ -367,7 +366,9 @@ class StopArea:
return False
return el["tags"].get("railway") in RAILWAY_TYPES
def __init__(self, station, city, stop_area=None):
def __init__(
self, station: Station, city: City, stop_area: StopArea | None = None
) -> None:
"""Call this with a Station object."""
self.element = stop_area or station.element
@ -375,9 +376,10 @@ class StopArea:
self.station = station
self.stops = set() # set of el_ids of stop_positions
self.platforms = set() # set of el_ids of platforms
self.exits = set() # el_id of subway_entrance for leaving the platform
self.entrances = set() # el_id of subway_entrance for entering
# the platform
self.exits = set() # el_id of subway_entrance/train_station_entrance
# for leaving the platform
self.entrances = set() # el_id of subway/train_station entrance
# for entering the platform
self.center = None # lon, lat of the station centre point
self.centers = {} # el_id -> (lon, lat) for all elements
self.transfer = None # el_id of a transfer relation
@ -400,62 +402,9 @@ class StopArea:
except ValueError as e:
city.warn(str(e), stop_area)
# If we have a stop area, add all elements from it
warned_about_tracks = False
for m in stop_area["members"]:
k = el_id(m)
m_el = city.elements.get(k)
if m_el and "tags" in m_el:
if Station.is_station(m_el, city.modes):
if k != station.id:
city.error(
"Stop area has multiple stations", stop_area
)
elif StopArea.is_stop(m_el):
self.stops.add(k)
elif StopArea.is_platform(m_el):
self.platforms.add(k)
elif m_el["tags"].get("railway") == "subway_entrance":
if m_el["type"] != "node":
city.warn("Subway entrance is not a node", m_el)
if (
m_el["tags"].get("entrance") != "exit"
and m["role"] != "exit_only"
):
self.entrances.add(k)
if (
m_el["tags"].get("entrance") != "entrance"
and m["role"] != "entry_only"
):
self.exits.add(k)
elif StopArea.is_track(m_el):
if not warned_about_tracks:
city.warn(
"Tracks in a stop_area relation", stop_area
)
warned_about_tracks = True
self._process_members(station, city, stop_area)
else:
# Otherwise add nearby entrances
center = station.center
for c_el in city.elements.values():
if c_el.get("tags", {}).get("railway") == "subway_entrance":
c_id = el_id(c_el)
if c_id not in city.stop_areas:
c_center = el_center(c_el)
if (
c_center
and distance(center, c_center)
<= MAX_DISTANCE_TO_ENTRANCES
):
if c_el["type"] != "node":
city.warn(
"Subway entrance is not a node", c_el
)
etag = c_el["tags"].get("entrance")
if etag != "exit":
self.entrances.add(c_id)
if etag != "entrance":
self.exits.add(c_id)
self._add_nearby_entrances(station, city)
if self.exits and not self.entrances:
city.warn(
@ -476,13 +425,77 @@ class StopArea:
self.center = station.center
else:
self.center = [0, 0]
for sp in self.stops | self.platforms:
for sp in chain(self.stops, self.platforms):
spc = self.centers[sp]
for i in range(2):
self.center[i] += spc[i]
for i in range(2):
self.center[i] /= len(self.stops) + len(self.platforms)
def _process_members(
self, station: Station, city: City, stop_area: dict
) -> None:
# If we have a stop area, add all elements from it
tracks_detected = False
for m in stop_area["members"]:
k = el_id(m)
m_el = city.elements.get(k)
if not m_el or "tags" not in m_el:
continue
if Station.is_station(m_el, city.modes):
if k != station.id:
city.error("Stop area has multiple stations", stop_area)
elif StopArea.is_stop(m_el):
self.stops.add(k)
elif StopArea.is_platform(m_el):
self.platforms.add(k)
elif (entrance_type := m_el["tags"].get("railway")) in (
"subway_entrance",
"train_station_entrance",
):
if m_el["type"] != "node":
city.warn(f"{entrance_type} is not a node", m_el)
if (
m_el["tags"].get("entrance") != "exit"
and m["role"] != "exit_only"
):
self.entrances.add(k)
if (
m_el["tags"].get("entrance") != "entrance"
and m["role"] != "entry_only"
):
self.exits.add(k)
elif StopArea.is_track(m_el):
tracks_detected = True
if tracks_detected:
city.warn("Tracks in a stop_area relation", stop_area)
def _add_nearby_entrances(self, station: Station, city: City) -> None:
center = station.center
for entrance_el in (
el
for el in city.elements.values()
if "tags" in el
and (entrance_type := el["tags"].get("railway"))
in ("subway_entrance", "train_station_entrance")
):
entrance_id = el_id(entrance_el)
if entrance_id in city.stop_areas:
continue # This entrance belongs to some stop_area
c_center = el_center(entrance_el)
if (
c_center
and distance(center, c_center) <= MAX_DISTANCE_TO_ENTRANCES
):
if entrance_el["type"] != "node":
city.warn(f"{entrance_type} is not a node", entrance_el)
etag = entrance_el["tags"].get("entrance")
if etag != "exit":
self.entrances.add(entrance_id)
if etag != "entrance":
self.exits.add(entrance_id)
def get_elements(self):
result = {self.id, self.station.id}
result.update(self.entrances)
@ -1816,7 +1829,7 @@ class City:
if len(transfer) > 1:
self.transfers.append(transfer)
def extract_routes(self):
def extract_routes(self) -> None:
# Extract stations
processed_stop_areas = set()
for el in self.elements.values():
@ -1850,7 +1863,7 @@ class City:
# Check that stops and platforms belong to
# a single stop_area
for sp in station.stops | station.platforms:
for sp in chain(station.stops, station.platforms):
if sp in self.stops_and_platforms:
self.notice(
f"A stop or a platform {sp} belongs to "
@ -2328,7 +2341,7 @@ def find_transfers(elements, cities):
return transfers
def get_unused_entrances_geojson(elements):
def get_unused_subway_entrances_geojson(elements: list[dict]) -> dict:
global used_entrances
features = []
for el in elements:

25
tests/assets/tiny_world.osm
View file

@ -56,6 +56,27 @@
<node id='105' visible='true' version='1' lat='0.01441106473' lon='0.01235481987'>
<tag k='public_transport' v='stop_position' />
</node>
<node id='201' visible='true' version='1' lat='0.00967473055' lon='0.01007169217'>
<tag k='railway' v='subway_entrance' />
<tag k='ref' v='3-1' />
</node>
<node id='202' visible='true' version='1' lat='0.00966936613' lon='0.01018702716'>
<tag k='railway' v='train_station_entrance' />
<tag k='ref' v='3-2' />
</node>
<node id='203' visible='true' version='1' lat='0.01042574907' lon='0.00959962338'>
<tag k='railway' v='train_station_entrance' />
<tag k='ref' v='7-2' />
</node>
<node id='204' visible='true' version='1' lat='0.01034796501' lon='0.00952183932'>
<tag k='railway' v='subway_entrance' />
<tag k='ref' v='7-1' />
</node>
<node id='205' visible='true' version='1' lat='0.01015484596' lon='0.000201163'>
<tag k='note' v='Though the entrance is not in the stop_area, the preprocessor should add the entrance to it' />
<tag k='railway' v='subway_entrance' />
<tag k='ref' v='4-1' />
</node>
<node id='1001' visible='true' version='1' lat='0.01' lon='0.01' />
<way id='1' visible='true' version='1'>
<nd ref='1' />
@ -95,6 +116,8 @@
</relation>
<relation id='3' visible='true' version='1'>
<member type='node' ref='3' role='' />
<member type='node' ref='201' role='' />
<member type='node' ref='202' role='' />
<tag k='public_transport' v='stop_area' />
<tag k='type' v='public_transport' />
</relation>
@ -102,6 +125,8 @@
<member type='node' ref='7' role='' />
<member type='node' ref='102' role='' />
<member type='node' ref='104' role='' />
<member type='node' ref='203' role='' />
<member type='node' ref='204' role='' />
<tag k='public_transport' v='stop_area' />
<tag k='type' v='public_transport' />
</relation>

BIN
tests/assets/tiny_world_gtfs.zip
View file

Binary file not shown.

3
tests/assets/tiny_world_gtfs/agency.txt Normal file
View file

@ -0,0 +1,3 @@
agency_id,agency_name,agency_url,agency_timezone,agency_lang,agency_phone
1,Intersecting 2 metro lines,,,,
2,One light rail line,,,,

2
tests/assets/tiny_world_gtfs/calendar.txt Normal file
View file

@ -0,0 +1,2 @@
service_id,monday,tuesday,wednesday,thursday,friday,saturday,sunday,start_date,end_date
always,1,1,1,1,1,1,1,19700101,30000101

7
tests/assets/tiny_world_gtfs/frequencies.txt Normal file
View file

@ -0,0 +1,7 @@
trip_id,start_time,end_time,headway_secs,exact_times
r7,05:00:00,25:00:00,150.0,
r8,05:00:00,25:00:00,150.0,
r12,05:00:00,25:00:00,150.0,
r13,05:00:00,25:00:00,150.0,
r9,05:00:00,25:00:00,150.0,
r10,05:00:00,25:00:00,150.0,

4
tests/assets/tiny_world_gtfs/routes.txt Normal file
View file

@ -0,0 +1,4 @@
route_id,agency_id,route_short_name,route_long_name,route_desc,route_type,route_url,route_color,route_text_color,route_sort_order,route_fare_class,line_id,listed_route
r15,1,1,Blue Line,,1,,0000ff,,,,,
r14,1,2,Red Line,,1,,ff0000,,,,,
r11,2,LR,LR Line,,1,,a52a2a,,,,,

15
tests/assets/tiny_world_gtfs/shapes.txt Normal file
View file

@ -0,0 +1,15 @@
shape_id,shape_pt_lat,shape_pt_lon,shape_pt_sequence,shape_dist_traveled
7,0.0,0.0,0,
7,0.0047037,0.0047037,1,
7,0.0099397,0.0099397,2,
8,0.0099397,0.0099397,0,
8,0.0047037,0.0047037,1,
8,0.0,0.0,2,
12,0.01,0.0,0,
12,0.0,0.01,1,
13,0.0,0.01,0,
13,0.01,0.0,1,
9,0.0102531,0.0097675,0,
9,0.0143445,0.0124562,1,
10,0.0143597,0.012321,0,
10,0.0103197,0.0096662,1,

17
tests/assets/tiny_world_gtfs/stop_times.txt Normal file
View file

@ -0,0 +1,17 @@
trip_id,arrival_time,departure_time,stop_id,stop_sequence,stop_headsign,pickup_type,drop_off_type,shape_dist_traveled,timepoint,checkpoint_id,continuous_pickup,continuous_drop_off
r7,,,n1_plt,0,,,,0,,,,
r7,,,r1_plt,1,,,,741,,,,
r7,,,r3_plt,2,,,,1565,,,,
r8,,,r3_plt,0,,,,0,,,,
r8,,,r1_plt,1,,,,824,,,,
r8,,,n1_plt,2,,,,1565,,,,
r12,,,n4_plt,0,,,,0,,,,
r12,,,r2_plt,1,,,,758,,,,
r12,,,n6_plt,2,,,,1575,,,,
r13,,,n6_plt,0,,,,0,,,,
r13,,,r2_plt,1,,,,817,,,,
r13,,,n4_plt,2,,,,1575,,,,
r9,,,r4_plt,0,,,,0,,,,
r9,,,r16_plt,1,,,,545,,,,
r10,,,r16_plt,0,,,,0,,,,
r10,,,r4_plt,1,,,,538,,,,

27
tests/assets/tiny_world_gtfs/stops.txt Normal file
View file

@ -0,0 +1,27 @@
stop_id,stop_code,stop_name,stop_desc,platform_code,platform_name,stop_lat,stop_lon,zone_id,stop_address,stop_url,level_id,location_type,parent_station,wheelchair_boarding,municipality,on_street,at_street,vehicle_type
n1_egress,n1_egress,Station 1,,,,0.0,0.0,,,,,2,n1_st,,,,,
n1_plt,n1_plt,Station 1,,,,0.0,0.0,,,,,0,n1_st,,,,,
n1_st,n1_st,Station 1,,,,0.0,0.0,,,,,1,,,,,,
n201_r3,n201_r3,Station 3 3-1,,,,0.0096747,0.0100717,,,,,2,r3_st,,,,,
n202_r3,n202_r3,Station 3 3-2,,,,0.0096694,0.010187,,,,,2,r3_st,,,,,
n203_r4,n203_r4,Station 7 7-2,,,,0.0104257,0.0095996,,,,,2,r4_st,,,,,
n204_r4,n204_r4,Station 7 7-1,,,,0.010348,0.0095218,,,,,2,r4_st,,,,,
n205_n4,n205_n4,Station 4 4-1,,,,0.0101548,0.0002012,,,,,2,n4_st,,,,,
n4_plt,n4_plt,Station 4,,,,0.01,0.0,,,,,0,n4_st,,,,,
n4_st,n4_st,Station 4,,,,0.01,0.0,,,,,1,,,,,,
n6_egress,n6_egress,Station 6,,,,0.0,0.01,,,,,2,n6_st,,,,,
n6_plt,n6_plt,Station 6,,,,0.0,0.01,,,,,0,n6_st,,,,,
n6_st,n6_st,Station 6,,,,0.0,0.01,,,,,1,,,,,,
r16_egress,r16_egress,Station 8,,,,0.0143778,0.0124055,,,,,2,r16_st,,,,,
r16_plt,r16_plt,Station 8,,,,0.0143778,0.0124055,,,,,0,r16_st,,,,,
r16_st,r16_st,Station 8,,,,0.0143778,0.0124055,,,,,1,,,,,,
r1_egress,r1_egress,Station 2,,,,0.0047037,0.0047037,,,,,2,r1_st,,,,,
r1_plt,r1_plt,Station 2,,,,0.0047037,0.0047037,,,,,0,r1_st,,,,,
r1_st,r1_st,Station 2,,,,0.0047037,0.0047037,,,,,1,,,,,,
r2_egress,r2_egress,Station 5,,,,0.0051474,0.0047719,,,,,2,r2_st,,,,,
r2_plt,r2_plt,Station 5,,,,0.0051474,0.0047719,,,,,0,r2_st,,,,,
r2_st,r2_st,Station 5,,,,0.0051474,0.0047719,,,,,1,,,,,,
r3_plt,r3_plt,Station 3,,,,0.0097589,0.0101204,,,,,0,r3_st,,,,,
r3_st,r3_st,Station 3,,,,0.0097589,0.0101204,,,,,1,,,,,,
r4_plt,r4_plt,Station 7,,,,0.0102864,0.0097169,,,,,0,r4_st,,,,,
r4_st,r4_st,Station 7,,,,0.0102864,0.0097169,,,,,1,,,,,,

5
tests/assets/tiny_world_gtfs/transfers.txt Normal file
View file

@ -0,0 +1,5 @@
from_stop_id,to_stop_id,transfer_type,min_transfer_time
r3_st,r4_st,0,106
r4_st,r3_st,0,106
r1_st,r2_st,0,81
r2_st,r1_st,0,81

7
tests/assets/tiny_world_gtfs/trips.txt Normal file
View file

@ -0,0 +1,7 @@
route_id,service_id,trip_id,trip_headsign,trip_short_name,direction_id,block_id,shape_id,wheelchair_accessible,trip_route_type,route_pattern_id,bikes_allowed
r15,always,r7,,,,,7,,,,
r15,always,r8,,,,,8,,,,
r14,always,r12,,,,,12,,,,
r14,always,r13,,,,,13,,,,
r11,always,r9,,,,,9,,,,
r11,always,r10,,,,,10,,,,

41
tests/sample_data_for_outputs.py
View file

@ -20,7 +20,7 @@ metro_samples = [
"networks": "network-2",
},
],
"gtfs_file": "assets/tiny_world_gtfs.zip",
"gtfs_dir": "assets/tiny_world_gtfs",
"json_dump": """
{
"stopareas": {
@ -49,7 +49,20 @@ metro_samples = [
0.0097589171
],
"name": "Station 3",
"entrances": []
"entrances": [
{
"id": "n201",
"name": null,
"ref": "3-1",
"center": [0.01007169217, 0.00967473055]
},
{
"id": "n202",
"name": null,
"ref": "3-2",
"center": [0.01018702716, 0.00966936613]
}
]
},
"n4": {
"id": "n4",
@ -58,7 +71,14 @@ metro_samples = [
0.01
],
"name": "Station 4",
"entrances": []
"entrances": [
{
"id": "n205",
"name": null,
"ref": "4-1",
"center": [0.000201163, 0.01015484596]
}
]
},
"r2": {
"id": "r2",
@ -85,7 +105,20 @@ metro_samples = [
0.010286367745
],
"name": "Station 7",
"entrances": []
"entrances": [
{
"id": "n204",
"name": null,
"ref": "7-1",
"center": [0.00952183932, 0.01034796501]
},
{
"id": "n203",
"name": null,
"ref": "7-2",
"center": [0.00959962338, 0.01042574907]
}
]
},
"r16": {
"id": "r16",

21
tests/test_gtfs_processor.py
View file

@ -1,13 +1,11 @@
import codecs
import csv
from functools import partial
from pathlib import Path
from zipfile import ZipFile
from processors._common import transit_to_dict
from processors.gtfs import dict_to_row, GTFS_COLUMNS, transit_data_to_gtfs
from tests.util import TestCase
from tests.sample_data_for_outputs import metro_samples
from tests.util import TestCase
class TestGTFS(TestCase):
@ -108,20 +106,19 @@ class TestGTFS(TestCase):
)
control_gtfs_data = self._readGtfs(
Path(__file__).resolve().parent / metro_sample["gtfs_file"]
Path(__file__).resolve().parent / metro_sample["gtfs_dir"]
)
self._compareGtfs(calculated_gtfs_data, control_gtfs_data)
@staticmethod
def _readGtfs(filepath: str) -> dict:
def _readGtfs(gtfs_dir: Path) -> dict:
gtfs_data = dict()
with ZipFile(filepath) as zf:
for gtfs_feature in GTFS_COLUMNS:
with zf.open(f"{gtfs_feature}.txt") as f:
reader = csv.reader(codecs.iterdecode(f, "utf-8"))
next(reader) # read header
rows = list(reader)
gtfs_data[gtfs_feature] = rows
for gtfs_feature in GTFS_COLUMNS:
with open(gtfs_dir / f"{gtfs_feature}.txt") as f:
reader = csv.reader(f)
next(reader) # read header
rows = list(reader)
gtfs_data[gtfs_feature] = rows
return gtfs_data
def _compareGtfs(

163
tests/test_overpass.py Normal file
View file

@ -0,0 +1,163 @@
from unittest import TestCase, mock
from process_subways import compose_overpass_request, overpass_request
class TestOverpassQuery(TestCase):
def test__compose_overpass_request__no_bboxes(self) -> None:
bboxes = []
for overground in (True, False):
with self.subTest(msg=f"{overground=}"):
with self.assertRaises(RuntimeError):
compose_overpass_request(overground, bboxes)
def test__compose_overpass_request__one_bbox(self) -> None:
bboxes = [[1, 2, 3, 4]]
expected = {
False: (
"[out:json][timeout:1000];"
"("
"("
'rel[route="light_rail"](1,2,3,4);'
'rel[route="monorail"](1,2,3,4);'
'rel[route="subway"](1,2,3,4);'
'rel[route="train"](1,2,3,4);'
");"
"rel(br)[type=route_master];"
"node[railway=subway_entrance](1,2,3,4);"
"node[railway=train_station_entrance](1,2,3,4);"
"rel[public_transport=stop_area](1,2,3,4);"
"rel(br)[type=public_transport]"
"[public_transport=stop_area_group];"
");"
"(._;>>;);"
"out body center qt;"
),
True: (
"[out:json][timeout:1000];"
"("
"("
'rel[route="aerialway"](1,2,3,4);'
'rel[route="bus"](1,2,3,4);'
'rel[route="ferry"](1,2,3,4);'
'rel[route="tram"](1,2,3,4);'
'rel[route="trolleybus"](1,2,3,4);'
");"
"rel(br)[type=route_master];"
"rel[public_transport=stop_area](1,2,3,4);"
"rel(br)[type=public_transport]"
"[public_transport=stop_area_group];"
");"
"(._;>>;);"
"out body center qt;"
),
}
for overground, expected_answer in expected.items():
with self.subTest(msg=f"{overground=}"):
self.assertEqual(
expected_answer,
compose_overpass_request(overground, bboxes),
)
def test__compose_overpass_request__several_bboxes(self) -> None:
bboxes = [[1, 2, 3, 4], [5, 6, 7, 8]]
expected = {
False: (
"[out:json][timeout:1000];"
"("
"("
'rel[route="light_rail"](1,2,3,4);'
'rel[route="monorail"](1,2,3,4);'
'rel[route="subway"](1,2,3,4);'
'rel[route="train"](1,2,3,4);'
");"
"rel(br)[type=route_master];"
"node[railway=subway_entrance](1,2,3,4);"
"node[railway=train_station_entrance](1,2,3,4);"
"rel[public_transport=stop_area](1,2,3,4);"
"rel(br)[type=public_transport][public_transport=stop_area_group];" # noqa E501
"("
'rel[route="light_rail"](5,6,7,8);'
'rel[route="monorail"](5,6,7,8);'
'rel[route="subway"](5,6,7,8);'
'rel[route="train"](5,6,7,8);'
");"
"rel(br)[type=route_master];"
"node[railway=subway_entrance](5,6,7,8);"
"node[railway=train_station_entrance](5,6,7,8);"
"rel[public_transport=stop_area](5,6,7,8);"
"rel(br)[type=public_transport][public_transport=stop_area_group];" # noqa E501
");"
"(._;>>;);"
"out body center qt;"
),
True: (
"[out:json][timeout:1000];"
"("
"("
'rel[route="aerialway"](1,2,3,4);'
'rel[route="bus"](1,2,3,4);'
'rel[route="ferry"](1,2,3,4);'
'rel[route="tram"](1,2,3,4);'
'rel[route="trolleybus"](1,2,3,4);'
");"
"rel(br)[type=route_master];"
"rel[public_transport=stop_area](1,2,3,4);"
"rel(br)[type=public_transport][public_transport=stop_area_group];" # noqa E501
"("
'rel[route="aerialway"](5,6,7,8);'
'rel[route="bus"](5,6,7,8);'
'rel[route="ferry"](5,6,7,8);'
'rel[route="tram"](5,6,7,8);'
'rel[route="trolleybus"](5,6,7,8);'
");"
"rel(br)[type=route_master];"
"rel[public_transport=stop_area](5,6,7,8);"
"rel(br)[type=public_transport][public_transport=stop_area_group];" # noqa E501
");"
"(._;>>;);"
"out body center qt;"
),
}
for overground, expected_answer in expected.items():
with self.subTest(msg=f"{overground=}"):
self.assertEqual(
expected_answer,
compose_overpass_request(overground, bboxes),
)
def test__overpass_request(self) -> None:
overpass_api = "http://overpass.example/"
overground = False
bboxes = [[1, 2, 3, 4]]
expected_url = (
"http://overpass.example/?data="
"%5Bout%3Ajson%5D%5Btimeout%3A1000%5D%3B%28%28"
"rel%5Broute%3D%22light_rail%22%5D%281%2C2%2C3%2C4"
"%29%3Brel%5Broute%3D%22monorail%22%5D%281%2C2%2C3%2C4%29%3B"
"rel%5Broute%3D%22subway%22%5D%281%2C2%2C3%2C4%29%3B"
"rel%5Broute%3D%22train%22%5D%281%2C2%2C3%2C4%29%3B%29%3B"
"rel%28br%29%5Btype%3Droute_master%5D%3B"
"node%5Brailway%3Dsubway_entrance%5D%281%2C2%2C3%2C4%29%3B"
"node%5Brailway%3Dtrain_station_entrance%5D%281%2C2%2C3%2C4%29%3B"
"rel%5Bpublic_transport%3Dstop_area%5D%281%2C2%2C3%2C4%29%3B"
"rel%28br%29%5Btype%3Dpublic_transport%5D%5Bpublic_transport%3D"
"stop_area_group%5D%3B%29%3B"
"%28._%3B%3E%3E%3B%29%3Bout%20body%20center%20qt%3B"
)
with mock.patch("process_subways.json.load") as load_mock:
load_mock.return_value = {"elements": []}
with mock.patch(
"process_subways.urllib.request.urlopen"
) as urlopen_mock:
urlopen_mock.return_value.getcode.return_value = 200
overpass_request(overground, overpass_api, bboxes)
urlopen_mock.assert_called_once_with(expected_url, timeout=1000)

46
tests/test_station.py Normal file
View file

@ -0,0 +1,46 @@
from unittest import TestCase
from subway_structure import Station
class TestStation(TestCase):
def test__get_modes(self) -> None:
cases = [
{"element": {"tags": {"railway": "station"}}, "modes": set()},
{
"element": {
"tags": {"railway": "station", "station": "train"}
},
"modes": {"train"},
},
{
"element": {"tags": {"railway": "station", "train": "yes"}},
"modes": {"train"},
},
{
"element": {
"tags": {
"railway": "station",
"station": "subway",
"train": "yes",
}
},
"modes": {"subway", "train"},
},
{
"element": {
"tags": {
"railway": "station",
"subway": "yes",
"train": "yes",
"light_rail": "yes",
"monorail": "yes",
}
},
"modes": {"subway", "train", "light_rail", "monorail"},
},
]
for case in cases:
element = case["element"]
expected_modes = case["modes"]
self.assertSetEqual(expected_modes, Station.get_modes(element))