diff --git a/anyway/parsers/cbs/dictionary_tables.py b/anyway/parsers/cbs/dictionary_tables.py index 0fefd1ba..700de180 100644 --- a/anyway/parsers/cbs/dictionary_tables.py +++ b/anyway/parsers/cbs/dictionary_tables.py @@ -1,221 +1,222 @@ -import glob -import logging -import math -import os -import re -from collections import defaultdict - -import pandas as pd - -from anyway.app_and_db import db -from anyway.models import ProviderCode -from anyway.utilities import ImporterUI - -DICTCOLUMN1 = "ms_tavla" -DICTCOLUMN2 = "kod" -DICTCOLUMN3 = "teur" -ACCIDENT_TYPE_REGEX = re.compile(r"accidents_type_(?P\d)") -DICTIONARY_FILENAME = "Dictionary.csv" - -TABLES_DICT = { - 0: "columns_description", - 2: "road_type", - 3: "entrance_exit", - 4: "accident_severity", - 5: "accident_type", - 6: "road_alignment", - 7: "infrastructure_type", - 8: "road_geometry", - 10: "one_lane", - 11: "multi_lane", - 12: "speed_limit", - 13: "road_intactness", - 14: "road_width", - 16: "road_light", - 17: "road_control", - 18: "weather", - 19: "road_surface", - 20: "vehicle_purpose", - 21: "road_object", - 22: "object_distance", - 23: "didnt_cross", - 24: "cross_mode", - 25: "cross_location", - 26: "cross_direction", - 28: "driving_directions", - 29: "vehicle_damage", - 31: "involved_type", - 33: "safety_measures_use", # need to verify if exists in new format - 34: "safety_measures", - 35: "injury_severity", - 37: "day_type", - 38: "day_night", - 39: "day_in_week", - 40: "traffic_light", - 42: "engine_volume", - 43: "vehicle_attribution", - 44: "total_weight", - 45: "vehicle_type", - 46: "late_deceased", - 47: "location_accuracy", - 48: "vehicle_type", - 50: "injured_type", - 52: "injured_position", - 60: "accident_month", - 66: "population_type", - 67: "sex", - 68: "geo_area", - 77: "region", - 78: "municipal_status", - 79: "district", - 80: "natural_area", - 81: "yishuv_shape", - 92: "age_group", - 93: "accident_hour_raw", -} - - -def read_dictionary(dictionary_file): - from anyway.parsers.cbs.executor import read_cbs_file - - cbs_dictionary = defaultdict(dict) - dictionary = read_cbs_file(dictionary_file) - for _, dic in dictionary.iterrows(): - cbs_dictionary[int(dic[DICTCOLUMN1])][int(dic[DICTCOLUMN2])] = dic[DICTCOLUMN3] - return cbs_dictionary - - -def fill_dictionary_tables(cbs_dictionary, provider_code, year): - if year < 2008: - return - for k, v in cbs_dictionary.items(): - if k == 27: - continue - try: - curr_table = TABLES_DICT[k] - except Exception as _: - logging.debug( - "A key " + str(k) + " was added to dictionary - update models, tables and classes" - ) - continue - for inner_k, inner_v in v.items(): - curr_table = TABLES_DICT[k] - if inner_v is None or (isinstance(inner_v, float) and math.isnan(inner_v)): - continue - sql_delete = ( - "DELETE FROM " - + curr_table - + " WHERE provider_code=" - + str(provider_code) - + " AND year=" - + str(year) - + " AND id=" - + str(inner_k) - ) - db.session.execute(sql_delete) - sql_insert = ( - "INSERT INTO " - + curr_table - + " VALUES (" - + str(inner_k) - + "," - + str(year) - + "," - + str(provider_code) - + "," - + "'" - + inner_v.replace("'", "") - + "'" - + ")" - + " ON CONFLICT DO NOTHING" - ) - db.session.execute(sql_insert) - try: - db.session.commit() - except Exception as e: - logging.error(f"Error updating Dictionary tables: {e}") - db.session.rollback() - logging.debug("Inserted/Updated dictionary values into table " + curr_table) - create_provider_code_table() - - -def truncate_dictionary_tables(dictionary_file): - cbs_dictionary = read_dictionary(dictionary_file) - for k, _ in cbs_dictionary.items(): - if k == 97: - continue - curr_table = TABLES_DICT[k] - sql_truncate = "TRUNCATE TABLE " + curr_table - db.session.execute(sql_truncate) - db.session.commit() - logging.debug("Truncated table " + curr_table) - - -def create_provider_code_table(): - provider_code_table = "provider_code" - provider_code_class = ProviderCode - table_entries = db.session.query(provider_code_class) - table_entries.delete() - provider_code_dict = { - 1: "הלשכה המרכזית לסטטיסטיקה - סוג תיק 1", - 2: "איחוד הצלה", - 3: "הלשכה המרכזית לסטטיסטיקה - סוג תיק 3", - 4: "שומרי הדרך", - } - for k, v in provider_code_dict.items(): - sql_insert = ( - "INSERT INTO " + provider_code_table + " VALUES (" + str(k) + "," + "'" + v + "'" + ")" - ) - db.session.execute(sql_insert) - try: - db.session.commit() - except Exception as e: - logging.error(f"Error updating table {provider_code_table}: {e}") - db.session.rollback() - - -def get_provider_code(directory_name=None): - if directory_name: - match = ACCIDENT_TYPE_REGEX.match(directory_name) - if match: - return int(match.groupdict()["type"]) - - ans = "" - while not ans.isdigit(): - ans = input("Directory provider code is invalid. Please enter a valid code: ") - if ans.isdigit(): - return int(ans) - - -def update_dictionary_tables(path): - import_ui = ImporterUI(path) - dir_name = import_ui.source_path() - dir_list = glob.glob("{0}/*/*".format(dir_name)) - - for directory in sorted(dir_list, reverse=True): - print(directory) - directory_name = os.path.basename(os.path.normpath(directory)) - year = directory_name[1:5] if directory_name[0] == "H" else directory_name[0:4] - if int(year) < 2008: - continue - parent_directory = os.path.basename(os.path.dirname(os.path.join(os.pardir, directory))) - provider_code = get_provider_code(parent_directory) - logging.debug("Importing Directory " + directory) - dictionary_file = _get_dictionary_file(directory) - if not dictionary_file: - return 0 - logging.debug("Filling dictionary for directory '{}'".format(directory)) - fill_dictionary_tables(read_dictionary(dictionary_file), provider_code, int(year)) - - -def _get_dictionary_file(directory): - files = [ - path - for path in os.listdir(directory) - if DICTIONARY_FILENAME.lower() in path.lower() and not path.startswith(".") - ] - if not files: - raise ValueError("Not found: '%s'" % DICTIONARY_FILENAME) - if len(files) > 1: - raise ValueError("Ambiguous: '%s'" % DICTIONARY_FILENAME) - return os.path.join(directory, files[0]) +import glob +import logging +import math +import os +import re +from collections import defaultdict + +import pandas as pd + +from anyway.app_and_db import db +from anyway.models import ProviderCode +from anyway.utilities import ImporterUI + +DICTCOLUMN1 = "ms_tavla" +DICTCOLUMN2 = "kod" +DICTCOLUMN3 = "teur" +ACCIDENT_TYPE_REGEX = re.compile(r"accidents_type_(?P\d)") +DICTIONARY_FILENAME = "Dictionary.csv" + +TABLES_DICT = { + 0: "columns_description", + 2: "road_type", + 3: "entrance_exit", + 4: "accident_severity", + 5: "accident_type", + 6: "road_alignment", + 7: "infrastructure_type", + 8: "road_geometry", + 10: "one_lane", + 11: "multi_lane", + 12: "speed_limit", + 13: "road_intactness", + 14: "road_width", + 16: "road_light", + 17: "road_control", + 18: "weather", + 19: "road_surface", + 20: "vehicle_purpose", + 21: "road_object", + 22: "object_distance", + 23: "didnt_cross", + 24: "cross_mode", + 25: "cross_location", + 26: "cross_direction", + 28: "driving_directions", + 29: "vehicle_damage", + 31: "involved_type", + 33: "safety_measures_use", # need to verify if exists in new format + 34: "safety_measures", + 35: "injury_severity", + 37: "day_type", + 38: "day_night", + 39: "day_in_week", + 40: "traffic_light", + 42: "engine_volume", + 43: "vehicle_attribution", + 44: "total_weight", + 45: "vehicle_type", + 46: "late_deceased", + 47: "location_accuracy", + 48: "vehicle_type", + 50: "injured_type", + 52: "injured_position", + 60: "accident_month", + 66: "population_type", + 67: "sex", + 68: "geo_area", + 77: "region", + 78: "municipal_status", + 79: "district", + 80: "natural_area", + 81: "yishuv_shape", + 92: "age_group", + 93: "accident_hour_raw", +} + + +def read_dictionary(dictionary_file): + from anyway.parsers.cbs.executor import read_cbs_file + + cbs_dictionary = defaultdict(dict) + dictionary = read_cbs_file(dictionary_file) + for _, dic in dictionary.iterrows(): + cbs_dictionary[int(dic[DICTCOLUMN1])][int(dic[DICTCOLUMN2])] = dic[DICTCOLUMN3] + return cbs_dictionary + + +def fill_dictionary_tables(cbs_dictionary, provider_code, year, should_commit=True): + if year < 2008: + return + for k, v in cbs_dictionary.items(): + if k == 27: + continue + try: + curr_table = TABLES_DICT[k] + except Exception as _: + logging.debug( + "A key " + str(k) + " was added to dictionary - update models, tables and classes" + ) + continue + for inner_k, inner_v in v.items(): + curr_table = TABLES_DICT[k] + if inner_v is None or (isinstance(inner_v, float) and math.isnan(inner_v)): + continue + sql_delete = ( + "DELETE FROM " + + curr_table + + " WHERE provider_code=" + + str(provider_code) + + " AND year=" + + str(year) + + " AND id=" + + str(inner_k) + ) + db.session.execute(sql_delete) + sql_insert = ( + "INSERT INTO " + + curr_table + + " VALUES (" + + str(inner_k) + + "," + + str(year) + + "," + + str(provider_code) + + "," + + "'" + + inner_v.replace("'", "") + + "'" + + ")" + + " ON CONFLICT DO NOTHING" + ) + db.session.execute(sql_insert) + try: + if should_commit: + db.session.commit() + except Exception as e: + logging.error(f"Error updating Dictionary tables: {e}") + raise e + logging.debug("Inserted/Updated dictionary values into table " + curr_table) + create_provider_code_table(should_commit) + + +def truncate_dictionary_tables(dictionary_file): + cbs_dictionary = read_dictionary(dictionary_file) + for k, _ in cbs_dictionary.items(): + if k == 97: + continue + curr_table = TABLES_DICT[k] + sql_truncate = "TRUNCATE TABLE " + curr_table + db.session.execute(sql_truncate) + db.session.commit() + logging.debug("Truncated table " + curr_table) + + +def create_provider_code_table(should_commit): + provider_code_table = "provider_code" + provider_code_class = ProviderCode + table_entries = db.session.query(provider_code_class) + table_entries.delete() + provider_code_dict = { + 1: "הלשכה המרכזית לסטטיסטיקה - סוג תיק 1", + 2: "איחוד הצלה", + 3: "הלשכה המרכזית לסטטיסטיקה - סוג תיק 3", + 4: "שומרי הדרך", + } + for k, v in provider_code_dict.items(): + sql_insert = ( + "INSERT INTO " + provider_code_table + " VALUES (" + str(k) + "," + "'" + v + "'" + ")" + ) + db.session.execute(sql_insert) + try: + if should_commit: + db.session.commit() + except Exception as e: + logging.error(f"Error updating table {provider_code_table}: {e}") + raise e + +def get_provider_code(directory_name=None): + if directory_name: + match = ACCIDENT_TYPE_REGEX.match(directory_name) + if match: + return int(match.groupdict()["type"]) + + ans = "" + while not ans.isdigit(): + ans = input("Directory provider code is invalid. Please enter a valid code: ") + if ans.isdigit(): + return int(ans) + + +def update_dictionary_tables(path): + import_ui = ImporterUI(path) + dir_name = import_ui.source_path() + dir_list = glob.glob("{0}/*/*".format(dir_name)) + + for directory in sorted(dir_list, reverse=True): + print(directory) + directory_name = os.path.basename(os.path.normpath(directory)) + year = directory_name[1:5] if directory_name[0] == "H" else directory_name[0:4] + if int(year) < 2008: + continue + parent_directory = os.path.basename(os.path.dirname(os.path.join(os.pardir, directory))) + provider_code = get_provider_code(parent_directory) + logging.debug("Importing Directory " + directory) + dictionary_file = _get_dictionary_file(directory) + if not dictionary_file: + return 0 + logging.debug("Filling dictionary for directory '{}'".format(directory)) + fill_dictionary_tables(read_dictionary(dictionary_file), provider_code, int(year)) + + +def _get_dictionary_file(directory): + files = [ + path + for path in os.listdir(directory) + if DICTIONARY_FILENAME.lower() in path.lower() and not path.startswith(".") + ] + if not files: + raise ValueError("Not found: '%s'" % DICTIONARY_FILENAME) + if len(files) > 1: + raise ValueError("Ambiguous: '%s'" % DICTIONARY_FILENAME) + return os.path.join(directory, files[0]) diff --git a/anyway/parsers/cbs/executor.py b/anyway/parsers/cbs/executor.py index 688ab757..fb1aa654 100644 --- a/anyway/parsers/cbs/executor.py +++ b/anyway/parsers/cbs/executor.py @@ -11,7 +11,7 @@ import math import pandas as pd -from sqlalchemy import or_, event +from sqlalchemy import or_ from typing import Dict, List from anyway.parsers.cbs import preprocessing_cbs_files @@ -522,9 +522,8 @@ def import_accidents(provider_code, accidents, streets, roads=None, non_urban_in marker = create_marker(provider_code, accident, streets, roads, non_urban_intersection) accidents_result.append(marker) db.session.bulk_insert_mappings(AccidentMarker, accidents_result) - db.session.commit() - logging.debug("Finished Importing markers") - logging.debug("Inserted " + str(len(accidents_result)) + " new accident markers") + logging.debug("Finished Processing markers") + logging.debug("Added " + str(len(accidents_result)) + " new accident markers to transaction") fill_db_geo_data() return len(accidents_result) @@ -585,8 +584,7 @@ def import_involved(provider_code, involved, **kwargs): } ) db.session.bulk_insert_mappings(Involved, involved_result) - db.session.commit() - logging.debug("Finished Importing involved") + logging.debug("Finished Processing involved") return len(involved_result) @@ -621,8 +619,7 @@ def import_vehicles(provider_code, vehicles, **kwargs): } ) db.session.bulk_insert_mappings(Vehicle, vehicles_result) - db.session.commit() - logging.debug("Finished Importing vehicles") + logging.debug("Finished Processing vehicles") return len(vehicles_result) @@ -703,7 +700,7 @@ def import_to_datastore(directory, provider_code, year, batch_size) -> int: # import dictionary with log_duration("Importing dictionary tables"): - fill_dictionary_tables(files_from_cbs[DICTIONARY], provider_code, year) + fill_dictionary_tables(files_from_cbs[DICTIONARY], provider_code, year, False) new_items = 0 with log_duration( @@ -714,7 +711,7 @@ def import_to_datastore(directory, provider_code, year, batch_size) -> int: ) logging.info( "Accident marker row counts: provider=%s year=%s " - "pandas_rows=%s committed_rows=%s", + "pandas_rows=%s inserted_rows=%s", provider_code, year, len(files_from_cbs[ACCIDENTS]), @@ -741,6 +738,7 @@ def import_to_datastore(directory, provider_code, year, batch_size) -> int: raise e +#not in use def delete_invalid_entries(batch_size): """ deletes all markers in the database with null latitude or longitude @@ -765,19 +763,16 @@ def delete_invalid_entries(batch_size): if q.all(): logging.debug("deleting invalid entries from Involved") q.delete(synchronize_session="fetch") - db.session.commit() q = db.session.query(Vehicle).filter(Vehicle.accident_id.in_(ids_chunk)) if q.all(): logging.debug("deleting invalid entries from Vehicle") q.delete(synchronize_session="fetch") - db.session.commit() q = db.session.query(AccidentMarker).filter(AccidentMarker.id.in_(ids_chunk)) if q.all(): logging.debug("deleting invalid entries from AccidentMarker") q.delete(synchronize_session="fetch") - db.session.commit() def delete_cbs_entries(start_year, batch_size): @@ -815,19 +810,16 @@ def delete_cbs_entries(start_year, batch_size): if q.all(): logging.debug("deleting entries from Involved") q.delete(synchronize_session=False) - db.session.commit() q = db.session.query(Vehicle).filter(Vehicle.accident_id.in_(ids_chunk)) if q.all(): logging.debug("deleting entries from Vehicle") q.delete(synchronize_session=False) - db.session.commit() q = db.session.query(AccidentMarker).filter(AccidentMarker.id.in_(ids_chunk)) if q.all(): logging.debug("deleting entries from AccidentMarker") q.delete(synchronize_session=False) - db.session.commit() def fill_db_geo_data(): @@ -839,7 +831,6 @@ def fill_db_geo_data(): "UPDATE markers SET geom = ST_SetSRID(ST_MakePoint(longitude,latitude),4326)\ WHERE geom IS NULL;" ) - db.session.commit() def get_provider_code(directory_name=None): @@ -855,82 +846,77 @@ def get_provider_code(directory_name=None): return int(ans) -def receive_rollback(conn, **kwargs): - """listen for the 'rollback' event""" - logging.debug(f"rollback in create_tables(). conn:{conn},kw:{kwargs}") - print("---------------------------------------------") - - -def create_tables(): +def create_tables(should_commit=True): chunk_size = 5000 try: - with db.get_engine().begin() as conn: - event.listen(conn, "rollback", receive_rollback) - with log_duration( - "Creating table '{}'".format(AccidentMarkerView.__tablename__) - ): - delete_all_rows_from_table(conn, AccidentMarkerView) - run_query_and_insert_to_table_in_chunks( - VIEWS.create_markers_hebrew_view(), - AccidentMarkerView, - AccidentMarker.id, - chunk_size, - conn, - ) - logging.debug("after insertion to markers_hebrew ") + conn = db.session.connection() + with log_duration( + "Creating table '{}'".format(AccidentMarkerView.__tablename__) + ): + delete_all_rows_from_table(conn, AccidentMarkerView) + run_query_and_insert_to_table_in_chunks( + VIEWS.create_markers_hebrew_view(), + AccidentMarkerView, + AccidentMarker.id, + chunk_size, + conn, + ) + logging.debug("after insertion to markers_hebrew ") - with log_duration( - "Creating table '{}'".format(InvolvedView.__tablename__) - ): - delete_all_rows_from_table(conn, InvolvedView) - run_query_and_insert_to_table_in_chunks( - VIEWS.create_involved_hebrew_view(), - InvolvedView, - Involved.id, - chunk_size, - conn, - ) - logging.debug("after insertion to involved_hebrew ") + with log_duration( + "Creating table '{}'".format(InvolvedView.__tablename__) + ): + delete_all_rows_from_table(conn, InvolvedView) + run_query_and_insert_to_table_in_chunks( + VIEWS.create_involved_hebrew_view(), + InvolvedView, + Involved.id, + chunk_size, + conn, + ) + logging.debug("after insertion to involved_hebrew ") - with log_duration( - "Creating table '{}'".format(VehiclesView.__tablename__) - ): - delete_all_rows_from_table(conn, VehiclesView) - run_query_and_insert_to_table_in_chunks( - VIEWS.create_vehicles_hebrew_view(), - VehiclesView, - Vehicle.id, - chunk_size, - conn, - ) - logging.debug("after insertion to vehicles_hebrew ") + with log_duration( + "Creating table '{}'".format(VehiclesView.__tablename__) + ): + delete_all_rows_from_table(conn, VehiclesView) + run_query_and_insert_to_table_in_chunks( + VIEWS.create_vehicles_hebrew_view(), + VehiclesView, + Vehicle.id, + chunk_size, + conn, + ) + logging.debug("after insertion to vehicles_hebrew ") - with log_duration( - "Creating table '{}'".format(VehicleMarkerView.__tablename__) - ): - delete_all_rows_from_table(conn, VehicleMarkerView) - run_query_and_insert_to_table_in_chunks( - VIEWS.create_vehicles_markers_hebrew_view(), - VehicleMarkerView, - VehiclesView.id, - chunk_size, - conn, - ) - logging.debug("after insertion to vehicles_markers_hebrew ") + with log_duration( + "Creating table '{}'".format(VehicleMarkerView.__tablename__) + ): + delete_all_rows_from_table(conn, VehicleMarkerView) + run_query_and_insert_to_table_in_chunks( + VIEWS.create_vehicles_markers_hebrew_view(), + VehicleMarkerView, + VehiclesView.id, + chunk_size, + conn, + ) + logging.debug("after insertion to vehicles_markers_hebrew ") - with log_duration( - "Creating table '{}'".format(InvolvedMarkerView.__tablename__) - ): - delete_all_rows_from_table(conn, InvolvedMarkerView) - run_query_and_insert_to_table_in_chunks( - VIEWS.create_involved_hebrew_markers_hebrew_view(), - InvolvedMarkerView, - InvolvedView.accident_id, - chunk_size, - conn, - ) - logging.debug("after insertion to involved_markers_hebrew") - logging.debug("Created DB Hebrew Tables") + with log_duration( + "Creating table '{}'".format(InvolvedMarkerView.__tablename__) + ): + delete_all_rows_from_table(conn, InvolvedMarkerView) + run_query_and_insert_to_table_in_chunks( + VIEWS.create_involved_hebrew_markers_hebrew_view(), + InvolvedMarkerView, + InvolvedView.accident_id, + chunk_size, + conn, + ) + logging.debug("after insertion to involved_markers_hebrew") + logging.debug("Created DB Hebrew Tables") + if should_commit: + db.session.commit() except Exception as e: logging.exception(f"Exception while creating hebrew tables, {e}", e) raise e @@ -964,8 +950,6 @@ def recreate_table_for_location_extraction(): AND (longitude is not null AND latitude is not null)) LOCATIONS)""" ) - db.session.commit() - def _validate_s3_files(s3_data_retriever, load_start_year, allow_missing): if ( @@ -1134,7 +1118,7 @@ def _import_from_local_dir(batch_size): dir_list = glob.glob("{0}/*/*".format(dir_name)) if import_ui.is_delete_all(): - truncate_tables(db, (Vehicle, Involved, AccidentMarker)) + truncate_tables(db, (Vehicle, Involved, AccidentMarker), commit=False) total = 0 for directory in sorted(dir_list, reverse=False): @@ -1164,14 +1148,14 @@ def _log_import_summary(total, started): def _build_hebrew_tables_and_derived_data(): fill_db_geo_data() - create_tables() + create_tables(should_commit=False) logging.debug("Finished Creating Hebrew DB Tables") with log_duration("Creating table 'cbs_locations'"): recreate_table_for_location_extraction() logging.debug("Finished Recreating tables for location extraction") logging.debug("Loading safety data tables") with log_duration("Importing safety data tables"): - sd_utils.load_data() + sd_utils.load_data(session=db.session) logging.debug("Completed load of safety data tables") @@ -1193,7 +1177,10 @@ def main(batch_size, source, load_start_year=None, allow_missing=False): _log_import_summary(total, started) _build_hebrew_tables_and_derived_data() + db.session.commit() + logging.debug("data committed successfully") except Exception as ex: + db.session.rollback() print("Traceback: {0}".format(traceback.format_exc())) raise CBSParsingFailed(message=str(ex)) # Todo - send an email that an exception occured diff --git a/anyway/utilities.py b/anyway/utilities.py index e5a16d4e..fb0d7341 100644 --- a/anyway/utilities.py +++ b/anyway/utilities.py @@ -1,397 +1,398 @@ -import argparse -import logging -import math -import os -import re -import sys -import threading -import time -import typing -from contextlib import contextmanager -from csv import DictReader -from datetime import datetime -from functools import partial -from urllib.parse import urlparse -from sqlalchemy import func, or_ -from sqlalchemy.sql import select - -import phonenumbers -from dateutil.relativedelta import relativedelta - -try: - from flask import Flask -except ModuleNotFoundError: - pass - -from phonenumbers import NumberParseException - -try: - from pyproj import Transformer -except ModuleNotFoundError: - pass - -try: - from validate_email import validate_email -except ModuleNotFoundError: - pass - -from anyway import config - -# Headless servers cannot use GUI file dialog and require raw user input -_fileDialogExist = True -try: - import tkFileDialog -except (ValueError, ImportError): - _fileDialogExist = False - -DATE_INPUT_FORMAT = "%d-%m-%Y" -_PROJECT_ROOT = os.path.join(os.path.dirname(__file__), "..") - - -@contextmanager -def log_duration(operation): - started = time.perf_counter() - status = "failed" - - try: - yield - status = "completed" - finally: - logging.info( - "%s %s in %.2f seconds", - operation, - status, - time.perf_counter() - started, - ) - - -def init_flask(): - """ - initializes a Flask instance with default values - """ - app = Flask( - "anyway", - template_folder=os.path.join(_PROJECT_ROOT, "templates"), - static_folder=os.path.join(_PROJECT_ROOT, "static"), - ) - app.config.from_object(config) - app.config["BABEL_TRANSLATION_DIRECTORIES"] = os.path.join(_PROJECT_ROOT, "translations") - if os.environ.get("PROXYFIX_X_FOR"): - from werkzeug.middleware.proxy_fix import ProxyFix - - app.wsgi_app = ProxyFix( - app.wsgi_app, - x_for=int(os.environ["PROXYFIX_X_FOR"]), - x_host=int(os.environ.get("PROXYFIX_X_HOST", "0")), - x_port=int(os.environ.get("PROXYFIX_X_PORT", "0")), - x_prefix=int(os.environ.get("PROXYFIX_X_PREFIX", "0")), - x_proto=int(os.environ.get("PROXYFIX_X_PROTO", "0")), - ) - return app - - -class ProgressSpinner(object): - def __init__(self): - self.counter = 0 - self.chars = ["|", "/", "-", "\\"] - - def show(self): - """ - prints a rotating spinner - """ - current_char = self.counter % len(self.chars) - sys.stderr.write("\r%s" % self.chars[current_char]) - self.counter += 1 - - -class CsvReader(object): - """ - loads and handles csv files - """ - - _digit_pattern = re.compile(r"^-?\d*(\.\d+)?$") - - def __init__(self, filename, encoding=None): - self._file = open(filename, encoding=encoding) - self._lock = threading.RLock() - self._closed = False - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - with self._lock: - if not self._closed: - self._file.close() - - def name(self): - """ - the filename of the csv file - :return: - """ - return self._file.name - - def close(self): - with self._lock: - if not self._closed: - self._file.close() - - def _convert(self, value): - """ - converts an str value to a typed one - """ - if value == "" or value is None: - return None - # the isdigit function doesn't match negative numbers - if CsvReader._digit_pattern.match(value): - return int(float(value)) - return value - - def __iter__(self): - for line in DictReader(self._file): - converted = dict([(key.upper(), self._convert(val)) for key, val in line.items()]) - yield converted - - -class ItmToWGS84(object): - def __init__(self): - # initializing WGS84 (epsg: 4326) and Israeli TM Grid (epsg: 2039) projections. - # for more info: https://epsg.io// - self.transformer = Transformer.from_proj(2039, 4326, always_xy=True) - - def convert(self, x, y): - """ - converts ITM to WGS84 coordinates - :type x: float - :type y: float - :rtype: tuple - :return: (longitude,latitude) - """ - longitude, latitude = self.transformer.transform(x, y) - return longitude, latitude - - -def time_delta(since): - delta = relativedelta(datetime.now(), since) - attrs = ["years", "months", "days", "hours", "minutes", "seconds"] - return " ".join( - "%d %s" % (getattr(delta, attr), getattr(delta, attr) > 1 and attr or attr[:-1]) - for attr in attrs - if getattr(delta, attr) - ) - - -def decode_hebrew(s): - return s - - -open_utf8 = partial(open, encoding="utf-8") - - -def row_to_dict(row): - return row._asdict() - - -def fetch_first_and_every_nth_value_for_column(conn, column_to_fetch, n): - sub_query = ( - select([]) - .column(column_to_fetch) - .column(func.row_number().over(order_by=column_to_fetch).label("row_number")) - .alias() - ) - select_query = select([sub_query]).where( - or_(func.mod(sub_query.c.row_number, n) == 0, sub_query.c.row_number == 1) - ) - ids_and_row_numbers = conn.execute(select_query).fetchall() - ids = [id_and_row_number[0] for id_and_row_number in ids_and_row_numbers] - return ids - - -def truncate_tables(db, tables): - logging.info("Deleting tables: " + ", ".join(table.__name__ for table in tables)) - for table in tables: - db.session.query(table).delete() - db.session.commit() - - -def delete_all_rows_from_table(conn, table): - table_name = table.__tablename__ - logging.info("Deleting all rows from table " + table_name) - conn.execute("DELETE FROM " + table_name) - - -def split_query_to_chunks_by_column(base_select, column_to_chunk_by, chunk_size, conn): - column_values = fetch_first_and_every_nth_value_for_column(conn, column_to_chunk_by, chunk_size) - for index in range(len(column_values)): - select = base_select.where(column_to_chunk_by >= column_values[index]) - if index + 1 < len(column_values): - select = select.where(column_to_chunk_by < column_values[index + 1]) - chunk = conn.execute(select).fetchall() - yield [dict(row.items()) for row in chunk] - logging.debug("after running query on all chunks") - - -def run_query_and_insert_to_table_in_chunks( - query, table_inserted_to, column_to_chunk_by, chunk_size, conn -): - for chunk in split_query_to_chunks_by_column(query, column_to_chunk_by, chunk_size, conn): - conn.execute(table_inserted_to.__table__.insert(), chunk) - - -def valid_date(date_string): - from datetime import datetime - - try: - return datetime.strptime(date_string, DATE_INPUT_FORMAT) - except ValueError: - msg = "Not a valid date: '{0}'. Date should be in the format DD-MM-YYYY".format(date_string) - raise argparse.ArgumentTypeError(msg) - - -class ImporterUI(object): - def __init__(self, source_path, specific_folder=False, delete_all=False): - self._specific_folder = specific_folder - self._delete_all = delete_all - self._source_path = os.path.abspath(source_path) - - def source_path(self): - if self._specific_folder: - if _fileDialogExist: - return tkFileDialog.askdirectory( - initialdir=self._source_path, title="Please select a directory" - ) - else: - return input("Please provide the directory path: ") - return self._source_path - - def is_delete_all(self): - if self._delete_all and self._specific_folder: - confirm_delete_all = input( - "Are you sure you want to delete all the current data? (y/n)\n" - ) - if confirm_delete_all.lower() == "n": - self._delete_all = False - return self._delete_all - - -def chunks(l, n): - """Yield successive n-sized chunks from l.""" - for i in range(0, len(l), n): - yield l[i : i + n] - - -def chunked_generator(input_generator, chunk_size): - chunk = [] - for item in input_generator: - chunk.append(item) - if len(chunk) == chunk_size: - yield chunk - chunk = [] - - if chunk: - yield chunk - - -def parse_age_from_range(age_range: int) -> typing.Optional[typing.Tuple[int, int]]: - # Convert from 'age_group' field in the table 'involved_markers_hebrew' to age range numbers - ret_age_code_to_age_range = { - 1: (0, 4), - 2: (5, 9), - 3: (10, 14), - 4: (15, 19), - 5: (20, 24), - 6: (25, 29), - 7: (30, 34), - 8: (35, 39), - 9: (40, 44), - 10: (45, 49), - 11: (50, 54), - 12: (55, 59), - 13: (60, 64), - 14: (65, 69), - 15: (70, 74), - 16: (75, 79), - 17: (80, 84), - 18: (85, 200), - 99: None, - } - return ret_age_code_to_age_range[age_range] - - -def is_valid_number(phone: str) -> bool: - try: - phone_obj = phonenumbers.parse(phone, "IL") - return phonenumbers.is_valid_number(phone_obj) - except NumberParseException: - return False - - -def is_a_safe_redirect_url(url: str) -> bool: - url_obj = urlparse(url) - if url_obj.scheme not in ["https", "http"]: - return False - - netloc = url_obj.netloc - if not netloc: - return False - - # Note that we don't support ipv6 localhost address or ipv4 localhost full range of address - if netloc in ["localhost", "127.0.0.1"]: - return True - else: # Check localhost with port - localhost_regex = re.compile(r"^127\.0\.0\.1:[0-9]{1,7}$|^localhost:[0-9]{1,7}$") - if localhost_regex.match(netloc): - return True - - if url_obj.scheme == "https" and netloc in [ - "www.anyway.co.il", - "anyway-infographics-staging.web.app", - "anyway-infographics.web.app", - "anyway-infographics-demo.web.app", - "media.anyway.co.il", - "www.safety-data.anyway.co.il", - "safety-data.anyway.co.il", - "safety-data.dfc2.anyway.co.il", - ]: - return True - - if ( - config.SERVER_ENV == "dev" - and url_obj.scheme == "https" - and netloc in ["dev.anyway.co.il", "www.dev.anyway.co.il"] - ): - return True - - return False - - -def is_a_valid_email(tmp_given_user_email: str) -> bool: - is_valid = validate_email( - email_address=tmp_given_user_email, check_regex=True, check_mx=False, use_blacklist=False - ) - return is_valid - - -def half_rounded_up(num: int): - return math.ceil(num / 2) - - -def trigger_airflow_dag(dag_id, conf=None): - import airflow_client.client - from airflow_client.client.api import dag_run_api - from airflow_client.client.model.dag_run import DAGRun - from anyway import secrets - - if conf is None: - conf = {} - airflow_api_url = "https://airflow.anyway.co.il/api/v1" - configuration = airflow_client.client.Configuration( - host=airflow_api_url, - username=secrets.get("AIRFLOW_USER"), - password=secrets.get("AIRFLOW_PASSWORD"), - ) - with airflow_client.client.ApiClient(configuration) as api_client: - dag_run_api_instance = dag_run_api.DAGRunApi(api_client) - dag_run = DAGRun(conf=conf) - return dag_run_api_instance.post_dag_run(dag_id, dag_run) - +import argparse +import logging +import math +import os +import re +import sys +import threading +import time +import typing +from contextlib import contextmanager +from csv import DictReader +from datetime import datetime +from functools import partial +from urllib.parse import urlparse +from sqlalchemy import func, or_ +from sqlalchemy.sql import select + +import phonenumbers +from dateutil.relativedelta import relativedelta + +try: + from flask import Flask +except ModuleNotFoundError: + pass + +from phonenumbers import NumberParseException + +try: + from pyproj import Transformer +except ModuleNotFoundError: + pass + +try: + from validate_email import validate_email +except ModuleNotFoundError: + pass + +from anyway import config + +# Headless servers cannot use GUI file dialog and require raw user input +_fileDialogExist = True +try: + import tkFileDialog +except (ValueError, ImportError): + _fileDialogExist = False + +DATE_INPUT_FORMAT = "%d-%m-%Y" +_PROJECT_ROOT = os.path.join(os.path.dirname(__file__), "..") + + +@contextmanager +def log_duration(operation): + started = time.perf_counter() + status = "failed" + + try: + yield + status = "completed" + finally: + logging.info( + "%s %s in %.2f seconds", + operation, + status, + time.perf_counter() - started, + ) + + +def init_flask(): + """ + initializes a Flask instance with default values + """ + app = Flask( + "anyway", + template_folder=os.path.join(_PROJECT_ROOT, "templates"), + static_folder=os.path.join(_PROJECT_ROOT, "static"), + ) + app.config.from_object(config) + app.config["BABEL_TRANSLATION_DIRECTORIES"] = os.path.join(_PROJECT_ROOT, "translations") + if os.environ.get("PROXYFIX_X_FOR"): + from werkzeug.middleware.proxy_fix import ProxyFix + + app.wsgi_app = ProxyFix( + app.wsgi_app, + x_for=int(os.environ["PROXYFIX_X_FOR"]), + x_host=int(os.environ.get("PROXYFIX_X_HOST", "0")), + x_port=int(os.environ.get("PROXYFIX_X_PORT", "0")), + x_prefix=int(os.environ.get("PROXYFIX_X_PREFIX", "0")), + x_proto=int(os.environ.get("PROXYFIX_X_PROTO", "0")), + ) + return app + + +class ProgressSpinner(object): + def __init__(self): + self.counter = 0 + self.chars = ["|", "/", "-", "\\"] + + def show(self): + """ + prints a rotating spinner + """ + current_char = self.counter % len(self.chars) + sys.stderr.write("\r%s" % self.chars[current_char]) + self.counter += 1 + + +class CsvReader(object): + """ + loads and handles csv files + """ + + _digit_pattern = re.compile(r"^-?\d*(\.\d+)?$") + + def __init__(self, filename, encoding=None): + self._file = open(filename, encoding=encoding) + self._lock = threading.RLock() + self._closed = False + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + with self._lock: + if not self._closed: + self._file.close() + + def name(self): + """ + the filename of the csv file + :return: + """ + return self._file.name + + def close(self): + with self._lock: + if not self._closed: + self._file.close() + + def _convert(self, value): + """ + converts an str value to a typed one + """ + if value == "" or value is None: + return None + # the isdigit function doesn't match negative numbers + if CsvReader._digit_pattern.match(value): + return int(float(value)) + return value + + def __iter__(self): + for line in DictReader(self._file): + converted = dict([(key.upper(), self._convert(val)) for key, val in line.items()]) + yield converted + + +class ItmToWGS84(object): + def __init__(self): + # initializing WGS84 (epsg: 4326) and Israeli TM Grid (epsg: 2039) projections. + # for more info: https://epsg.io// + self.transformer = Transformer.from_proj(2039, 4326, always_xy=True) + + def convert(self, x, y): + """ + converts ITM to WGS84 coordinates + :type x: float + :type y: float + :rtype: tuple + :return: (longitude,latitude) + """ + longitude, latitude = self.transformer.transform(x, y) + return longitude, latitude + + +def time_delta(since): + delta = relativedelta(datetime.now(), since) + attrs = ["years", "months", "days", "hours", "minutes", "seconds"] + return " ".join( + "%d %s" % (getattr(delta, attr), getattr(delta, attr) > 1 and attr or attr[:-1]) + for attr in attrs + if getattr(delta, attr) + ) + + +def decode_hebrew(s): + return s + + +open_utf8 = partial(open, encoding="utf-8") + + +def row_to_dict(row): + return row._asdict() + + +def fetch_first_and_every_nth_value_for_column(conn, column_to_fetch, n): + sub_query = ( + select([]) + .column(column_to_fetch) + .column(func.row_number().over(order_by=column_to_fetch).label("row_number")) + .alias() + ) + select_query = select([sub_query]).where( + or_(func.mod(sub_query.c.row_number, n) == 0, sub_query.c.row_number == 1) + ) + ids_and_row_numbers = conn.execute(select_query).fetchall() + ids = [id_and_row_number[0] for id_and_row_number in ids_and_row_numbers] + return ids + + +def truncate_tables(db, tables, commit=True): + logging.info("Deleting tables: " + ", ".join(table.__name__ for table in tables)) + for table in tables: + db.session.query(table).delete() + if commit: + db.session.commit() + + +def delete_all_rows_from_table(conn, table): + table_name = table.__tablename__ + logging.info("Deleting all rows from table " + table_name) + conn.execute("DELETE FROM " + table_name) + + +def split_query_to_chunks_by_column(base_select, column_to_chunk_by, chunk_size, conn): + column_values = fetch_first_and_every_nth_value_for_column(conn, column_to_chunk_by, chunk_size) + for index in range(len(column_values)): + select = base_select.where(column_to_chunk_by >= column_values[index]) + if index + 1 < len(column_values): + select = select.where(column_to_chunk_by < column_values[index + 1]) + chunk = conn.execute(select).fetchall() + yield [dict(row.items()) for row in chunk] + logging.debug("after running query on all chunks") + + +def run_query_and_insert_to_table_in_chunks( + query, table_inserted_to, column_to_chunk_by, chunk_size, conn +): + for chunk in split_query_to_chunks_by_column(query, column_to_chunk_by, chunk_size, conn): + conn.execute(table_inserted_to.__table__.insert(), chunk) + + +def valid_date(date_string): + from datetime import datetime + + try: + return datetime.strptime(date_string, DATE_INPUT_FORMAT) + except ValueError: + msg = "Not a valid date: '{0}'. Date should be in the format DD-MM-YYYY".format(date_string) + raise argparse.ArgumentTypeError(msg) + + +class ImporterUI(object): + def __init__(self, source_path, specific_folder=False, delete_all=False): + self._specific_folder = specific_folder + self._delete_all = delete_all + self._source_path = os.path.abspath(source_path) + + def source_path(self): + if self._specific_folder: + if _fileDialogExist: + return tkFileDialog.askdirectory( + initialdir=self._source_path, title="Please select a directory" + ) + else: + return input("Please provide the directory path: ") + return self._source_path + + def is_delete_all(self): + if self._delete_all and self._specific_folder: + confirm_delete_all = input( + "Are you sure you want to delete all the current data? (y/n)\n" + ) + if confirm_delete_all.lower() == "n": + self._delete_all = False + return self._delete_all + + +def chunks(l, n): + """Yield successive n-sized chunks from l.""" + for i in range(0, len(l), n): + yield l[i : i + n] + + +def chunked_generator(input_generator, chunk_size): + chunk = [] + for item in input_generator: + chunk.append(item) + if len(chunk) == chunk_size: + yield chunk + chunk = [] + + if chunk: + yield chunk + + +def parse_age_from_range(age_range: int) -> typing.Optional[typing.Tuple[int, int]]: + # Convert from 'age_group' field in the table 'involved_markers_hebrew' to age range numbers + ret_age_code_to_age_range = { + 1: (0, 4), + 2: (5, 9), + 3: (10, 14), + 4: (15, 19), + 5: (20, 24), + 6: (25, 29), + 7: (30, 34), + 8: (35, 39), + 9: (40, 44), + 10: (45, 49), + 11: (50, 54), + 12: (55, 59), + 13: (60, 64), + 14: (65, 69), + 15: (70, 74), + 16: (75, 79), + 17: (80, 84), + 18: (85, 200), + 99: None, + } + return ret_age_code_to_age_range[age_range] + + +def is_valid_number(phone: str) -> bool: + try: + phone_obj = phonenumbers.parse(phone, "IL") + return phonenumbers.is_valid_number(phone_obj) + except NumberParseException: + return False + + +def is_a_safe_redirect_url(url: str) -> bool: + url_obj = urlparse(url) + if url_obj.scheme not in ["https", "http"]: + return False + + netloc = url_obj.netloc + if not netloc: + return False + + # Note that we don't support ipv6 localhost address or ipv4 localhost full range of address + if netloc in ["localhost", "127.0.0.1"]: + return True + else: # Check localhost with port + localhost_regex = re.compile(r"^127\.0\.0\.1:[0-9]{1,7}$|^localhost:[0-9]{1,7}$") + if localhost_regex.match(netloc): + return True + + if url_obj.scheme == "https" and netloc in [ + "www.anyway.co.il", + "anyway-infographics-staging.web.app", + "anyway-infographics.web.app", + "anyway-infographics-demo.web.app", + "media.anyway.co.il", + "www.safety-data.anyway.co.il", + "safety-data.anyway.co.il", + "safety-data.dfc2.anyway.co.il", + ]: + return True + + if ( + config.SERVER_ENV == "dev" + and url_obj.scheme == "https" + and netloc in ["dev.anyway.co.il", "www.dev.anyway.co.il"] + ): + return True + + return False + + +def is_a_valid_email(tmp_given_user_email: str) -> bool: + is_valid = validate_email( + email_address=tmp_given_user_email, check_regex=True, check_mx=False, use_blacklist=False + ) + return is_valid + + +def half_rounded_up(num: int): + return math.ceil(num / 2) + + +def trigger_airflow_dag(dag_id, conf=None): + import airflow_client.client + from airflow_client.client.api import dag_run_api + from airflow_client.client.model.dag_run import DAGRun + from anyway import secrets + + if conf is None: + conf = {} + airflow_api_url = "https://airflow.anyway.co.il/api/v1" + configuration = airflow_client.client.Configuration( + host=airflow_api_url, + username=secrets.get("AIRFLOW_USER"), + password=secrets.get("AIRFLOW_PASSWORD"), + ) + with airflow_client.client.ApiClient(configuration) as api_client: + dag_run_api_instance = dag_run_api.DAGRunApi(api_client) + dag_run = DAGRun(conf=conf) + return dag_run_api_instance.post_dag_run(dag_id, dag_run) + diff --git a/anyway/views/safety_data/sd_utils.py b/anyway/views/safety_data/sd_utils.py index 32c5e70f..3796b77c 100644 --- a/anyway/views/safety_data/sd_utils.py +++ b/anyway/views/safety_data/sd_utils.py @@ -1,213 +1,223 @@ -import json -import logging -from typing import Iterable, Dict, Any, List -from sqlalchemy.orm import sessionmaker, Session -from sqlalchemy import and_, func -from flask import request, Response -from anyway.models import ( - Involved, - SDAccident, - SDInvolved, - AccidentMarkerView, - Involved, -) -from anyway.app_and_db import db -from anyway.utilities import chunked_generator - -GEO_PARAM = "geo" - - -def load_data(): - conn = db.get_engine().connect() - trans = conn.begin() - sess = sessionmaker()(bind=conn) - try: - sess.query(SDInvolved).delete() - sess.query(SDAccident).delete() - sd_load_accident(sess) - sd_load_involved(sess) - trans.commit() - return Response(json.dumps("Tables loaded", default=str), mimetype="application/json") - except Exception as e: - trans.rollback() - logging.exception("Error loading data: %s", e) - finally: - sess.close() - conn.close() - - -def sd_load_involved(sess: Session): - for chunk in chunked_generator(get_involved_data(sess), 4069): - sess.execute(SDInvolved.__table__.insert(), chunk) - - -def get_involved_data(sess: Session): - for d in ( - sess.query(Involved, SDAccident) - .join( - SDAccident, - and_( - SDAccident.provider_code == Involved.provider_code, - SDAccident.accident_id == Involved.accident_id, - SDAccident.accident_year == Involved.accident_year, - ), - ) - .with_entities( - Involved.id, - Involved.accident_id, - Involved.accident_year, - Involved.provider_code, - Involved.age_group, - Involved.injured_type, - Involved.injury_severity, - Involved.population_type, - Involved.sex, - Involved.vehicle_type, - ) - ): - yield { - "_id": d.id, - "accident_id": d.accident_id, - "accident_year": d.accident_year, - "provider_code": d.provider_code, - "age_group": d.age_group, - "injured_type": d.injured_type, - "injury_severity": d.injury_severity, - "population_type": d.population_type, - "sex": d.sex, - "vehicle_type": d.vehicle_type, - } - - -def sd_load_accident(sess: Session): - sd_load_accident_main(sess) - set_vehicles_in_sd_acc_table(sess) - set_geom_in_sd_acc_table(sess) - - -def sd_load_accident_main(sess: Session): - for chunk in chunked_generator(sd_get_accident_data(sess), 1024): - sess.execute(SDAccident.__table__.insert(), chunk) - - -def sd_get_accident_data(sess: Session) -> Iterable[Dict[str, Any]]: - return ( - { - "accident_id": d.id, - "accident_year": d.accident_year, - "provider_code": d.provider_code, - "accident_month": d.accident_month, - "accident_timestamp": d.accident_timestamp, - "accident_type": d.accident_type, - "accident_yishuv_symbol": d.yishuv_symbol, - "day_in_week": d.day_in_week, - "day_night": d.day_night, - "location_accuracy": d.location_accuracy, - "multi_lane": d.multi_lane, - "one_lane": d.one_lane, - "road1": d.road1, - "road2": d.road2, - "road_segment_id": d.road_segment_id, - "road_type": d.road_type, - "road_width": d.road_width, - "speed_limit": d.speed_limit, - "street1": d.street1, - "street2": d.street2, - "latitude": d.latitude, - "longitude": d.longitude, - } - for d in sess.query(AccidentMarkerView).with_entities( - AccidentMarkerView.id, - AccidentMarkerView.accident_year, - AccidentMarkerView.provider_code, - AccidentMarkerView.accident_month, - AccidentMarkerView.accident_timestamp, - AccidentMarkerView.accident_type, - AccidentMarkerView.yishuv_symbol, - AccidentMarkerView.day_in_week, - AccidentMarkerView.day_night, - AccidentMarkerView.location_accuracy, - AccidentMarkerView.multi_lane, - AccidentMarkerView.one_lane, - AccidentMarkerView.road1, - AccidentMarkerView.road2, - AccidentMarkerView.road_segment_id, - AccidentMarkerView.road_type, - AccidentMarkerView.road_width, - AccidentMarkerView.speed_limit, - AccidentMarkerView.street1, - AccidentMarkerView.street2, - AccidentMarkerView.latitude, - AccidentMarkerView.longitude, - ) - ) - - -def set_vehicles_in_sd_acc_table(sess: Session): - """ - vehicles is a bitmap of vehicle types involved in the accident - """ - sess.execute( - """ - UPDATE safety_data_accident - SET vehicles=subquery.vt_bitmap - FROM - (SELECT accident_id, - provider_code, - accident_year, - bit_or(vt_power2) AS vt_bitmap - FROM - (SELECT DISTINCT vehicles.accident_id, - vehicles.provider_code, - vehicles.accident_year, - 1::bigint << vehicle_type AS vt_power2 - FROM vehicles - LEFT JOIN vehicle_type ON vehicles.vehicle_type = vehicle_type.id - AND vehicles.accident_year = vehicle_type.year - AND vehicles.provider_code = vehicle_type.provider_code) IVT - GROUP BY accident_id, - provider_code, - accident_year) AS subquery - WHERE safety_data_accident.accident_id=subquery.accident_id - AND safety_data_accident.accident_year=subquery.accident_year - AND safety_data_accident.provider_code=subquery.provider_code - """ - ) - - -def set_geom_in_sd_acc_table(sess: Session): - """ - geom is a PostGIS point built from longitude/latitude, used for spatial queries. - """ - sess.query(SDAccident).filter( - SDAccident.longitude.isnot(None), - SDAccident.latitude.isnot(None), - ).update( - { - SDAccident.geom: func.ST_SetSRID( - func.ST_MakePoint(SDAccident.longitude, SDAccident.latitude), 4326 - ) - }, - synchronize_session=False, - ) - - -def get_params() -> dict: - def f(v: List[str]) -> List[str]: - res = [] - [res.extend(x.split(",")) for x in v] - return res - - params = request.values - vals = {k: f(params.getlist(key=k)) for k in params.keys() if k != GEO_PARAM} - - if GEO_PARAM in request.values: - vals[GEO_PARAM] = request.values.getlist(key=GEO_PARAM) - - if request.is_json: - body = request.get_json(silent=True) - if isinstance(body, dict) and GEO_PARAM in body: - geo_val = body[GEO_PARAM] - vals[GEO_PARAM] = [json.dumps(geo_val) if not isinstance(geo_val, str) else geo_val] - - return vals +import json +import logging +from typing import Iterable, Dict, Any, List +from sqlalchemy.orm import sessionmaker, Session +from sqlalchemy import and_, func +from flask import request, Response +from anyway.models import ( + Involved, + SDAccident, + SDInvolved, + AccidentMarkerView, + Involved, +) +from anyway.app_and_db import db +from anyway.utilities import chunked_generator + +GEO_PARAM = "geo" + + +def load_data(session=None): + own_resources = session is None + conn = None + trans = None + if own_resources: + conn = db.get_engine().connect() + trans = conn.begin() + sess = sessionmaker()(bind=conn) + else: + sess = session + try: + sess.query(SDInvolved).delete() + sess.query(SDAccident).delete() + sd_load_accident(sess) + sd_load_involved(sess) + if own_resources: + trans.commit() + return Response(json.dumps("Tables loaded", default=str), mimetype="application/json") + except Exception as e: + if own_resources: + trans.rollback() + logging.exception("Error loading data: %s", e) + raise + finally: + if own_resources: + sess.close() + conn.close() + + +def sd_load_involved(sess: Session): + for chunk in chunked_generator(get_involved_data(sess), 4069): + sess.execute(SDInvolved.__table__.insert(), chunk) + + +def get_involved_data(sess: Session): + for d in ( + sess.query(Involved, SDAccident) + .join( + SDAccident, + and_( + SDAccident.provider_code == Involved.provider_code, + SDAccident.accident_id == Involved.accident_id, + SDAccident.accident_year == Involved.accident_year, + ), + ) + .with_entities( + Involved.id, + Involved.accident_id, + Involved.accident_year, + Involved.provider_code, + Involved.age_group, + Involved.injured_type, + Involved.injury_severity, + Involved.population_type, + Involved.sex, + Involved.vehicle_type, + ) + ): + yield { + "_id": d.id, + "accident_id": d.accident_id, + "accident_year": d.accident_year, + "provider_code": d.provider_code, + "age_group": d.age_group, + "injured_type": d.injured_type, + "injury_severity": d.injury_severity, + "population_type": d.population_type, + "sex": d.sex, + "vehicle_type": d.vehicle_type, + } + + +def sd_load_accident(sess: Session): + sd_load_accident_main(sess) + set_vehicles_in_sd_acc_table(sess) + set_geom_in_sd_acc_table(sess) + + +def sd_load_accident_main(sess: Session): + for chunk in chunked_generator(sd_get_accident_data(sess), 1024): + sess.execute(SDAccident.__table__.insert(), chunk) + + +def sd_get_accident_data(sess: Session) -> Iterable[Dict[str, Any]]: + return ( + { + "accident_id": d.id, + "accident_year": d.accident_year, + "provider_code": d.provider_code, + "accident_month": d.accident_month, + "accident_timestamp": d.accident_timestamp, + "accident_type": d.accident_type, + "accident_yishuv_symbol": d.yishuv_symbol, + "day_in_week": d.day_in_week, + "day_night": d.day_night, + "location_accuracy": d.location_accuracy, + "multi_lane": d.multi_lane, + "one_lane": d.one_lane, + "road1": d.road1, + "road2": d.road2, + "road_segment_id": d.road_segment_id, + "road_type": d.road_type, + "road_width": d.road_width, + "speed_limit": d.speed_limit, + "street1": d.street1, + "street2": d.street2, + "latitude": d.latitude, + "longitude": d.longitude, + } + for d in sess.query(AccidentMarkerView).with_entities( + AccidentMarkerView.id, + AccidentMarkerView.accident_year, + AccidentMarkerView.provider_code, + AccidentMarkerView.accident_month, + AccidentMarkerView.accident_timestamp, + AccidentMarkerView.accident_type, + AccidentMarkerView.yishuv_symbol, + AccidentMarkerView.day_in_week, + AccidentMarkerView.day_night, + AccidentMarkerView.location_accuracy, + AccidentMarkerView.multi_lane, + AccidentMarkerView.one_lane, + AccidentMarkerView.road1, + AccidentMarkerView.road2, + AccidentMarkerView.road_segment_id, + AccidentMarkerView.road_type, + AccidentMarkerView.road_width, + AccidentMarkerView.speed_limit, + AccidentMarkerView.street1, + AccidentMarkerView.street2, + AccidentMarkerView.latitude, + AccidentMarkerView.longitude, + ) + ) + + +def set_vehicles_in_sd_acc_table(sess: Session): + """ + vehicles is a bitmap of vehicle types involved in the accident + """ + sess.execute( + """ + UPDATE safety_data_accident + SET vehicles=subquery.vt_bitmap + FROM + (SELECT accident_id, + provider_code, + accident_year, + bit_or(vt_power2) AS vt_bitmap + FROM + (SELECT DISTINCT vehicles.accident_id, + vehicles.provider_code, + vehicles.accident_year, + 1::bigint << vehicle_type AS vt_power2 + FROM vehicles + LEFT JOIN vehicle_type ON vehicles.vehicle_type = vehicle_type.id + AND vehicles.accident_year = vehicle_type.year + AND vehicles.provider_code = vehicle_type.provider_code) IVT + GROUP BY accident_id, + provider_code, + accident_year) AS subquery + WHERE safety_data_accident.accident_id=subquery.accident_id + AND safety_data_accident.accident_year=subquery.accident_year + AND safety_data_accident.provider_code=subquery.provider_code + """ + ) + + +def set_geom_in_sd_acc_table(sess: Session): + """ + geom is a PostGIS point built from longitude/latitude, used for spatial queries. + """ + sess.query(SDAccident).filter( + SDAccident.longitude.isnot(None), + SDAccident.latitude.isnot(None), + ).update( + { + SDAccident.geom: func.ST_SetSRID( + func.ST_MakePoint(SDAccident.longitude, SDAccident.latitude), 4326 + ) + }, + synchronize_session=False, + ) + + +def get_params() -> dict: + def f(v: List[str]) -> List[str]: + res = [] + [res.extend(x.split(",")) for x in v] + return res + + params = request.values + vals = {k: f(params.getlist(key=k)) for k in params.keys() if k != GEO_PARAM} + + if GEO_PARAM in request.values: + vals[GEO_PARAM] = request.values.getlist(key=GEO_PARAM) + + if request.is_json: + body = request.get_json(silent=True) + if isinstance(body, dict) and GEO_PARAM in body: + geo_val = body[GEO_PARAM] + vals[GEO_PARAM] = [json.dumps(geo_val) if not isinstance(geo_val, str) else geo_val] + + return vals diff --git a/tests/parsers/cbs/test_executor.py b/tests/parsers/cbs/test_executor.py index 14ca3377..54c37f69 100644 --- a/tests/parsers/cbs/test_executor.py +++ b/tests/parsers/cbs/test_executor.py @@ -1,37 +1,285 @@ -from unittest.mock import MagicMock - -import pytest - -from anyway.parsers.cbs.exceptions import CBSParsingFailed -from anyway.parsers.cbs.executor import main - -@pytest.fixture -def mock_s3_data_retriever(monkeypatch): - monkeypatch.setattr('anyway.parsers.cbs.executor.S3DataRetriever', MagicMock()) - -@pytest.fixture -def mock_shutil(monkeypatch): - monkeypatch.setattr('anyway.parsers.cbs.executor.shutil', MagicMock()) - -def test_import_streets_is_called_once_when_source_is_s3(monkeypatch, mock_s3_data_retriever, mock_shutil): - # Arrange - delete_cbs_entries = MagicMock() - monkeypatch.setattr('anyway.parsers.cbs.executor.delete_cbs_entries', delete_cbs_entries) - monkeypatch.setattr('anyway.parsers.cbs.executor.fill_db_geo_data', MagicMock()) - monkeypatch.setattr('anyway.parsers.cbs.executor.create_tables', MagicMock()) - - # Act - main(batch_size=MagicMock(), source='s3') - - # Assert - delete_cbs_entries.assert_called_once() - - -@pytest.mark.skip(reason="Test should be improved when improving testing for cbs pipiline") -def test_cbs_parsing_failed_is_raised_when_something_bad_happens(monkeypatch): - monkeypatch.setattr('anyway.parsers.cbs.executor.create_tables', - MagicMock(side_effect=Exception('something bad'))) - - with pytest.raises(CBSParsingFailed, match='Exception occurred while loading the cbs data: something bad'): - main(batch_size=MagicMock(), source=MagicMock()) - +from pathlib import Path +from shutil import copyfile +from unittest.mock import MagicMock + +import pandas as pd +import pytest + +from anyway.app_and_db import app, db +from anyway.models import ( + AccidentMarker, + AccidentMarkerView, + CBSLocations, + Involved, + InvolvedMarkerView, + InvolvedView, + SDAccident, + SDInvolved, + Vehicle, + VehicleMarkerView, + VehiclesView, +) +from anyway.parsers.cbs import executor +from anyway.parsers.cbs.exceptions import CBSParsingFailed + +TEMPLATE_CBS_DIRECTORY = Path("static/data/cbs/accidents_type_1/H20191111") +TEMPLATE_BATCH_NAME = "H20191111" +TEMPLATE_ACCIDENT_ID = 2019000069 +REPLACEMENT_ACCIDENT_ID = 1 +REPLACEMENT_PROVIDER_CODES = (1, 3) +ACCIDENT_YEAR = 2019 +ACCIDENT_COLUMN = "TeunaID_FKT" +PROVIDER_COLUMN = "SemelSugTikLMS" +FAILURE_MESSAGE = "failure before commit" + +RAW_CBS_MODELS = ( + AccidentMarker, + Vehicle, + Involved, +) + +HEBREW_MODELS = ( + AccidentMarkerView, + VehiclesView, + InvolvedView, + VehicleMarkerView, + InvolvedMarkerView, +) + +DERIVED_MODELS = ( + CBSLocations, + SDAccident, + SDInvolved, +) + +AFFECTED_MODELS = ( + *RAW_CBS_MODELS, + *HEBREW_MODELS, + *DERIVED_MODELS, +) + + +def snapshot_table(model): + columns = list(model.__table__.columns) + primary_key = list(model.__table__.primary_key.columns) + rows = db.session.query(*columns).order_by(*primary_key).all() + return [ + tuple(None if value is None else str(value) for value in row) + for row in rows + ] + + +def snapshot_affected_tables(): + return {model.__tablename__: snapshot_table(model) for model in AFFECTED_MODELS} + + +def assert_preloaded_data_exists(): + assert ( + AccidentMarker.query.filter( + AccidentMarker.id == TEMPLATE_ACCIDENT_ID, + AccidentMarker.provider_code == 1, + AccidentMarker.accident_year == ACCIDENT_YEAR, + ).count() + == 1 + ) + + assert ( + Vehicle.query.filter( + Vehicle.accident_id == TEMPLATE_ACCIDENT_ID, + Vehicle.provider_code == 1, + Vehicle.accident_year == ACCIDENT_YEAR, + ).count() + > 0 + ) + + assert ( + Involved.query.filter( + Involved.accident_id == TEMPLATE_ACCIDENT_ID, + Involved.provider_code == 1, + Involved.accident_year == ACCIDENT_YEAR, + ).count() + > 0 + ) + + +def find_template_file(suffix): + matches = list(TEMPLATE_CBS_DIRECTORY.glob(f"*{suffix}")) + assert len(matches) == 1, f"Expected one {suffix} file in {TEMPLATE_CBS_DIRECTORY}" + return matches[0] + + +def select_template_row(dataframe): + rows = dataframe.loc[dataframe[ACCIDENT_COLUMN] == TEMPLATE_ACCIDENT_ID].copy() + assert not rows.empty, ( + f"Accident {TEMPLATE_ACCIDENT_ID} is missing from the CBS fixture" + ) + return rows.iloc[[0]] + + +def rewrite_as_replacement_row(row, provider_code): + row = row.copy() + row[ACCIDENT_COLUMN] = REPLACEMENT_ACCIDENT_ID + row[PROVIDER_COLUMN] = provider_code + return row + + +def write_minimal_cbs_table(suffix, target_dir, provider_code): + dataframe = pd.read_csv( + find_template_file(suffix), encoding=executor.CONTENT_ENCODING + ) + row = rewrite_as_replacement_row(select_template_row(dataframe), provider_code) + row.to_csv( + target_dir / f"{TEMPLATE_BATCH_NAME}{suffix}", + index=False, + encoding=executor.CONTENT_ENCODING, + ) + + +def copy_required_lookup_files(target_dir): + for suffix in ("Dictionary.csv", "DicStreets.csv"): + copyfile( + find_template_file(suffix), + target_dir / f"{TEMPLATE_BATCH_NAME}{suffix}", + ) + + +def create_replacement_cbs_directory(tmp_path): + """Create replacement input from the template CBS fixture. + + Select one accident and one related vehicle/involved row, change the + accident ID, and duplicate the resulting dataset for providers 1 and 3. + """ + root = tmp_path / "cbs" + for provider_code in REPLACEMENT_PROVIDER_CODES: + target = root / f"accidents_type_{provider_code}" / TEMPLATE_BATCH_NAME + target.mkdir(parents=True) + for suffix in ("AccData.csv", "VehData.csv", "InvData.csv"): + write_minimal_cbs_table(suffix, target, provider_code) + copy_required_lookup_files(target) + return root + + +def configure_replacement_import(monkeypatch, replacement_directory): + monkeypatch.setattr( + executor.ImporterUI, + "source_path", + lambda self: str(replacement_directory), + ) + monkeypatch.setattr( + executor.ImporterUI, + "is_delete_all", + lambda self: True, + ) + + +def count_replacement_rows(model, accident_id_column): + return ( + model.query.filter( + accident_id_column == REPLACEMENT_ACCIDENT_ID, + model.provider_code.in_(REPLACEMENT_PROVIDER_CODES), + model.accident_year == ACCIDENT_YEAR, + ).count() + ) + + +def assert_replacement_reached_raw_and_safety_data_tables(): + expected_rows = len(REPLACEMENT_PROVIDER_CODES) + + # Raw CBS import completed for both providers. + assert count_replacement_rows(AccidentMarker, AccidentMarker.id) == expected_rows + assert count_replacement_rows(Vehicle, Vehicle.accident_id) == expected_rows + assert count_replacement_rows(Involved, Involved.accident_id) == expected_rows + + # Derived safety-data generation also completed. + assert count_replacement_rows(SDAccident, SDAccident.accident_id) == expected_rows + assert count_replacement_rows(SDInvolved, SDInvolved.accident_id) == expected_rows + + +def fail_when_pipeline_attempts_to_commit(monkeypatch): + def fail_commit(): + assert_replacement_reached_raw_and_safety_data_tables() + raise RuntimeError(FAILURE_MESSAGE) + + monkeypatch.setattr(db.session, "commit", fail_commit) + + +def assert_tables_unchanged(before, after): + for table_name, original_rows in before.items(): + assert after[table_name] == original_rows, ( + f"{table_name} changed despite rollback" + ) + + +@pytest.fixture +def mock_s3_data_retriever(monkeypatch): + monkeypatch.setattr("anyway.parsers.cbs.executor.S3DataRetriever", MagicMock()) + + +@pytest.fixture +def mock_shutil(monkeypatch): + monkeypatch.setattr("anyway.parsers.cbs.executor.shutil", MagicMock()) + + +def test_import_streets_is_called_once_when_source_is_s3( + monkeypatch, mock_s3_data_retriever, mock_shutil +): + # Arrange + delete_cbs_entries = MagicMock() + monkeypatch.setattr("anyway.parsers.cbs.executor.delete_cbs_entries", delete_cbs_entries) + monkeypatch.setattr("anyway.parsers.cbs.executor.fill_db_geo_data", MagicMock()) + monkeypatch.setattr("anyway.parsers.cbs.executor.create_tables", MagicMock()) + monkeypatch.setattr( + "anyway.parsers.cbs.executor.recreate_table_for_location_extraction", MagicMock() + ) + monkeypatch.setattr("anyway.parsers.cbs.executor.sd_utils.load_data", MagicMock()) + monkeypatch.setattr("anyway.parsers.cbs.executor.db.session.commit", MagicMock()) + + # Act + executor.main(batch_size=MagicMock(), source="s3") + + # Assert + delete_cbs_entries.assert_called_once() + + +def test_cbs_parsing_failed_is_raised_when_something_bad_happens(monkeypatch): + monkeypatch.setattr( + "anyway.parsers.cbs.executor._import_from_s3", MagicMock(return_value=0) + ) + monkeypatch.setattr("anyway.parsers.cbs.executor.fill_db_geo_data", MagicMock()) + monkeypatch.setattr( + "anyway.parsers.cbs.executor.create_tables", + MagicMock(side_effect=Exception("something bad")), + ) + rollback = MagicMock() + monkeypatch.setattr("anyway.parsers.cbs.executor.db.session.rollback", rollback) + + with pytest.raises( + CBSParsingFailed, + match="Exception occurred while loading the cbs data: something bad", + ): + executor.main(batch_size=MagicMock(), source="s3") + + rollback.assert_called_once() + + +#The test asserts data exists before import, creates replacement input, +#asserts data was imported, triggers rollback before final commit, +#and asserts final snapshot matches original snapshot. +@pytest.mark.partial_db +def test_failed_cbs_import_preserves_existing_data(monkeypatch, tmp_path): + """A failed CBS replacement import must preserve the existing database.""" + + with app.app_context(): + assert_preloaded_data_exists() + original_state = snapshot_affected_tables() + + replacement_directory = create_replacement_cbs_directory(tmp_path) + configure_replacement_import(monkeypatch, replacement_directory) + fail_when_pipeline_attempts_to_commit(monkeypatch) + + with pytest.raises(CBSParsingFailed, match=FAILURE_MESSAGE): + executor.main(batch_size=100, source="local_dir_for_tests_only") + + db.session.expire_all() + state_after_rollback = snapshot_affected_tables() + assert_tables_unchanged(before=original_state, after=state_after_rollback)