[DEV] refactor osm_maps_functions.py like geofabrik.py - using a interface and two implementing classes

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()

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()

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