diff --git a/.gitignore b/.gitignore index 274a41c..7b5b868 100644 --- a/.gitignore +++ b/.gitignore @@ -9,4 +9,8 @@ resources/germany.osm.pbf resources/monaco.osm.pbf resources/performance_report.json python/performance/performance_report.json -pgpass.conf \ No newline at end of file +pgpass.conf + +# poetry environment exclusions +poetry.lock +pyproject.toml diff --git a/SQL/after_import.sql b/SQL/after_import.sql index dd3fff5..f7f3ffa 100644 --- a/SQL/after_import.sql +++ b/SQL/after_import.sql @@ -1,65 +1,69 @@ -\echo 'Altering column nodes.id to integer...' +DO $$ +BEGIN +RAISE NOTICE 'Altering column nodes.id to integer...'; ALTER TABLE nodes ALTER COLUMN id TYPE integer; -\echo 'Altering column ways.id to integer...' +RAISE NOTICE 'Altering column ways.id to integer...'; ALTER TABLE ways ALTER COLUMN id TYPE integer; -\echo 'Altering column relations.id to integer...' +RAISE NOTICE 'Altering column relations.id to integer...'; ALTER TABLE relations ALTER COLUMN id TYPE integer; -\echo 'Altering column nodes_ways.id to integer...' +RAISE NOTICE 'Altering column nodes_ways.id to integer...'; ALTER TABLE nodes_ways ALTER COLUMN way_id TYPE integer; -- PRIMARY KEYS -\echo 'Adding PRIMARY KEY constraints to table nodes...' +RAISE NOTICE 'Adding PRIMARY KEY constraints to table nodes...'; ALTER TABLE nodes ADD CONSTRAINT pk_nodes PRIMARY KEY (id); -\echo 'Adding PRIMARY KEY constraints to table nodes_ways' +RAISE NOTICE 'Adding PRIMARY KEY constraints to table nodes_ways'; ALTER TABLE nodes_ways ADD CONSTRAINT nodes_ways_pk PRIMARY KEY (id); -\echo 'Adding PRIMARY KEY constraints to table ways...' +RAISE NOTICE 'Adding PRIMARY KEY constraints to table ways...'; ALTER TABLE ways ADD CONSTRAINT pk_ways PRIMARY KEY (id); -\echo 'Adding PRIMARY KEY constraints to table relations...' +RAISE NOTICE 'Adding PRIMARY KEY constraints to table relations...'; ALTER TABLE relations ADD CONSTRAINT pk_relations PRIMARY KEY (id); -- FOREIGN KEYS -- ways table foreign keys -\echo 'Adding FOREIGN KEY constraints to table ways...' +RAISE NOTICE 'Adding FOREIGN KEY constraints to table ways...'; ALTER TABLE ways ADD CONSTRAINT fk_ways_from FOREIGN KEY ("from") REFERENCES nodes(id); ALTER TABLE ways ADD CONSTRAINT fk_ways_to FOREIGN KEY ("to") REFERENCES nodes(id); ALTER TABLE ways ADD CONSTRAINT fk_ways_area FOREIGN KEY (area) REFERENCES areas(id); -- nodes table foreign key -\echo 'Adding FOREIGN KEY constraints to table nodes...' +RAISE NOTICE 'Adding FOREIGN KEY constraints to table nodes...'; ALTER TABLE nodes ADD CONSTRAINT fk_nodes_area FOREIGN KEY (area) REFERENCES areas(id); -- edges table foreign keys -\echo 'Adding FOREIGN KEY constraints to table edges...' +RAISE NOTICE 'Adding FOREIGN KEY constraints to table edges...'; ALTER TABLE edges ADD CONSTRAINT fk_edges_from FOREIGN KEY ("from") REFERENCES nodes(id); ALTER TABLE edges ADD CONSTRAINT fk_edges_to FOREIGN KEY ("to") REFERENCES nodes(id); -- trip_locations table foreign keys -\echo 'Adding FOREIGN KEY constraints to table trip_location...' +RAISE NOTICE 'Adding FOREIGN KEY constraints to table trip_location...'; ALTER TABLE trip_locations ADD CONSTRAINT fk_trip_locations_destination FOREIGN KEY (destination) REFERENCES nodes(id); ALTER TABLE trip_locations ADD CONSTRAINT fk_trip_locations_origin FOREIGN KEY (origin) REFERENCES nodes(id); -- nodes_ways table foreign key -\echo 'Adding FOREIGN KEY constraints to table nodes_ways...' +RAISE NOTICE 'Adding FOREIGN KEY constraints to table nodes_ways...'; ALTER TABLE nodes_ways ADD CONSTRAINT fk_nodes_ways_area FOREIGN KEY (area) REFERENCES areas(id); -- nodes_ways_speeds table foreign keys -\echo 'Adding FOREIGN KEY constraints to table nodes_ways_speeds...' +RAISE NOTICE 'Adding FOREIGN KEY constraints to table nodes_ways_speeds...'; ALTER TABLE nodes_ways_speeds ADD CONSTRAINT fk_nodes_ways_speeds_from FOREIGN KEY (from_node_ways_id) REFERENCES nodes_ways(id); ALTER TABLE nodes_ways_speeds ADD CONSTRAINT fk_nodes_ways_speeds_to FOREIGN KEY (to_node_ways_id) REFERENCES nodes_ways(id); +END +$$; diff --git a/SQL/main.sql b/SQL/main.sql index 96471d0..ba473d9 100644 --- a/SQL/main.sql +++ b/SQL/main.sql @@ -154,7 +154,7 @@ CREATE TABLE IF NOT EXISTS public.demand ( -- CREATE TABLE IF NOT EXISTS public.nodes ( - id bigint NOT NULL, + id integer NOT NULL, geom public.geometry(Point,4326) NOT NULL, area integer, contracted boolean DEFAULT false NOT NULL @@ -522,12 +522,12 @@ CREATE TABLE IF NOT EXISTS public.trip_times ( -- CREATE TABLE IF NOT EXISTS public.ways ( - id bigint NOT NULL, + id integer NOT NULL, tags public.hstore, geom public.geometry(Geometry,4326) NOT NULL, area integer, - "from" bigint NOT NULL, - "to" bigint NOT NULL, + "from" integer NOT NULL, + "to" integer NOT NULL, oneway boolean NOT NULL ); diff --git a/SQL/testing_extension.sql b/SQL/testing_extension.sql index 6728d96..476d469 100644 --- a/SQL/testing_extension.sql +++ b/SQL/testing_extension.sql @@ -262,25 +262,25 @@ BEGIN -- drop every sequence in test_scheme_name FOR tmp IN (SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = test_scheme_name) LOOP - EXECUTE format('DROP SEQUENCE %I.%I', test_scheme_name, tmp); + EXECUTE format('DROP SEQUENCE %I.%I CASCADE', test_scheme_name, tmp); END LOOP; -- drop every view in test_scheme_name FOR table_name_i IN (SELECT table_name FROM information_schema.views WHERE table_schema = test_scheme_name) LOOP - EXECUTE format('DROP VIEW %I.%I', test_scheme_name, table_name_i); + EXECUTE format('DROP VIEW %I.%I CASCADE', test_scheme_name, table_name_i); END LOOP; -- drop every table in test_scheme_name FOR table_name_i IN (SELECT table_name FROM information_schema.tables WHERE table_schema = test_scheme_name) LOOP - EXECUTE format('DROP TABLE %I.%I', test_scheme_name, table_name_i); + EXECUTE format('DROP TABLE %I.%I CASCADE', test_scheme_name, table_name_i); END LOOP; -- drop routines in test_scheme_name FOR table_name_i IN (SELECT routine_name FROM information_schema.routines WHERE routine_schema = test_scheme_name) LOOP - EXECUTE format('DROP FUNCTION %I.%I', test_scheme_name, table_name_i); + EXECUTE format('DROP FUNCTION %I.%I CASCADE', test_scheme_name, table_name_i); END LOOP; -- update search path diff --git a/python/roadgraphtool/db.py b/python/roadgraphtool/db.py index 3252cfa..fd67589 100644 --- a/python/roadgraphtool/db.py +++ b/python/roadgraphtool/db.py @@ -1,17 +1,22 @@ import atexit import logging -import psycopg2 -import sshtunnel -import sqlalchemy +from pathlib import Path + +import geopandas as gpd import pandas as pd +import psycopg2 import psycopg2.errors -import geopandas as gpd -from pathlib import Path +import sqlalchemy +import sshtunnel from sqlalchemy.engine import Row from roadgraphtool.credentials_config import CREDENTIALS -logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s', datefmt='%H:%M:%S') +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + datefmt="%H:%M:%S", +) def connect_db_if_required(db_function): @@ -21,7 +26,7 @@ def connect_db_if_required(db_function): def wrapper(*args, **kwargs): db = args[0] - if hasattr(db, 'server'): + if hasattr(db, "server"): db.start_or_restart_ssh_connection_if_needed() if not db.is_connected(): db.set_up_db_connections() @@ -39,6 +44,7 @@ class __Database: Import as: from db import db """ + config = CREDENTIALS def __init__(self): @@ -46,7 +52,7 @@ def __init__(self): self.db_server_port = self.config.db_server_port self.db_name = self.config.db_name self.ssh_tunnel_local_port = 1113 - if hasattr(self.config, 'server'): + if hasattr(self.config, "server"): self.server = self.config.server self.ssh_server = None self.host = self.config.host @@ -58,7 +64,9 @@ def __init__(self): self._sql_alchemy_engine_str = None def is_connected(self): - return (self._psycopg2_connection is not None) and (self._sqlalchemy_engine is not None) + return (self._psycopg2_connection is not None) and ( + self._sqlalchemy_engine is not None + ) def set_up_db_connections(self): # psycopg2 connection object @@ -75,22 +83,24 @@ def set_ssh_to_db_server_and_set_port(self): ssh_pkey=self.config.private_key_path, ssh_username=self.config.server_username, ssh_private_key_password=self.config.private_key_phrase, - remote_bind_address=('localhost', self.db_server_port), - local_bind_address=('localhost', self.ssh_tunnel_local_port) + remote_bind_address=("localhost", self.db_server_port), + local_bind_address=("localhost", self.ssh_tunnel_local_port), ) try: self.ssh_server = sshtunnel.open_tunnel(self.server, **ssh_kwargs) except sshtunnel.paramiko.SSHException as e: # sshtunnel dependency paramiko may attempt to use ssh-agent and crashes if it fails logging.warning(f"sshtunnel.paramiko.SSHException: '{e}'") - self.ssh_server = sshtunnel.open_tunnel(self.server, **ssh_kwargs, allow_agent=False) + self.ssh_server = sshtunnel.open_tunnel( + self.server, **ssh_kwargs, allow_agent=False + ) self.ssh_server.start() logging.info( "SSH tunnel established from %s to %s/%s", self.ssh_server.local_bind_address, self.ssh_server.ssh_host, - self.db_server_port + self.db_server_port, ) self.db_server_port = self.ssh_server.local_bind_port @@ -110,12 +120,15 @@ def start_or_restart_ssh_connection_if_needed(self): self.ssh_server.restart() def get_sql_alchemy_engine_str(self): - sql_alchemy_engine_str = 'postgresql+psycopg2://{user}:{password}@{host}:{port}/{dbname}'.format( - user=self.config.username, - password=self.config.db_password, - host=self.host, - port=self.db_server_port, - dbname=self.db_name) + sql_alchemy_engine_str = ( + "postgresql+psycopg2://{user}:{password}@{host}:{port}/{dbname}".format( + user=self.config.username, + password=self.config.db_password, + host=self.host, + port=self.db_server_port, + dbname=self.db_name, + ) + ) return sql_alchemy_engine_str @@ -129,7 +142,7 @@ def get_new_psycopg2_connection(self): password=self.config.db_password, host=self.config.db_host, port=self.db_server_port, - dbname=self.db_name + dbname=self.db_name, ) atexit.register(psycopg2_connection.close) @@ -148,29 +161,40 @@ def execute_sql(self, query, *args, schema='public', use_transactions=True) -> N if not use_transactions: connection.execution_options(isolation_level="AUTOCOMMIT") with connection.begin(): - connection.execute(sqlalchemy.text(f"SET search_path TO {schema};")) + search_path = connection.execute(sqlalchemy.text("SHOW search_path;")).all()[0][0] + connection.execute(sqlalchemy.text(f"SET search_path TO {','.join([schema, search_path])};")) connection.execute(sqlalchemy.text(query), *args) - connection.execute(sqlalchemy.text(f"SET search_path TO public;")) + connection.execute(sqlalchemy.text(f"SET search_path TO {search_path};")) @connect_db_if_required - def execute_sql_and_fetch_all_rows(self, query, *args) -> list[Row]: + def execute_sql_and_fetch_all_rows(self, query, *args, schema=None) -> list[Row]: with self._sqlalchemy_engine.connect() as conn: - result = conn.execute(sqlalchemy.text(query), *args).all() + result = None + if schema: + search_path = conn.execute(sqlalchemy.text("SHOW search_path;")).all()[0][0] + conn.execute(sqlalchemy.text(f"SET search_path TO {','.join([schema, search_path])};")) + result = conn.execute(sqlalchemy.text(query), *args).all() + conn.execute(sqlalchemy.text(f"SET search_path TO {search_path};")) + else: + result = conn.execute(sqlalchemy.text(query), *args).all() return result @connect_db_if_required - def execute_script(self, script_path: Path): + def execute_script(self, script_path: Path) -> int: with open(script_path) as f: script = f.read() cursor = self._psycopg2_connection.cursor() + retcode = 0 try: cursor.execute(script) self._psycopg2_connection.commit() except Exception as e: logging.error(f"Error executing script {script_path}: {e}") self._psycopg2_connection.rollback() + retcode = 1 finally: cursor.close() + return retcode def set_schema(self, schema: str): """ @@ -215,7 +239,9 @@ def execute_query_to_pandas(self, sql: str, **kwargs) -> pd.DataFrame: return data @connect_db_if_required - def dataframe_to_db_table(self, df: pd.DataFrame, table_name: str, **kwargs) -> None: + def dataframe_to_db_table( + self, df: pd.DataFrame, table_name: str, **kwargs + ) -> None: """ Save DataFrame to a new table in the database @@ -224,7 +250,9 @@ def dataframe_to_db_table(self, df: pd.DataFrame, table_name: str, **kwargs) -> https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.to_sql.html """ - df.to_sql(table_name, con=self._sqlalchemy_engine, if_exists='append', index=False) + df.to_sql( + table_name, con=self._sqlalchemy_engine, if_exists="append", index=False + ) @connect_db_if_required def db_table_to_pandas(self, table_name: str, **kwargs) -> pd.DataFrame: diff --git a/python/roadgraphtool/db_operations.py b/python/roadgraphtool/db_operations.py new file mode 100644 index 0000000..968e69c --- /dev/null +++ b/python/roadgraphtool/db_operations.py @@ -0,0 +1,93 @@ +from roadgraphtool.db import db + +from .insert_area import insert_area + + +def get_area_for_demand( + srid_plain: int, + dataset_ids: list, + zone_types: list, + buffer_meters: int, + min_requests_in_zone: int, + datetime_min: str, + datetime_max: str, + center_point: tuple, + max_distance_from_center_point_meters: int, +) -> list: + sql_query = """ + select * from get_area_for_demand( + srid_plain := :srid_plain, + dataset_ids := (:dataset_ids)::smallint[], + zone_types := (:zone_types)::smallint[], + buffer_meters := (:buffer_meters)::smallint, + min_requests_in_zone := (:min_requests_in_zone)::smallint, + datetime_min := :datetime_min, + datetime_max := :datetime_max, + center_point := st_makepoint(:center_x, :center_y), + max_distance_from_center_point_meters := (:max_distance_from_center_point_meters)::smallint + );""" + params = { + "srid_plain": srid_plain, + "dataset_ids": dataset_ids, + "zone_types": zone_types, + "buffer_meters": buffer_meters, + "min_requests_in_zone": min_requests_in_zone, + "datetime_min": datetime_min, + "datetime_max": datetime_max, + "center_x": center_point[0], + "center_y": center_point[1], + "max_distance_from_center_point_meters": max_distance_from_center_point_meters, + } + return db.execute_sql_and_fetch_all_rows(sql_query, params) + + +def contract_graph_in_area( + target_area_id: int, target_area_srid: int, fill_speed: bool = True, schema=None +): + sql_query = f'call public.contract_graph_in_area({target_area_id}::smallint, {target_area_srid}::int{", FALSE" if not fill_speed else ""})' + db.execute_sql(sql_query, schema=schema) + + +def select_network_nodes_in_area(target_area_id: int) -> list: + sql_query = ( + f"select * from select_network_nodes_in_area({target_area_id}::smallint)" + ) + return db.execute_sql_and_fetch_all_rows(sql_query) + + +def assign_average_speed_to_all_segments_in_area( + target_area_id: int, target_area_srid: int +): + sql_query = ( + f"call public.assign_average_speed_to_all_segments_in_area({target_area_id}::smallint, " + f"{target_area_srid}::int)" + ) + db.execute_sql(sql_query) + + +def compute_strong_components(target_area_id: int, schema=None): + sql_query = f"call public.compute_strong_components({target_area_id}::smallint)" + db.execute_sql(sql_query, schema=schema) + + +def compute_speeds_for_segments( + target_area_id: int, + speed_records_dataset: int, + hour: int, + day_of_week: int, +): + sql_query = ( + f"call public.compute_speeds_for_segments({target_area_id}::smallint, " + f"{speed_records_dataset}::smallint, {hour}::smallint, {day_of_week}::smallint)" + ) + db.execute_sql(sql_query) + + +def compute_speeds_from_neighborhood_segments( + target_area_id: int, target_area_srid: int +): + sql_query = ( + f"call public.compute_speeds_from_neighborhood_segments({target_area_id}::smallint, " + f"{target_area_srid}::int)" + ) + db.execute_sql(sql_query) diff --git a/python/roadgraphtool/insert_area.py b/python/roadgraphtool/insert_area.py index c3d97c9..5c6de6e 100644 --- a/python/roadgraphtool/insert_area.py +++ b/python/roadgraphtool/insert_area.py @@ -2,10 +2,10 @@ import json import sys -from .db import db +from roadgraphtool.db import db -def insert_area(id: int | None, name: str, description: str | None, geom: dict): +def insert_area(id: int | None, name: str, description: str | None, geom: dict, schema: str | None = None): """ Insert a new area into the areas table. @@ -18,16 +18,21 @@ def insert_area(id: int | None, name: str, description: str | None, geom: dict): Returns: None """ + schema = "public" if schema is None else schema + if description is None: description = "" + # result ignored as the pgsql function returns void if id is None: db.execute_sql( - f"SELECT insert_area('{name}', '{json.dumps(geom)}', NULL, '{description}')" + f"SELECT insert_area('{name}', '{json.dumps(geom)}', NULL, '{description}')", + schema=schema ) else: db.execute_sql( - f"SELECT insert_area('{name}', '{json.dumps(geom)}', {id}, '{description}')" + f"SELECT insert_area('{name}', '{json.dumps(geom)}', {id}, '{description}')", + schema=schema ) diff --git a/python/roadgraphtool/map.py b/python/roadgraphtool/map.py index 4fc7f86..953464e 100644 --- a/python/roadgraphtool/map.py +++ b/python/roadgraphtool/map.py @@ -22,10 +22,11 @@ def add_node_highway_tags(nodes, G): nodes.loc[nodes.index[[v]], 'highway'] = tag -def _get_map_from_db(config: dict) -> Tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]: - nodes = get_map_nodes_from_db(config['area_id']) +def _get_map_from_db(config: dict, schema: str | None = None) -> Tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]: + schema = schema if schema else 'public' + nodes = get_map_nodes_from_db(config['area_id'], schema) logging.info(f"{len(nodes)} nodes fetched from db") - edges = get_map_edges_from_db(config) + edges = get_map_edges_from_db(config, schema) logging.info(f"{len(edges)} edges fetched from db") return nodes, edges @@ -91,7 +92,7 @@ def _save_graph_shapefile(nodes: gpd.GeoDataFrame, edges: gpd.GeoDataFrame, shap edges.to_file(str(filepath_edges), driver="ESRI Shapefile", index=False, encoding="utf-8") -def get_map(config: Dict) -> Tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]: +def get_map(config: Dict, schema: str | None = None) -> Tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]: """ Loads filtered map nodes geodataframe. If th dataframe is not generated yet, then the map is downloaded and processed to obtain the filtered nodes dataframe @@ -121,7 +122,7 @@ def get_map(config: Dict) -> Tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]: if 'place' in config['map']: nodes, edges = _get_map(config) else: - nodes, edges = _get_map_from_db(config) + nodes, edges = _get_map_from_db(config, schema=schema) # save map to shapefile (for visualising) map_dir = config["map"]["path"] @@ -135,7 +136,7 @@ def get_map(config: Dict) -> Tuple[gpd.GeoDataFrame, gpd.GeoDataFrame]: # Filter nodes by config/area. Only these nodes should be used for demand/vehicle generation/selection if 'area' in config: sql = f"""SELECT geom FROM areas WHERE name = '{config['area']}'""" - area_shape = db.execute_query_to_geopandas(sql) + area_shape = db.execute_query_to_geopandas(sql, schema=schema if schema else 'public') mask = nodes.within(area_shape.loc[0, 'geom']) nodes = nodes.loc[mask] diff --git a/python/scripts/filter_osm.py b/python/scripts/filter_osm.py index 3f55c0f..48c8378 100644 --- a/python/scripts/filter_osm.py +++ b/python/scripts/filter_osm.py @@ -1,12 +1,14 @@ import argparse from pathlib import Path -import re +import logging import os +import re import subprocess import tempfile from typing import Any + import requests -import logging + from roadgraphtool.exceptions import InvalidInputError, MissingInputError @@ -16,7 +18,9 @@ def setup_logger(logger_name: str) -> logging.Logger: log = logging.getLogger(logger_name) log.setLevel(logging.INFO) # setup formatting - formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') + formatter = logging.Formatter( + "%(asctime)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" + ) handler = logging.StreamHandler() handler.setLevel(logging.INFO) handler.setFormatter(formatter) @@ -25,23 +29,29 @@ def setup_logger(logger_name: str) -> logging.Logger: log.propagate = False return log -logger = setup_logger('filter_osm') + +logger = setup_logger("filter_osm") + def is_valid_extension(file: str) -> bool: """Return True if the file has a valid extension. - + Valid extensions: osm, osm.pbf, osm.bz2 """ valid_extensions = ["osm", "osm.pbf", "osm.bz2"] return any(file.endswith(f".{ext}") for ext in valid_extensions) + def check_strategy(strategy: str | None): """Raise InvalidInputError if strategy type is not valid.""" valid_strategies = ["simple", "complete_ways", "smart"] if strategy and strategy not in valid_strategies: - raise InvalidInputError(f"Invalid strategy type. Call {os.path.basename(__file__)} -h/--help to display help.") + raise InvalidInputError( + f"Invalid strategy type. Call {os.path.basename(__file__)} -h/--help to display help." + ) logger.debug("Strategy validity checked.") + def load_multipolygon_by_id(relation_id: str) -> bytes | Any: """Return multipolygon content based on relation ID.""" url = f"https://www.openstreetmap.org/api/0.6/relation/{relation_id}/full" @@ -50,6 +60,7 @@ def load_multipolygon_by_id(relation_id: str) -> bytes | Any: logger.debug("Multipolygon content loaded.") return response.content + def extract_id(input_file: str, relation_id: str, strategy: str = None): """Filter out data based on relation ID.""" logger.debug("Extracting multipolygon with relation ID %s...") @@ -58,20 +69,37 @@ def extract_id(input_file: str, relation_id: str, strategy: str = None): with tempfile.NamedTemporaryFile(delete=False, suffix=".osm") as tmp_file: tmp_file.write(content) tmp_file_path = tmp_file.name - cmd = ["osmium", "extract", "-p", tmp_file_path, input_file, "-o", RESOURCES_DIR / "id_extract.osm"] + cmd = [ + "osmium", + "extract", + "-p", + tmp_file_path, + input_file, + "-o", + RESOURCES_DIR / "id_extract.osm", + ] if strategy: cmd.extend(["-s", strategy]) res = subprocess.run(cmd) if not res.returncode: logger.debug("ID extraction completed.") - + + def extract_bbox(input_file: str, coords: str, strategy: str = None): """Extract data based on bounding box with osmium.""" - float_regex = r'[0-9]+(.[0-9]+)?' # should match four floats - coords_regex = f'{float_regex},{float_regex},{float_regex},{float_regex}' + float_regex = r"[0-9]+(.[0-9]+)?" # should match four floats + coords_regex = f"{float_regex},{float_regex},{float_regex},{float_regex}" if re.match(coords_regex, coords): logger.debug("Extracting bounding box with coords %s...", coords) - cmd = ["osmium", "extract", "-b", coords, input_file, "-o", "extracted-bbox.osm.pbf"] + cmd = [ + "osmium", + "extract", + "-b", + coords, + input_file, + "-o", + "extracted-bbox.osm.pbf", + ] elif os.path.isfile(coords) and coords.endswith((".json", ".geojson")): logger.debug("Extracting bounding box with coords in file %s...", coords) cmd = ["osmium", "extract", "-c", coords, input_file] @@ -80,28 +108,38 @@ def extract_bbox(input_file: str, coords: str, strategy: str = None): if strategy: cmd.extend(["-s", strategy]) - + res = subprocess.run(cmd) if not res.returncode: logger.info("Bounding box extraction completed.") + def run_osmium_filter(input_file: str, expression_file: str, omit_referenced: bool): """Filter objects based on tags in expression file. - Untagged nodes and members referenced in ways and relations respectively will not + Untagged nodes and members referenced in ways and relations respectively will not be added to output if omit_referenced set to True. """ - cmd = ["osmium", "tags-filter", input_file, "-e", expression_file, "-o", "filtered.osm.pbf"] + cmd = [ + "osmium", + "tags-filter", + input_file, + "-e", + expression_file, + "-o", + "filtered.osm.pbf", + ] if omit_referenced: cmd.extend(["-R"]) res = subprocess.run(cmd) if not res.returncode: logger.info("Tag filtering completed.") + def filter_highways(input_file: str, omit_referenced: bool): - """Filter objects with highway tag. - - Untagged nodes and members referenced in ways and relations respectively will not + """Filter objects with highway tag. + + Untagged nodes and members referenced in ways and relations respectively will not be added to output if omit_referenced set to True. """ content = "nwr/highway" @@ -139,31 +177,36 @@ def parse_args(arg_list: list[str] | None) -> argparse.Namespace: return args + def main(arg_list: list[str] | None = None): args = parse_args(arg_list) if not os.path.exists(args.input_file): raise FileNotFoundError(f"File '{args.input_file}' does not exist.") elif not is_valid_extension(args.input_file): - raise InvalidInputError("File must have one of the following extensions: osm, osm.pbf, osm.bz2") - + raise InvalidInputError( + "File must have one of the following extensions: osm, osm.pbf, osm.bz2" + ) + match args.flag: case "id": # Filter geographic objects based on relation ID if not args.relation_id: raise MissingInputError("Existing relation ID must be specified.") - + check_strategy(args.strategy) - + extract_id(args.input_file, args.relation_id, args.strategy) case "b": # Filter geographic objects based on bounding box (with osmium) if not args.coords: - raise MissingInputError("Coordinates or config file need to be specified with the 'b' flag.") - + raise MissingInputError( + "Coordinates or config file need to be specified with the 'b' flag." + ) + check_strategy(args.strategy) - + extract_bbox(args.input_file, args.coords, args.strategy) case "f": @@ -171,13 +214,17 @@ def main(arg_list: list[str] | None = None): if not args.expression_file: raise MissingInputError("Expression file needs to be specified.") elif not os.path.exists(args.expression_file): - raise FileNotFoundError(f"File '{args.expression_file}' does not exist.") + raise FileNotFoundError( + f"File '{args.expression_file}' does not exist." + ) - run_osmium_filter(args.input_file, args.expression_file, args.omit_referenced) + run_osmium_filter( + args.input_file, args.expression_file, args.omit_referenced + ) case "h": # Filter objects based on highway tag filter_highways(args.input_file, args.omit_referenced) -if __name__ == '__main__': - main() \ No newline at end of file +if __name__ == "__main__": + main() diff --git a/python/scripts/find_bbox.py b/python/scripts/find_bbox.py index e2941c0..51588e3 100644 --- a/python/scripts/find_bbox.py +++ b/python/scripts/find_bbox.py @@ -1,19 +1,19 @@ -import xml.etree.ElementTree as ET import sys +import xml.etree.ElementTree as ET def find_min_max(xml) -> tuple[float, float, float, float]: """Return tuple of floats representing bounding box borders.""" root = ET.fromstring(xml) - min_lon = float('inf') - min_lat = float('inf') - max_lon = float('-inf') - max_lat = float('-inf') + min_lon = float("inf") + min_lat = float("inf") + max_lon = float("-inf") + max_lat = float("-inf") - for node in root.findall('.//node'): - lat = float(node.get('lat')) - lon = float(node.get('lon')) + for node in root.findall(".//node"): + lat = float(node.get("lat")) + lon = float(node.get("lon")) min_lon = min(min_lon, lon) min_lat = min(min_lat, lat) max_lon = max(max_lon, lon) diff --git a/python/scripts/install_sql.py b/python/scripts/install_sql.py index f93d91e..f43deca 100644 --- a/python/scripts/install_sql.py +++ b/python/scripts/install_sql.py @@ -5,89 +5,100 @@ from roadgraphtool.db import db db_name = db.db_name -sql_dir = Path(__file__).parent.parent.parent / "SQL" +SQL_DIR = Path(__file__).parent.parent.parent / "SQL" -def execute_sql_file(sql_file: Path, multistatement: bool = False): +def execute_sql_file(sql_file: Path, multistatement: bool = False) -> int: logging.info(f"Executing {sql_file}") + retcode = 0 if multistatement: - db.execute_script(sql_file) + retcode = db.execute_script(sql_file) else: with sql_file.open() as f: sql = f.read() db.execute_sql(sql) - - -logging.basicConfig(level=logging.INFO) - -# Check availability status of extensions in db -logging.info("Checking availability of extensions in the database") -extension_list = ["postgis", "pgrouting", "hstore", "pgtap"] - -sql = f""" -WITH extension_list(name) AS ( - VALUES - {','.join(f"('{ext}')" for ext in extension_list)} -) -SELECT - el.name AS extension_name, - CASE - WHEN ae.name IS NOT NULL THEN 'Available' - ELSE 'Not Available' - END AS status -FROM extension_list el -LEFT JOIN pg_available_extensions ae ON el.name = ae.name -ORDER BY el.name; -""" - -extensions = {} -for extension_name, status in db.execute_sql_and_fetch_all_rows(sql): - logging.info(f"{extension_name}: {status}") - extensions[extension_name] = status == "Available" - -# Assert that critically needed extensions are present -if not extensions["postgis"] or not extensions["pgrouting"] or not extensions["hstore"]: - raise Exception("Missing critical extensions") - -# initialize database if it's empty -sql = f"""SELECT EXISTS ( - SELECT FROM information_schema.tables - WHERE table_schema = 'public' AND table_name = 'areas' - ); - """ -if not db.execute_sql_and_fetch_all_rows(sql)[0][0]: - main_sql_path = sql_dir / "main.sql" - logging.info("Initializing the database") - execute_sql_file(main_sql_path, multistatement=True) - -# enable pgtap if it's not enabled -sql = """SELECT * FROM pg_extension WHERE extname = 'pgtap'""" -if extensions["pgtap"] and not db.execute_sql_and_fetch_all_rows(sql): - logging.info("Enabling pgtap") - execute_sql_file(sql_dir / "testing_extension.sql", multistatement=True) -else: - if not extensions["pgtap"]: - logging.warning("pgtap extension is not available") + return retcode + + +def main(): + logging.basicConfig(level=logging.INFO) + + # Check availability status of extensions in db + logging.info("Checking availability of extensions in the database") + extension_list = ["postgis", "pgrouting", "hstore", "pgtap"] + + sql = f""" + WITH extension_list(name) AS ( + VALUES + {','.join(f"('{ext}')" for ext in extension_list)} + ) + SELECT + el.name AS extension_name, + CASE + WHEN ae.name IS NOT NULL THEN 'Available' + ELSE 'Not Available' + END AS status + FROM extension_list el + LEFT JOIN pg_available_extensions ae ON el.name = ae.name + ORDER BY el.name; + """ + + extensions = {} + for extension_name, status in db.execute_sql_and_fetch_all_rows(sql): + logging.info(f"{extension_name}: {status}") + extensions[extension_name] = status == "Available" + + # Assert that critically needed extensions are present + if ( + not extensions["postgis"] + or not extensions["pgrouting"] + or not extensions["hstore"] + ): + raise Exception("Missing critical extensions") + + # initialize database if it's empty + sql = f"""SELECT EXISTS ( + SELECT FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'areas' + ); + """ + if not db.execute_sql_and_fetch_all_rows(sql)[0][0]: + main_sql_path = SQL_DIR / "main.sql" + logging.info("Initializing the database") + execute_sql_file(main_sql_path, multistatement=True) + + # enable pgtap if it's not enabled + sql = """SELECT * FROM pg_extension WHERE extname = 'pgtap'""" + if extensions["pgtap"] and not db.execute_sql_and_fetch_all_rows(sql): + logging.info("Enabling pgtap") + execute_sql_file(SQL_DIR / "testing_extension.sql", multistatement=True) else: - logging.info("pgtap extension is already enabled") - -# Create test schema if it doesn't exist -sql = """SELECT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'test_env')""" -if not db.execute_sql_and_fetch_all_rows(sql)[0][0]: - logging.info("Creating test schema") - db.execute_sql("CREATE SCHEMA test_env") - -functions_dir = sql_dir / "functions" -logging.info("Importing functions from %s", functions_dir) -for sql_function_file in functions_dir.rglob("*.sql"): - execute_sql_file(sql_function_file) - -procedures_dir = sql_dir / "procedures" -logging.info("Importing procedures from %s", procedures_dir) -for sql_procedure_file in procedures_dir.rglob("*.sql"): - execute_sql_file(sql_procedure_file) - -test_dir = sql_dir / "tests" -logging.info("Importing test functions from %s", test_dir) -for sql_test_file in test_dir.rglob("*.sql"): - execute_sql_file(sql_test_file, multistatement=True) + if not extensions["pgtap"]: + logging.warning("pgtap extension is not available") + else: + logging.info("pgtap extension is already enabled") + + # Create test schema if it doesn't exist + sql = """SELECT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'test_env')""" + if not db.execute_sql_and_fetch_all_rows(sql)[0][0]: + logging.info("Creating test schema") + db.execute_sql("CREATE SCHEMA test_env") + + functions_dir = SQL_DIR / "functions" + logging.info("Importing functions from %s", functions_dir) + for sql_function_file in functions_dir.rglob("*.sql"): + execute_sql_file(sql_function_file) + + procedures_dir = SQL_DIR / "procedures" + logging.info("Importing procedures from %s", procedures_dir) + for sql_procedure_file in procedures_dir.rglob("*.sql"): + execute_sql_file(sql_procedure_file) + + test_dir = SQL_DIR / "tests" + logging.info("Importing test functions from %s", test_dir) + for sql_test_file in test_dir.rglob("*.sql"): + execute_sql_file(sql_test_file, multistatement=True) + + +if __name__ == "__main__": + exit(main()) diff --git a/python/scripts/main.py b/python/scripts/main.py index 884f111..fe61438 100644 --- a/python/scripts/main.py +++ b/python/scripts/main.py @@ -1,111 +1,21 @@ import argparse -import json import logging + import psycopg2.errors import yaml import types +from roadgraphtool.db_operations import ( + assign_average_speed_to_all_segments_in_area, compute_speeds_for_segments, + compute_speeds_from_neighborhood_segments, compute_strong_components, + contract_graph_in_area, get_area_for_demand, insert_area, + select_network_nodes_in_area) from roadgraphtool.credentials_config import CREDENTIALS -from roadgraphtool.db import db from scripts.process_osm import import_osm_to_db, DEFAULT_STYLE_FILE from roadgraphtool.export import get_map_nodes_from_db -def get_area_for_demand( - srid_plain: int, - dataset_ids: list, - zone_types: list, - buffer_meters: int, - min_requests_in_zone: int, - datetime_min: str, - datetime_max: str, - center_point: tuple, - max_distance_from_center_point_meters: int, -) -> list: - sql_query = """ - select * from get_area_for_demand( - srid_plain := :srid_plain, - dataset_ids := (:dataset_ids)::smallint[], - zone_types := (:zone_types)::smallint[], - buffer_meters := (:buffer_meters)::smallint, - min_requests_in_zone := (:min_requests_in_zone)::smallint, - datetime_min := :datetime_min, - datetime_max := :datetime_max, - center_point := st_makepoint(:center_x, :center_y), - max_distance_from_center_point_meters := (:max_distance_from_center_point_meters)::smallint - );""" - params = { - "srid_plain": srid_plain, - "dataset_ids": dataset_ids, - "zone_types": zone_types, - "buffer_meters": buffer_meters, - "min_requests_in_zone": min_requests_in_zone, - "datetime_min": datetime_min, - "datetime_max": datetime_max, - "center_x": center_point[0], - "center_y": center_point[1], - "max_distance_from_center_point_meters": max_distance_from_center_point_meters, - } - return db.execute_sql_and_fetch_all_rows(sql_query, params) - - -def insert_area(name: str, coordinates: list): - geom_json = {"type": "MultiPolygon", "coordinates": coordinates} - params = {"name": name, "json_data": json.dumps(geom_json)} - sql_query = """insert into areas (name, geom) values (:name, st_geomfromgeojson(:json_data))""" - db.execute_sql(sql_query, params) - - -def contract_graph_in_area( - target_area_id: int, target_area_srid: int, fill_speed: bool = True -): - sql_query = f'call public.contract_graph_in_area({target_area_id}::smallint, {target_area_srid}::int{", FALSE" if not fill_speed else ""})' - db.execute_sql(sql_query) - - -def select_network_nodes_in_area(target_area_id: int) -> list: - sql_query = ( - f"select * from select_network_nodes_in_area({target_area_id}::smallint)" - ) - return db.execute_sql_and_fetch_all_rows(sql_query) - - -def assign_average_speed_to_all_segments_in_area( - target_area_id: int, target_area_srid: int -): - sql_query = ( - f"call public.assign_average_speed_to_all_segments_in_area({target_area_id}::smallint, " - f"{target_area_srid}::int)" - ) - db.execute_sql(sql_query) - - -def compute_strong_components(target_area_id: int): - sql_query = f"call public.compute_strong_components({target_area_id}::smallint)" - db.execute_sql(sql_query) - - -def compute_speeds_for_segments( - target_area_id: int, speed_records_dataset: int, hour: int, day_of_week: int -): - sql_query = ( - f"call public.compute_speeds_for_segments({target_area_id}::smallint, " - f"{speed_records_dataset}::smallint, {hour}::smallint, {day_of_week}::smallint)" - ) - db.execute_sql(sql_query) - - -def compute_speeds_from_neighborhood_segments( - target_area_id: int, target_area_srid: int -): - sql_query = ( - f"call public.compute_speeds_from_neighborhood_segments({target_area_id}::smallint, " - f"{target_area_srid}::int)" - ) - db.execute_sql(sql_query) - - def configure_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="File containing the main flow of your application" @@ -164,9 +74,9 @@ def configure_arg_parser() -> argparse.ArgumentParser: required=False ) parser.add_argument( - "-W", - dest="password", - action="store_true", + "-W", + dest="password", + action="store_true", help="Force password prompt instead of using pgpass file.") @@ -253,5 +163,5 @@ def main(arg_list: list[str] | None = None): compute_speeds_from_neighborhood_segments(area_id, area_srid) -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/python/scripts/process_osm.py b/python/scripts/process_osm.py index b111b8e..01fc010 100644 --- a/python/scripts/process_osm.py +++ b/python/scripts/process_osm.py @@ -18,7 +18,8 @@ POSTPROCESS_DICT = {"pipeline.lua": "after_import.sql"} -logger = setup_logger('process_osm') +logger = setup_logger("process_osm") + def extract_bbox(relation_id: int) -> tuple[float, float, float, float]: """Return tuple of floats based on bounding box coordinations.""" @@ -27,10 +28,13 @@ def extract_bbox(relation_id: int) -> tuple[float, float, float, float]: logger.debug(f"Bounding box found: {min_lon},{min_lat},{max_lon},{max_lat}.") return min_lon, min_lat, max_lon, max_lat + def run_osmium_cmd(flag: str, input_file: str, output_file: str = None): """Run osmium command based on flag.""" if output_file and not is_valid_extension(output_file): - raise InvalidInputError("File must have one of the following extensions: osm, osm.pbf, osm.bz2") + raise InvalidInputError( + "File must have one of the following extensions: osm, osm.pbf, osm.bz2" + ) match flag: case "d": subprocess.run(["osmium", "show", input_file]) @@ -38,27 +42,29 @@ def run_osmium_cmd(flag: str, input_file: str, output_file: str = None): subprocess.run(["osmium", "fileinfo", input_file]) case "ie": subprocess.run(["osmium", "fileinfo", "-e", input_file]) - case 'r': + case "r": res = subprocess.run(["osmium", "renumber", input_file, "-o", output_file]) if not res.returncode: logger.info("Renumbering of OSM data completed.") - case 's': + case "s": res = subprocess.run(["osmium", "sort", input_file, "-o", output_file]) if not res.returncode: logger.info("Sorting of OSM data completed.") - case 'sr': - tmp_file = 'tmp.osm' + case "sr": + tmp_file = "tmp.osm" res = subprocess.run(["osmium", "sort", input_file, "-o", tmp_file]) if not res.returncode: logger.info("Sorting of OSM data completed.") - res = subprocess.run(["osmium", "renumber", tmp_file, "-o", output_file]) + res = subprocess.run( + ["osmium", "renumber", tmp_file, "-o", output_file] + ) if not res.returncode: logger.info("Renumbering of OSM data completed.") os.remove(tmp_file) def setup_ssh_tunnel(config: CredentialsConfig) -> int: """Set up SSH tunnel if needed and returns port number.""" - if hasattr(config, "server"): # remote connection + if hasattr(config, "server") and config.server is not None: # remote connection db.start_or_restart_ssh_connection_if_needed() config.db_server_port = db.ssh_tunnel_local_port return db.ssh_tunnel_local_port @@ -84,19 +90,19 @@ def run_osm2pgsql_cmd(config: CredentialsConfig, input_file: str, style_file_pat if logger.level == logging.DEBUG: cmd.extend(['--log-level=debug']) - + if not pgpass: cmd.extend(["-W"]) - + logger.info(f"Begin importing...") logger.debug(' '.join(cmd)) if pgpass: logger.info("Setting up pgpass file...") config.setup_pgpass() - + res = subprocess.run(cmd).returncode - + if pgpass: logger.info("Deleting pgpass file...") config.remove_pgpass() @@ -109,10 +115,10 @@ def postprocess_osm_import(config: CredentialsConfig, style_file_path: str, sche """Apply postprocessing SQL associated with **style_file_path** to data in **schema** after importing. """ style_file_path = os.path.basename(style_file_path) - + if style_file_path in POSTPROCESS_DICT: sql_file_path = str(SQL_DIR / POSTPROCESS_DICT[style_file_path]) - cmd = ["psql", "-d", config.db_name, "-U", config.username, "-h", config.db_host, "-p", + cmd = ["psql", "-d", config.db_name, "-U", config.username, "-h", config.db_host, "-p", str(config.db_server_port), "-c", f"SET search_path TO {schema};", "-f", sql_file_path] logger.info("Post-processing OSM data after import...") @@ -147,6 +153,7 @@ def import_osm_to_db(input_file: str, force: bool, pgpass: bool, style_file_path + def parse_args(arg_list: list[str] | None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Process OSM files and interact with PostgreSQL database.", formatter_class=argparse.RawTextHelpFormatter) @@ -180,30 +187,35 @@ def parse_args(arg_list: list[str] | None) -> argparse.Namespace: return args + def main(arg_list: list[str] | None = None): args = parse_args(arg_list) if not os.path.exists(args.input_file): raise FileNotFoundError(f"File '{args.input_file}' does not exist.") elif not is_valid_extension(args.input_file): - raise InvalidInputError("File must have one of the following extensions: osm, osm.pbf, osm.bz2.") + raise InvalidInputError( + "File must have one of the following extensions: osm, osm.pbf, osm.bz2." + ) elif args.style_file: if not os.path.exists(args.style_file): raise FileNotFoundError(f"File '{args.style_file}' does not exist.") elif not str(args.style_file).endswith(".lua"): raise InvalidInputError("File must have the '.lua' extension.") - + match args.flag: - case 'd' | 'i' | 'ie': + case "d" | "i" | "ie": # Display content or (extended) information of OSM file run_osmium_cmd(args.flag, args.input_file) - case 's' | 'r' | 'sr': + case "s" | "r" | "sr": # Sort, renumber OSM file or do both if not args.output_file: - raise MissingInputError("An output file must be specified with '-o' flag.") + raise MissingInputError( + "An output file must be specified with '-o' flag." + ) run_osmium_cmd(args.flag, args.input_file, args.output_file) - + case "u": # Preprocess and upload OSM file to PostgreSQL database and then postprocess the data import_osm_to_db(args.input_file, args.force, args.pgpass, args.style_file, args.schema) diff --git a/python/tests/data/integration_area.json b/python/tests/data/integration_area.json new file mode 100644 index 0000000..88e08ea --- /dev/null +++ b/python/tests/data/integration_area.json @@ -0,0 +1,29 @@ +{ + "type": "MultiPolygon", + "coordinates": [ + [ + [ + [ + 13.403920115445864, + 52.51912951075764 + ], + [ + 13.404591964637179, + 52.518789787533855 + ], + [ + 13.409067805024401, + 52.51991834946755 + ], + [ + 13.407071182779786, + 52.5223884170219 + ], + [ + 13.403920115445864, + 52.51912951075764 + ] + ] + ] + ] +} diff --git a/python/tests/data/integration_test.osm b/python/tests/data/integration_test.osm new file mode 100644 index 0000000..d38f753 --- /dev/null +++ b/python/tests/data/integration_test.osm @@ -0,0 +1,43 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/python/tests/integrations_test.py b/python/tests/integrations_test.py new file mode 100644 index 0000000..e5eff55 --- /dev/null +++ b/python/tests/integrations_test.py @@ -0,0 +1,293 @@ +import hashlib +import os +import shutil +import xml.etree.ElementTree as ET +from pathlib import Path + +import pytest + +from roadgraphtool.db import db +from roadgraphtool.db_operations import (compute_strong_components, + contract_graph_in_area) +from roadgraphtool.insert_area import insert_area +from roadgraphtool.insert_area import read_json_file as read_area +from roadgraphtool.map import get_map as export_nodes_edges +from scripts.install_sql import main as pre_pocessing +from scripts.process_osm import import_osm_to_db + +TEST_DATA_PATH = Path(__file__).parent / "data" +TMP_DIR = Path(__file__).parent / "TMP" +TEST_SCHEMA = "test_env" + +# Fixtures + + +@pytest.fixture() +def cleanup_for_base_flow(): + yield # wait for the call after test + + # 5) Environment destruction + # remove all created files + if TMP_DIR.is_file() or TMP_DIR.is_symlink(): + TMP_DIR.unlink() + elif TMP_DIR.is_dir(): + shutil.rmtree(TMP_DIR) + + # test environment desctructor + db.execute_sql("CALL test_env_destructor();") + # TODO: remove tables from public scheme + + +# Helper functions + + +def check_setup_of_database(): + query = """SELECT EXISTS ( + SELECT FROM information_schema.tables + WHERE table_schema = 'public' + AND table_name = 'areas' +);""" + return db.execute_count_query(query) + + +def parse_osm_file(file_path): + # Parse the XML file + tree = ET.parse(file_path) + root = tree.getroot() + + # Initialize the dictionary to store the OSM data + osm_data = {"nodes": {}, "ways": {}} + + # Parse nodes + for node in root.findall("node"): + node_id = node.get("id") + node_data = { + "id": node_id, + "lat": node.get("lat"), + "lon": node.get("lon"), + "version": node.get("version"), + "timestamp": node.get("timestamp"), + "uid": node.get("uid"), + "user": node.get("user"), + "tags": {}, + } + + # Parse tags for nodes + for tag in node.findall("tag"): + node_data["tags"][tag.get("k")] = tag.get("v") + + osm_data["nodes"][node_id] = node_data + + # Parse ways + for way in root.findall("way"): + way_id = way.get("id") + way_data = { + "id": way_id, + "version": way.get("version"), + "timestamp": way.get("timestamp"), + "uid": way.get("uid"), + "user": way.get("user"), + "nodes": [], + "tags": {}, + } + + # Parse node references for ways + for nd in way.findall("nd"): + way_data["nodes"].append(nd.get("ref")) + + # Parse tags for ways + for tag in way.findall("tag"): + way_data["tags"][tag.get("k")] = tag.get("v") + + osm_data["ways"][way_id] = way_data + + return osm_data + + +def get_file_md5_hash(file_path): + with open(file_path, "rb") as f: + return hashlib.md5(f.read()).hexdigest() + + +# Testing functions + + +def test_integration_base_flow(cleanup_for_base_flow): + """ + Integration test: + Base flow: + 1) Import test data: + i) set up basic tables (optionally) + ii) import osm file + iii) add test area + 2) Execute contraction of graph in the area + 3) Execute computation of strong components in the area + 4) Export data to files + 5) Destroy testing environment + """ + + # 1) setting up the database with needed information + if not check_setup_of_database(): + pre_pocessing() + + # change the main schema + db.execute_sql("CALL test_env_constructor();") + + import_osm_to_db(str(TEST_DATA_PATH / "integration_test.osm"), force=True, pgpass=False, schema=TEST_SCHEMA) + + insert_area( + 1, + "Deutschland", + "test area", + read_area(str(TEST_DATA_PATH / "integration_area.json")), + schema=f"{TEST_SCHEMA}, public" + ) + + # read osm file to dictionary + osm_dict = parse_osm_file(TEST_DATA_PATH / "integration_test.osm") + + # Test that importing osm data and area was done successfully + nodes = db.execute_sql_and_fetch_all_rows( + "SELECT id, ST_X(geom), ST_Y(geom) FROM nodes;", + schema=TEST_SCHEMA + ) + nodes_set = set(nodes) + + expected_nodes = osm_dict["nodes"] + expected_nodes_set = set( + [ + (int(key), float(value["lon"]), float(value["lat"])) + for key, value in expected_nodes.items() + ] + ) + + assert nodes_set == expected_nodes_set + + ways = db.execute_sql_and_fetch_all_rows( + 'SELECT id, tags, "from", "to", oneway FROM ways;', + schema=TEST_SCHEMA + ) + + ways_set = set( + [ + (way[0], str(dict(sorted(way[1].items()))), way[2], way[3], way[4]) + for way in ways + ] + ) + + expected_ways = osm_dict["ways"] + expected_ways_set = set( + [ + ( + int(key), + str( + { + k: v for k, v in value["tags"].items() if k != "oneway" + } # the same tags, but without "oneway" tag + ), + int(value["nodes"][0]), + int(value["nodes"][-1]), + value["tags"]["oneway"] == "yes", + ) + for key, value in expected_ways.items() + ] + ) + + assert expected_ways_set == ways_set + + nodes_ways = db.execute_sql_and_fetch_all_rows( + "SELECT way_id, node_id FROM nodes_ways;", + schema=TEST_SCHEMA + ) + nodes_ways_set = set(nodes_ways) + + expected_nodes_ways = set() + for way_id, way in expected_ways.items(): + for node_id in way["nodes"]: + expected_nodes_ways.add((int(way_id), int(node_id))) + + assert nodes_ways_set == expected_nodes_ways + + area = db.execute_sql_and_fetch_all_rows( + "SELECT id, name, description FROM areas;", + schema=TEST_SCHEMA + )[0] + + expected_area = (1, "Deutschland", "test area") + + assert area == expected_area + + # 2) Call contraction + contract_graph_in_area(1, 4326, fill_speed=False, schema=TEST_SCHEMA) + + # GET updated data from db and assert + contracted_nodes = db.execute_sql_and_fetch_all_rows( + "SELECT id FROM nodes WHERE contracted;", + schema=TEST_SCHEMA + ) + assert {(6,), (7,), (8,)} == set(contracted_nodes) + + edges = db.execute_sql_and_fetch_all_rows('SELECT "from", "to" FROM edges;', schema=TEST_SCHEMA) + + assert {(3, 4), (2, 1), (1, 2), (5, 3), (5, 2), (4, 3), (2, 3), (3, 2)} == set( + edges + ) + + # 3) Call Compute Strong Components + compute_strong_components(1, schema=TEST_SCHEMA) + + # GET updated data from db and assert + component_data = db.execute_sql_and_fetch_all_rows( + "SELECT component_id, node_id FROM component_data", + schema=TEST_SCHEMA + ) + + assert {(0, 1), (0, 2), (0, 3), (0, 4), (1, 5)} == set(component_data) + + # 4) Export data to files + os.mkdir(str(TMP_DIR)) + map_path = TMP_DIR / "map" + area_dir = TMP_DIR / "area_dir" + os.mkdir(str(map_path)) + os.mkdir(str(area_dir)) + + config = { + "map": { + "path": str(map_path), + "SRID_plane": 4326, + }, + "area_dir": str(area_dir), + "area": "Deutschland", + "area_id": 1, + } + _ = export_nodes_edges(config, schema=f"{TEST_SCHEMA}, \"$user\"") + + # assert created files by expected hash + expected_hashes = { + str(map_path / "edges.csv"): "96b1bc002e872cc13ba6d83823889070", + str(map_path / "nodes.csv"): "7d582c1906dd68bd1c38b99baa6da191", + str(map_path / "shapefiles/edges.cpg"): "ae3b3df9970b49b6523e608759bc957d", + str(map_path / "shapefiles/edges.dbf"): "0d327b70d2394bed6a6fd412d7886bc8", + str(map_path / "shapefiles/edges.prj"): "c742bee3d4edfc2948a2ad08de1790a5", + str(map_path / "shapefiles/edges.shp"): "5854b916fbaf5dadef0d18891c8aa4c8", + str(map_path / "shapefiles/edges.shx"): "0544fbe8da9bfff9360bae6b2f1fefe9", + str(map_path / "shapefiles/nodes.cpg"): "ae3b3df9970b49b6523e608759bc957d", + str(map_path / "shapefiles/nodes.dbf"): "f774112b1580f9d93c7afe45e2a5ee9f", + str(map_path / "shapefiles/nodes.prj"): "c742bee3d4edfc2948a2ad08de1790a5", + str(map_path / "shapefiles/nodes.shp"): "6a5b95f286089ad19ba943410c25d5f5", + str(map_path / "shapefiles/nodes.shx"): "499035dc9713ee99510385119f067710", + } + passage = True + result_dict = dict() + + for file_path, expected_hash in expected_hashes.items(): + actual_hash = get_file_md5_hash(file_path) + result = expected_hash != actual_hash + result_dict[file_path] = { + "assertion_result": result, + "message": "Ok" if result else f"File content mismatch for {file_path}", + } + if not result: + passage = result + + assert passage, f"Result dictionary of assertion: {result_dict}" diff --git a/python/tests/test_filter_osm.py b/python/tests/test_filter_osm.py index 4bcc152..058d9d6 100644 --- a/python/tests/test_filter_osm.py +++ b/python/tests/test_filter_osm.py @@ -6,7 +6,7 @@ from roadgraphtool.exceptions import InvalidInputError, MissingInputError from scripts.filter_osm import check_strategy, extract_id, is_valid_extension, load_multipolygon_by_id, extract_bbox, main, RESOURCES_DIR - + TESTS_DIR = pathlib.Path(__file__).parent.parent.parent / "python/tests/data" @pytest.fixture