[FEAT] Refactor a lot to decouple and make maintenance easier (focus …

…osm_maps_functions.py) (#152)

* import file_directory functions one by one

- and not by importing the whole file "as" fd_fct

* create OsmData object and insert into class OsmMaps

* move check for -co and -xy to input.py

* shuffle a lot

* go throught calc_border_countries also when no border countries are wanted

* break OsmData in small pieces

* refactor border country calc + fix unittests

* force_processing refactoring, all checks in OsmData now!

- move to function process_input_of_the_tool
- move corresponding stuff from check_and_download_files also
- additional needed changes - mainiy in downloader.py
- adjust tests & do nearly refactoring

* import needed functions + pylint findings

* refactor imports to "single import" and not the whole file

* fix pylint finding

* Bump to version v2.1.0a13

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = wahoomc
-version = 2.0.2
+version = 2.1.0a13
 author = Benjamin Kreuscher
 author_email = [email protected]
 description = Create maps for your Wahoo bike computer based on latest OSM maps 

tests/test_downloader.py

@@ -83,6 +83,25 @@ def test_download_geofabrik_file(self):
 
         self.assertTrue(os.path.exists(path))
 
+    def test_download_geofabrik_file_2(self):
+        """
+        Test the download of geofabrik file via URL.
+        check & download via built functions of downloader class
+        """
+        path = os.path.join(constants.USER_DL_DIR, 'geofabrik.json')
+
+        if os.path.exists(path):
+            os.remove(path)
+
+        self.assertTrue(
+            self.o_downloader.should_geofabrik_file_be_downloaded())
+
+        self.o_downloader.download_geofabrik_file()
+
+        self.assertTrue(os.path.exists(path))
+        self.assertFalse(
+            self.o_downloader.should_geofabrik_file_be_downloaded())
+
     def test_download_malta_osm_pbf_file(self):
         """
         Test the download of geofabrik file via URL
@@ -115,6 +134,21 @@ def test_delete_not_existing_file(self):
         with self.assertRaises(FileNotFoundError):
             os.remove(path)
 
+    def test_check_dl_needed_geofabrik(self):
+        """
+        Test if geofabrik file needs to be downloaded
+        """
+        path = os.path.join(constants.USER_DL_DIR, 'geofabrik.json')
+
+        if os.path.exists(path):
+            os.remove(path)
+
+        self.o_downloader.max_days_old = 24
+        # self.o_downloader.check_geofabrik_file()
+
+        self.assertTrue(
+            self.o_downloader.should_geofabrik_file_be_downloaded())
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_geofabrik.py

@@ -46,7 +46,7 @@ def setUp(self):
 
         if not os.path.isfile(constants.GEOFABRIK_PATH):
             o_downloader = Downloader(24, False)
-            o_downloader.check_and_download_geofabrik_if_needed()
+            o_downloader.download_geofabrik_file()
 
     def test_tiles_via_geofabrik_malta(self):
         """

tests/test_osm_maps.py

@@ -8,7 +8,7 @@
 # import custom python packages
 # sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from wahoomc.osm_maps_functions import OsmMaps
+from wahoomc.osm_maps_functions import OsmData
 from wahoomc.osm_maps_functions import get_tile_by_one_xy_combination_from_jsons
 from wahoomc.osm_maps_functions import get_xy_coordinates_from_input
 from wahoomc.osm_maps_functions import TileNotFoundError
@@ -23,21 +23,21 @@ class TestOsmMaps(unittest.TestCase):
     tests for the OSM maps file
     """
 
-    def setUp(self):
-        o_input_data = InputData()
-
-        self.o_osm_maps = OsmMaps(o_input_data)
+    # def setUp(self):
 
     def test_input_country_malta(self):
         """
         Test "malta" as input to the wahooMapsCreator
         check, if the given input-parameter is saved to the OsmMaps instance
         """
 
-        self.o_osm_maps.o_input_data.country = 'malta'
-        self.o_osm_maps.process_input(True)
+        o_input_data = InputData()
+        o_input_data.country = 'malta'
+
+        o_osm_data = OsmData()
+        o_osm_data.process_input_of_the_tool(o_input_data)
 
-        result = self.o_osm_maps.country_name
+        result = o_osm_data.country_name
         self.assertEqual(result, 'malta')
 
     def test_calc_border_countries_input_country(self):
@@ -123,16 +123,27 @@ def process_and_check_border_countries(self, inp_val, calc_border_c, exp_result,
         helper method to process a country or json file and check the calculated border countries
         """
 
+        o_input_data = InputData()
         if inp_mode == 'country':
-            self.o_osm_maps.o_input_data.country = inp_val
+            o_input_data.country = inp_val
         elif inp_mode == 'xy_coordinate':
-            self.o_osm_maps.o_input_data.xy_coordinates = inp_val
+            o_input_data.xy_coordinates = inp_val
+        o_input_data.process_border_countries = calc_border_c
+
+        o_osm_data = OsmData()
+        o_osm_data.process_input_of_the_tool(o_input_data)
 
-        self.o_osm_maps.process_input(calc_border_c)
-        result = self.o_osm_maps.border_countries
+        result = o_osm_data.border_countries
 
+        # delete the path to the file, here, only the correct border countries are checked
+        for res in result:
+            result[res] = {}
+
+        # self.assertEqual(result, exp_result)
         self.assertEqual(result, exp_result)
 
+        # self.assertDictEqual()  Equal(tIn(member=exp_result, container=result)
+
 
 class TestStaticJsons(unittest.TestCase):
     """

wahoomc/downloader.py

@@ -12,15 +12,13 @@
 import logging
 
 # import custom python packages
-from wahoomc import file_directory_functions as fd_fct
-from wahoomc import constants_functions as const_fct
+from wahoomc.file_directory_functions import download_url_to_file, unzip
+from wahoomc.constants_functions import translate_country_input_to_geofabrik, get_geofabrik_region_of_country
 
-# User
 from wahoomc.constants import USER_DL_DIR
 from wahoomc.constants import USER_MAPS_DIR
 from wahoomc.constants import LAND_POLYGONS_PATH
 from wahoomc.constants import GEOFABRIK_PATH
-from wahoomc.constants import USER_OUTPUT_DIR
 
 log = logging.getLogger('main-logger')
 
@@ -46,14 +44,14 @@ def download_file(target_filepath, url, is_zip):
         last_part = url.rsplit('/', 1)[-1]
         dl_file_path = os.path.join(USER_DL_DIR, last_part)
         # download URL to file
-        fd_fct.download_url_to_file(url, dl_file_path)
+        download_url_to_file(url, dl_file_path)
         # unpack it - should work on macOS and Windows
-        fd_fct.unzip(dl_file_path, USER_DL_DIR)
+        unzip(dl_file_path, USER_DL_DIR)
         # delete .zip file
         os.remove(dl_file_path)
     else:
         # no zipping --> directly download to given target filepath
-        fd_fct.download_url_to_file(url, target_filepath)
+        download_url_to_file(url, target_filepath)
     # Check if file exists (if target file exists)
     if not os.path.isfile(target_filepath):
         log.error('! failed to find %s', target_filepath)
@@ -67,8 +65,8 @@ def get_osm_pbf_filepath_url(country):
     build the geofabrik download url to a countries' OSM file and download filepath
     """
 
-    transl_c = const_fct.translate_country_input_to_geofabrik(country)
-    region = const_fct.get_geofabrik_region_of_country(country)
+    transl_c = translate_country_input_to_geofabrik(country)
+    region = get_geofabrik_region_of_country(country)
     if region != 'no':
         url = 'https://download.geofabrik.de/' + region + \
             '/' + transl_c + '-latest.osm.pbf'
@@ -87,69 +85,61 @@ class Downloader:
     This is the class to check and download maps / artifacts"
     """
 
-    def __init__(self, max_days_old, force_download):
+    def __init__(self, max_days_old, force_download, tiles_from_json=None, border_countries=None):
         self.max_days_old = max_days_old
         self.force_download = force_download
-        self.tiles_from_json = []
-        self.border_countries = {}
+        self.tiles_from_json = tiles_from_json
+        self.border_countries = border_countries
 
-    def check_and_download_geofabrik_if_needed(self):
+        self.need_to_dl = []
+
+    def should_geofabrik_file_be_downloaded(self):
         """
         check geofabrik file if not existing or is not up-to-date
-        and download if needed
-        """
-
-        force_processing = False
 
+        # if geofabrik file needs to be downloaded, force-processing is set because there might be
+        # a change in the geofabrik file
+        """
         if self.check_file(GEOFABRIK_PATH) is True or \
                 self.force_download is True:
             log.info('# Need to download geofabrik file')
-            download_file(GEOFABRIK_PATH,
-                          'https://download.geofabrik.de/index-v1.json', False)
-            force_processing = True
 
-        log.info('+ check geofabrik.json file: OK')
+            return True
 
-        return force_processing
+        return False
 
-    def check_and_download_files_if_needed(self):
+    def download_geofabrik_file(self):
         """
-        check land_polygons and OSM map files if not existing or are not up-to-date
-        and download if needed
+        download geofabrik file
         """
+        download_file(GEOFABRIK_PATH,
+                      'https://download.geofabrik.de/index-v1.json', False)
 
-        force_processing = False
+        log.info('+ download geofabrik.json file: OK')
+
+    def check_land_polygons_file(self):
+        """
+        check land_polygons file if not existing or are not up-to-date
+        """
 
         if self.check_file(LAND_POLYGONS_PATH) is True or \
                 self.force_download is True:
             log.info('# Need to download land polygons file')
+            self.need_to_dl.append('land_polygons')
+
+    def download_files_if_needed(self):
+        """
+        check land_polygons and OSM map files if not existing or are not up-to-date
+        and download if needed
+        """
+        if 'land_polygons' in self.need_to_dl:
             download_file(LAND_POLYGONS_PATH,
                           'https://osmdata.openstreetmap.de/download/land-polygons-split-4326.zip', True)
-            force_processing = True
-
-        log.info('+ check land_polygons.shp file: OK')
-
-        if self.check_osm_pbf_file() is True or self.force_download is True:
-            # trigger download of relevant countries' OSM files
-            for country, item in self.border_countries.items():
-                try:
-                    if item['download'] is True:
-                        map_file_path, url = get_osm_pbf_filepath_url(country)
-                        download_file(map_file_path, url, False)
-                        self.border_countries[country] = {
-                            'map_file': map_file_path}
-                except KeyError:
-                    pass
 
-            force_processing = True
+        # log.info('+ download land_polygons.shp file: OK')
 
-        log.info('+ Check countries .osm.pbf files: OK')
-
-        # if force_processing is True:
-        fd_fct.create_empty_directories(
-            USER_OUTPUT_DIR, self.tiles_from_json)
-
-        return force_processing
+        if 'osm_pbf' in self.need_to_dl:
+            self.download_osm_pbf_file()
 
     def check_file(self, target_filepath):
         """
@@ -186,7 +176,6 @@ def check_osm_pbf_file(self):
         check if the relevant countries' OSM files are up-to-date
         """
 
-        need_to_download = False
         log.info('-' * 80)
         log.info('# check countries .osm.pbf files')
 
@@ -197,7 +186,7 @@ def check_osm_pbf_file(self):
             # get translated country (geofabrik) of country
             # do not download the same file for different countries
             # --> e.g. China, Hong Kong and Macao, see Issue #11
-            transl_c = const_fct.translate_country_input_to_geofabrik(country)
+            transl_c = translate_country_input_to_geofabrik(country)
 
             # check for already existing .osm.pbf file
             map_file_path = glob.glob(
@@ -212,7 +201,7 @@ def check_osm_pbf_file(self):
                     log.info(
                         '+ mapfile for %s: deleted. Input: %s.', transl_c, country)
                     os.remove(map_file_path[0])
-                    need_to_download = True
+                    self.need_to_dl.append('osm_pbf')
                 else:
                     self.border_countries[country] = {
                         'map_file': map_file_path[0]}
@@ -223,11 +212,21 @@ def check_osm_pbf_file(self):
             map_file_path = self.border_countries[country].get('map_file')
             if map_file_path is None or (not os.path.isfile(map_file_path) or self.force_download):
                 self.border_countries[country]['download'] = True
-                need_to_download = True
-
-        # self.border_countries = border_countries
+                self.need_to_dl.append('osm_pbf')
 
-        return need_to_download
+    def download_osm_pbf_file(self):
+        """
+        download countries' OSM files
+        """
+        for country, item in self.border_countries.items():
+            try:
+                if item['download'] is True:
+                    map_file_path, url = get_osm_pbf_filepath_url(country)
+                    download_file(map_file_path, url, False)
+                    self.border_countries[country] = {
+                        'map_file': map_file_path}
+            except KeyError:
+                pass
 
     def should_file_be_downloaded(self, file_path):
         """

wahoomc/geofabrik.py

@@ -12,7 +12,8 @@
 from shapely.geometry import Polygon, shape
 
 # import custom python packages
-from wahoomc import constants
+from wahoomc.constants import GEOFABRIK_PATH
+from wahoomc.constants import special_regions, geofabrik_regions, block_download
 
 log = logging.getLogger('main-logger')
 
@@ -142,7 +143,7 @@ def geom(wanted):
     input parameter is the name of the desired country/region as use by Geofabric
     in their json file.
     """
-    with open(constants.GEOFABRIK_PATH, encoding='utf8') as file_handle:
+    with open(GEOFABRIK_PATH, encoding='utf8') as file_handle:
         data = geojson.load(file_handle)
     file_handle.close()
 
@@ -196,7 +197,7 @@ def find_needed_countries(bbox_tiles, wanted_map, wanted_region_polygon):
     """
     output = []
 
-    with open(constants.GEOFABRIK_PATH, encoding='utf8') as file_handle:
+    with open(GEOFABRIK_PATH, encoding='utf8') as file_handle:
         geofabrik_json_data = geojson.load(file_handle)
     file_handle.close()
 
@@ -244,19 +245,19 @@ def find_needed_countries(bbox_tiles, wanted_map, wanted_region_polygon):
                 if rshape.intersects(poly):  # if so
                     # catch special_regions like (former) colonies where the map of the region is not fysically in the map of the parent country.
                     # example Guadeloupe, it's parent country is France but Guadeloupe is not located within the region covered by the map of France
-                    if wanted_map not in constants.special_regions:
+                    if wanted_map not in special_regions:
                         # If we are proseccing a sub-region add the parent of this sub-region
                         # to the must download list.
                         # This to prevent downloading several small regions AND it's containing region
                         # we are processing a sub-regiongo find the parent region:
-                        if parent not in constants.geofabrik_regions and regionname not in constants.geofabrik_regions:
+                        if parent not in geofabrik_regions and regionname not in geofabrik_regions:
                             # we are processing a sub-regiongo find the parent region
                             x_value = 0
                             # handle sub-sub-regions like unterfranken->bayern->germany
-                            while parent not in constants.geofabrik_regions:
+                            while parent not in geofabrik_regions:
                                 parent, child = find_geofbrik_parent(
                                     parent, geofabrik_json_data)
-                                if parent in constants.geofabrik_regions:
+                                if parent in geofabrik_regions:
                                     parent = child
                                     break
                                 if x_value > 10:  # prevent endless loop
@@ -288,7 +289,7 @@ def find_needed_countries(bbox_tiles, wanted_map, wanted_region_polygon):
             if regionname != wanted_map:
                 # check if we are processing a country or a sub-region.
                 # For countries only process other countries. also block special geofabrik sub regions
-                if parent in constants.geofabrik_regions and regionname not in constants.block_download:
+                if parent in geofabrik_regions and regionname not in block_download:
                     # processing a country and no special sub-region
                     # check if rshape is subset of desired region. If so discard it
                     if wanted_region_polygon.contains(rshape):