From aa5b5416bb4503b15041075ff89d2f09943da63c Mon Sep 17 00:00:00 2001 From: Benjamin K <53038537+treee111@users.noreply.github.com> Date: Mon, 3 Apr 2023 10:03:11 +0200 Subject: [PATCH] [DEV] refactor osm_maps_functions.py like geofabrik.py - using a interface and two implementing classes (#194) refactor osm_maps_functions.py like geofabrik.py --- tests/test_osm_maps.py | 20 +-- wahoomc/main.py | 11 +- wahoomc/osm_maps_functions.py | 234 +++++++++++++++++++++------------- 3 files changed, 166 insertions(+), 99 deletions(-) diff --git a/tests/test_osm_maps.py b/tests/test_osm_maps.py index d6d1e888..eb675be5 100644 --- a/tests/test_osm_maps.py +++ b/tests/test_osm_maps.py @@ -6,7 +6,7 @@ import unittest # import custom python packages -from wahoomc.osm_maps_functions import OsmData +from wahoomc.osm_maps_functions import CountryOsmData, XYOsmData from wahoomc.osm_maps_functions import OsmMaps from wahoomc.osm_maps_functions import get_xy_coordinates_from_input # from wahoomc.osm_maps_functions import TileNotFoundError @@ -106,14 +106,16 @@ def process_and_check_border_countries(self, inp_val, calc_border_c, exp_result, """ o_input_data = InputData() + o_input_data.process_border_countries = calc_border_c + if inp_mode == 'country': o_input_data.country = inp_val + o_osm_data = CountryOsmData(o_input_data) elif inp_mode == 'xy_coordinate': o_input_data.xy_coordinates = inp_val - o_input_data.process_border_countries = calc_border_c + o_osm_data = XYOsmData(o_input_data) - o_osm_data = OsmData() - o_osm_data.process_input_of_the_tool(o_input_data) + o_osm_data.process_input_of_the_tool() result = o_osm_data.border_countries @@ -138,8 +140,8 @@ def test_input_country_malta(self): o_input_data = InputData() o_input_data.country = 'malta' - o_osm_data = OsmData() - o_osm_data.process_input_of_the_tool(o_input_data) + o_osm_data = CountryOsmData(o_input_data) + o_osm_data.process_input_of_the_tool() result = o_osm_data.country_name self.assertEqual(result, 'malta') @@ -211,8 +213,10 @@ def test_version_and_tags_of_country_config_file(self): # prevent from downloading land_polygons each time o_input_data.max_days_old = 1000 - o_osm_data = OsmData() - o_downloader = o_osm_data.process_input_of_the_tool(o_input_data) + o_osm_data = CountryOsmData(o_input_data) + o_osm_data.process_input_of_the_tool() + + o_downloader = o_osm_data.get_downloader() # download files marked for download to fill up map_file per country to write to config o_downloader.download_files_if_needed() diff --git a/wahoomc/main.py b/wahoomc/main.py index bb6171a4..1ccf6283 100644 --- a/wahoomc/main.py +++ b/wahoomc/main.py @@ -15,7 +15,7 @@ from wahoomc.downloader import download_tooling from wahoomc.osm_maps_functions import OsmMaps -from wahoomc.osm_maps_functions import OsmData +from wahoomc.osm_maps_functions import CountryOsmData, XYOsmData # logging used in the terminal output: # # means top-level command @@ -55,9 +55,14 @@ def run(run_level): # Is there something to do? o_input_data.is_required_input_given_or_exit(issue_message=True) - o_osm_data = OsmData() + if o_input_data.country: + o_osm_data = CountryOsmData(o_input_data) + elif o_input_data.xy_coordinates: + o_osm_data = XYOsmData(o_input_data) + # Check for not existing or expired files. Mark for download, if dl is needed - o_downloader = o_osm_data.process_input_of_the_tool(o_input_data) + o_osm_data.process_input_of_the_tool() + o_downloader = o_osm_data.get_downloader() # Download files marked for download o_downloader.download_files_if_needed() diff --git a/wahoomc/osm_maps_functions.py b/wahoomc/osm_maps_functions.py index c9eb21c5..5ad12916 100644 --- a/wahoomc/osm_maps_functions.py +++ b/wahoomc/osm_maps_functions.py @@ -89,148 +89,206 @@ def get_timestamp_last_changed(file_path): return datetime.fromtimestamp(chg_time).isoformat() -class OsmData(): # pylint: disable=too-few-public-methods +class InformalOsmDataInterface: """ object with all internal parameters to process maps """ - def __init__(self): + def __init__(self, o_input_data): """ - xxx + steps in constructor: + 1. take over input paramters (force_processing is changed in the function further down) + 2. check + download geofabrik file (always) """ self.force_processing = False self.tiles = [] self.border_countries = {} self.country_name = '' - def process_input_of_the_tool(self, o_input_data): + self.o_downloader = Downloader( + o_input_data.max_days_old, o_input_data.force_download, self.border_countries) + # takeover what is given by user first for force_processing + self.force_processing = o_input_data.force_processing + self.process_border_countries = o_input_data.process_border_countries + + log.info('-' * 80) + + # geofabrik file + if self.o_downloader.should_geofabrik_file_be_downloaded(): + self.force_processing = True + self.o_downloader.download_geofabrik_file() + + def process_input_of_the_tool(self): """ Process input: get relevant tiles and if border countries should be calculated The three primary inputs are giving by a separate value each and have separate processing: 1. country name 2. x/y combinations + """ - # Check for not existing or expired files. Mark for download, if dl is needed - # - land polygons file - # - .osm.pbf files + def calc_tiles(self): + """ + calculate relevant tiles for input country or xy coordinate + """ - steps in this function: - 1. take over input paramters (force_processing is changed in the function further down) - 2. check + download geofabrik file if geofabrik-processing - 3. calculate relevant tiles for map creation - 4. calculate border countries for map creation - 5. evaluate the country-name for folder cration during processing - 6. calculate if force_processing should be set to true + def calc_border_countries(self): """ + calculate the border countries for the given tiles. i.e. + - if CLI/GUI input by user + - if processing x/y coordinates + """ + log.info('# Determine involved/border countries') - o_downloader = Downloader( - o_input_data.max_days_old, o_input_data.force_download, self.border_countries) - # takeover what is given by user first for force_processing - self.force_processing = o_input_data.force_processing + # Build list of countries needed + for tile in self.tiles: + for country in tile['countries']: + if country not in self.border_countries: + self.border_countries[country] = {} - log.info('-' * 80) + def log_border_countries(self): + """ + write calculated border countries/involved countries to log + """ + for country in self.border_countries: + log.info('+ Involved country: %s', country) - # geofabrik file - if o_downloader.should_geofabrik_file_be_downloaded(): + # input can be only one country, if there are more than one, + # border countries must be selected + if len(self.border_countries) > 1: + log.info('+ Border countries will be processed') + + def get_downloader(self): + """ + steps in this function: + 1. Check for not existing or expired files. Mark for download, if dl is needed + - land polygons file + - .osm.pbf files + 2. Calculate if force_processing should be set to true + """ + # calc force processing + # Check for not existing or expired files. Mark for download, if dl is needed + self.o_downloader.check_land_polygons_file() + self.o_downloader.check_osm_pbf_file() + + # If one of the files needs to be downloaded, reprocess all files + if self.o_downloader.need_to_dl: self.force_processing = True - o_downloader.download_geofabrik_file() + + return self.o_downloader + + +class CountryOsmData(InformalOsmDataInterface): + """ + object with all internal parameters to process maps for countries + """ + + def __init__(self, o_input_data): + super().__init__(o_input_data) + self.input_country = o_input_data.country + + self.o_geofabrik = CountryGeofabrik(self.input_country) + + def process_input_of_the_tool(self): + """ + steps in this function: + 1. calculate relevant tiles for map creation + 2. calculate border countries for map creation + 3. evaluate the country-name for folder cration during processing + """ # calc tiles - if o_input_data.country: - self.calc_tiles_country(o_input_data) - elif o_input_data.xy_coordinates: - self.calc_tiles_xy(o_input_data) + self.calc_tiles() # calc border countries log.info('-' * 80) - if o_input_data.country: - self.calc_border_countries_country(o_input_data) - elif o_input_data.xy_coordinates: - self.calc_border_countries() + self.calc_border_countries() # log border countries when and when not calculated to output the processed country(s) self.log_border_countries() # calc country name - if o_input_data.country: - # country name is the input argument - self.country_name = o_input_data.country - elif o_input_data.xy_coordinates: - self.calc_country_name_xy() - - # calc force processing - # Check for not existing or expired files. Mark for download, if dl is needed - o_downloader.check_land_polygons_file() - o_downloader.check_osm_pbf_file() - - # If one of the files needs to be downloaded, reprocess all files - if o_downloader.need_to_dl: - self.force_processing = True - - return o_downloader + self.calc_country_name() - def calc_tiles_country(self, o_input_data): + def calc_tiles(self): """ option 1: input a country as parameter, e.g. germany """ - log.info('# Input country: %s.', o_input_data.country) + log.info('# Input country: %s.', self.input_country) # use Geofabrik-URL to calculate the relevant tiles - o_geofabrik = CountryGeofabrik(o_input_data.country) - self.tiles = o_geofabrik.get_tiles_of_wanted_map() + self.tiles = self.o_geofabrik.get_tiles_of_wanted_map() - def calc_tiles_xy(self, o_input_data): - """ - option 2: input a x/y combinations as parameter, e.g. 134/88 or 133/88,130/100 - """ - log.info( - '# Input X/Y coordinates: %s.', o_input_data.xy_coordinates) - - # use Geofabrik-URL to get the relevant tiles - xy_coordinates = get_xy_coordinates_from_input( - o_input_data.xy_coordinates) - - o_geofabrik = XYGeofabrik(xy_coordinates) - # find the tiles for x/y combinations in the geofabrik json files - self.tiles = o_geofabrik.get_tiles_of_wanted_map() - - def calc_border_countries_country(self, o_input_data): + def calc_border_countries(self): """ calculate the border countries for the given tiles when input is a country - if CLI/GUI input by user """ - if o_input_data.process_border_countries: - self.calc_border_countries() + if self.process_border_countries: + super().calc_border_countries() # set the to-be-processed country as border country else: - self.border_countries[o_input_data.country] = {} + self.border_countries[self.input_country] = {} - def calc_border_countries(self): + def calc_country_name(self): """ - calculate the border countries for the given tiles. i.e. - - if CLI/GUI input by user - - if processing x/y coordinates + country name is the country + >1 countries are separated by underscore """ - log.info('# Determine involved/border countries') + self.country_name = self.input_country - # Build list of countries needed - for tile in self.tiles: - for country in tile['countries']: - if country not in self.border_countries: - self.border_countries[country] = {} - def log_border_countries(self): +class XYOsmData(InformalOsmDataInterface): + """ + object with all internal parameters to process maps for XY coordinates + """ + + def __init__(self, o_input_data): + super().__init__(o_input_data) + self.input_xy_coordinates = o_input_data.xy_coordinates + + def process_input_of_the_tool(self): """ - write calculated border countries/involved countries to log + Process input: get relevant tiles and if border countries should be calculated + The three primary inputs are giving by a separate value each and have separate processing: + 1. country name + 2. x/y combinations + + # Check for not existing or expired files. Mark for download, if dl is needed + # - land polygons file + # - .osm.pbf files + + steps in this function: + 1. calculate relevant tiles for map creation + 2. calculate border countries for map creation + 3. evaluate the country-name for folder cration during processing """ - for country in self.border_countries: - log.info('+ Involved country: %s', country) - # input can be only one country, if there are more than one, - # border countries must be selected - if len(self.border_countries) > 1: - log.info('+ Border countries will be processed') + # calc tiles + self.calc_tiles() + + # calc border countries + log.info('-' * 80) + self.calc_border_countries() + # log border countries when and when not calculated to output the processed country(s) + self.log_border_countries() + + # calc country name + self.calc_country_name() + + def calc_tiles(self): + """ + option 2: input a x/y combinations as parameter, e.g. 134/88 or 133/88,130/100 + """ + log.info('# Input X/Y coordinates: %s.', self.input_xy_coordinates) + + # use Geofabrik-URL to get the relevant tiles + xy_coordinates = get_xy_coordinates_from_input( + self.input_xy_coordinates) + + o_geofabrik = XYGeofabrik(xy_coordinates) + # find the tiles for x/y combinations in the geofabrik json files + self.tiles = o_geofabrik.get_tiles_of_wanted_map() - def calc_country_name_xy(self): + def calc_country_name(self): """ country name is the X/Y combinations separated by minus >1 x/y combinations are separated by underscore