[DEV] multiple small corrections & unifications. pylint findings, log…

…ging, comments, unittests (#107)

* adjust config for unittest & delete .dbf files

- .dbf files are always different. If the file is deleted it will not be compared in unittest

* refactor logging output to be consistent with #/+

* only process countries which are involved in the tile

- in merge_splitted_tiles_with_land_and_sea() the other ones are note included as well!

* refactor .osm.pbf file download

- adjust unittests + add new constant unittests

* launch-config: STG->gardasee

* correct pylint findings

- specific exceptions
- specific encoding
- unittests for the changes above
- unified pylint disablement

.vscode/launch.json

@@ -137,7 +137,8 @@
                 "-fp",
                 "-c",
                 "-md",
-                "9999"
+                "9999",
+                "-nbc"
             ]
         },
         {
@@ -175,6 +176,19 @@
                 "-xy",
                 "138/100,133/88"
             ]
+        },
+        {
+            "name": "x/y: stg->gardasee (ten tiles)",
+            "type": "python",
+            "request": "launch",
+            "program": "${workspaceRoot}/wahoo_map_creator.py",
+            "console": "integratedTerminal",
+            "args": [
+                "cli",
+                "-xy",
+                "134/88,134/89,134/90,134/91,135/88,135/89,135/90,135/91,136/90,136/91",
+                "-om"
+            ]
         }
     ]
 }

common_python/constants_functions.py

@@ -83,7 +83,7 @@ def translate_country_input_to_geofabrik(county):
 
     try:
         c_translated = constants.Translate_Country[f'{county}']
-    except:
+    except KeyError:
         c_translated = county
 
     return c_translated

common_python/downloader.py

@@ -33,7 +33,7 @@ def download_file(target_filepath, url, is_zip):
     """
     logging_filename = target_filepath.split(os.sep)[-1]
     log.info('-' * 80)
-    log.info('# Downloading %s file', logging_filename)
+    log.info('+ Downloading %s file', logging_filename)
     if is_zip:
         # build target-filepath based on last element of URL
         last_part = url.rsplit('/', 1)[-1]
@@ -55,11 +55,11 @@ def download_file(target_filepath, url, is_zip):
         log.info('+ Downloaded: %s', target_filepath)
 
 
-def download_osm_pbf_file(country):
+def get_osm_pbf_filepath_url(country):
     """
-    download a countries' OSM file
+    evaluate a countries' OSM file URL and download filepath
     """
-    log.info('+ Trying to download missing map of %s.', country)
+
     # get .osm.pbf region of country
     transl_c = const_fct.translate_country_input_to_geofabrik(country)
     region = const_fct.get_geofabrik_region_of_country(country)
@@ -68,11 +68,10 @@ def download_osm_pbf_file(country):
             '/' + transl_c + '-latest.osm.pbf'
     else:
         url = 'https://download.geofabrik.de/' + transl_c + '-latest.osm.pbf'
-    # download URL to file
     map_file_path = os.path.join(
         fd_fct.MAPS_DIR, f'{transl_c}' + '-latest.osm.pbf')
-    download_file(map_file_path, url, False)
-    return map_file_path
+    # return URL and download filepath
+    return map_file_path, url
 
 
 class Downloader:
@@ -91,10 +90,12 @@ def check_and_download_geofabrik_if_needed(self):
         check geofabrik file if not existing or is not up-to-date
         and download if needed
         """
+
         force_processing = False
 
         if self.check_file(fd_fct.GEOFABRIK_PATH) is True or \
                 self.force_download is True:
+            log.info('# Need to download geofabrik file')
             download_file(fd_fct.GEOFABRIK_PATH,
                           'https://download.geofabrik.de/index-v1.json', False)
             force_processing = True
@@ -108,10 +109,12 @@ def check_and_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
         """
+
         force_processing = False
 
         if self.check_file(fd_fct.LAND_POLYGONS_PATH) is True or \
                 self.force_download is True:
+            log.info('# Need to download land polygons file')
             download_file(fd_fct.LAND_POLYGONS_PATH,
                           'https://osmdata.openstreetmap.de/download/land-polygons-split-4326.zip', True)
             force_processing = True
@@ -123,7 +126,8 @@ def check_and_download_files_if_needed(self):
             for country, item in self.border_countries.items():
                 try:
                     if item['download'] is True:
-                        map_file_path = download_osm_pbf_file(country)
+                        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:
@@ -156,7 +160,7 @@ def check_file(self, target_filepath):
                 os.remove(target_filepath)
                 need_to_download = True
 
-        except:
+        except FileNotFoundError:
             need_to_download = True
 
         # if file does not exists --> download
@@ -172,6 +176,7 @@ 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')

common_python/file_directory_functions.py

@@ -57,8 +57,8 @@ def unzip(source_filename, dest_dir):
             for word in words[:-1]:
                 while True:
                     drive, word = os.path.splitdrive(word)
-                    head, word = os.path.split(
-                        word)  # pylint: disable=unused-variable
+                    head, word = os.path.split(  # pylint: disable=unused-variable
+                        word)
                     if not drive:
                         break
                 if word in (os.curdir, os.pardir, ''):
@@ -96,7 +96,7 @@ def read_json_file(json_file_path):
     log.debug('-' * 80)
     log.debug('# Read json file')
 
-    with open(json_file_path) as json_file:
+    with open(json_file_path, encoding="utf-8") as json_file:
         tiles_from_json = json.load(json_file)
         json_file.close()
     if tiles_from_json == '':
@@ -127,7 +127,7 @@ def write_to_file(file_path, request):
     """
     write content of request into given file path
     """
-    with open(file_path, 'wb') as file_handle:
+    with open(file_path, mode='wb') as file_handle:
         for chunk in request.iter_content(chunk_size=1024*100):
             file_handle.write(chunk)
 

common_python/osm_maps_functions.py

@@ -25,6 +25,10 @@
 log = logging.getLogger('main-logger')
 
 
+class TileNotFoundError(Exception):
+    """Raised when no tile is found for x/y combination"""
+
+
 def get_xy_coordinates_from_input(input_xy_coordinates):
     """
     extract/split x/y combinations by given X/Y coordinates.
@@ -64,6 +68,9 @@ def get_tile_by_one_xy_combination_from_jsons(xy_combination):
                 if tile['x'] == xy_combination['x'] and tile['y'] == xy_combination['y']:
                     return tile
 
+    # if function is processed until here, there is no tile found for the x/y combination --> Exception
+    raise TileNotFoundError
+
 
 def run_subprocess_and_log_output(cmd, error_message, cwd=""):
     """
@@ -163,15 +170,19 @@ def process_input(self, calc_border_countries):
 
                 # loop through x/y combinations and find each tile in the json files
                 for xy_comb in xy_coordinates:
-                    self.tiles.append(get_tile_by_one_xy_combination_from_jsons(
-                        xy_comb))
+                    try:
+                        self.tiles.append(get_tile_by_one_xy_combination_from_jsons(
+                            xy_comb))
 
-                    # country name is the X/Y combinations separated by minus
-                    # >1 x/y combinations are separated by underscore
-                    if not self.country_name:
-                        self.country_name = f'{xy_comb["x"]}-{xy_comb["y"]}'
-                    else:
-                        self.country_name = f'{self.country_name}_{xy_comb["x"]}-{xy_comb["y"]}'
+                        # country name is the X/Y combinations separated by minus
+                        # >1 x/y combinations are separated by underscore
+                        if not self.country_name:
+                            self.country_name = f'{xy_comb["x"]}-{xy_comb["y"]}'
+                        else:
+                            self.country_name = f'{self.country_name}_{xy_comb["x"]}-{xy_comb["y"]}'
+
+                    except TileNotFoundError:
+                        pass
 
             # calc border country when input X/Y coordinates
             calc_border_countries = True
@@ -384,7 +395,7 @@ def generate_sea(self):
             if not os.path.isfile(out_file) or self.force_processing is True:
                 log.info(
                     '+ Generate sea %s of %s for Coordinates: %s,%s', tile_count, len(self.tiles), tile["x"], tile["y"])
-                with open(os.path.join(fd_fct.COMMON_DIR, 'sea.osm')) as sea_file:
+                with open(os.path.join(fd_fct.COMMON_DIR, 'sea.osm'), encoding="utf-8") as sea_file:
                     sea_data = sea_file.read()
 
                     # Try to prevent getting outside of the +/-180 and +/- 90 degrees borders. Normally the +/- 0.1 are there to prevent white lines at tile borders
@@ -407,7 +418,7 @@ def generate_sea(self):
                         sea_data = sea_data.replace(
                             '$TOP', f'{tile["top"]+0.1:.6f}')
 
-                    with open(out_file, 'w') as output_file:
+                    with open(out_file, mode='w', encoding="utf-8") as output_file:
                         output_file.write(sea_data)
             tile_count += 1
 
@@ -424,6 +435,8 @@ def split_filtered_country_files_to_tiles(self):
         for tile in self.tiles:
 
             for country, val in self.border_countries.items():
+                if country not in tile['countries']:
+                    continue
                 log.info(
                     '+ Splitting tile %s of %s for Coordinates: %s,%s from map of %s', tile_count, len(self.tiles), tile["x"], tile["y"], country)
                 out_file = os.path.join(fd_fct.OUTPUT_DIR,
@@ -661,7 +674,7 @@ def create_map_files(self, save_cruiser, tag_wahoo_xml):
                     cmd, f'! Error creating map files for tile: {tile["x"]},{tile["y"]}')
 
             # Create "tile present" file
-            with open(out_file + '.lzma.12', 'wb') as tile_present_file:
+            with open(out_file + '.lzma.12', mode='wb') as tile_present_file:
                 tile_present_file.close()
 
             tile_count += 1
@@ -704,7 +717,7 @@ def make_and_zip_files(self, keep_map_folders, extension):
                 os.makedirs(outdir)
             try:
                 shutil.copy2(src, dst)
-            except Exception as exception:
+            except Exception as exception:  # pylint: disable=broad-except
                 log.error(
                     '! Error copying %s files for country %s: %s', extension, self.country_name, exception)
                 sys.exit()
@@ -716,7 +729,7 @@ def make_and_zip_files(self, keep_map_folders, extension):
                     os.makedirs(outdir)
                 try:
                     shutil.copy2(src, dst)
-                except Exception as exception:
+                except Exception as exception:  # pylint: disable=broad-except
                     log.error(
                         '! Error copying version files for country %s: %s', self.country_name, exception)
                     sys.exit()

tests/test_constants.py

@@ -0,0 +1,43 @@
+"""
+tests for the downloader file
+"""
+import unittest
+
+
+from common_python.constants_functions import translate_country_input_to_geofabrik
+
+
+class TestConstants(unittest.TestCase):
+    """
+    tests for constants and constants functions files
+    """
+
+    def test_translated_countries_to_china(self):
+        """
+        Test countries which have no own geofabrik country but are included in china
+        """
+
+        expected = 'china'
+
+        transl_c = translate_country_input_to_geofabrik('hong_kong')
+        self.assertEqual(expected, transl_c)
+
+        transl_c = translate_country_input_to_geofabrik('macao')
+        self.assertEqual(expected, transl_c)
+
+        transl_c = translate_country_input_to_geofabrik('paracel_islands')
+        self.assertEqual(expected, transl_c)
+
+    def test_translated_countries_no_mapping(self):
+        """
+        Test countries which have no own geofabrik country but are included in china
+        """
+
+        expected = 'germany'
+
+        transl_c = translate_country_input_to_geofabrik('germany')
+        self.assertEqual(expected, transl_c)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_downloader.py

@@ -11,6 +11,7 @@
 
 from common_python.downloader import older_than_x_days
 from common_python.downloader import download_file
+from common_python.downloader import get_osm_pbf_filepath_url
 from common_python.downloader import Downloader
 from common_python import file_directory_functions as fd_fct
 
@@ -82,6 +83,38 @@ def test_download_geofabrik_file(self):
 
         self.assertTrue(os.path.exists(path))
 
+    def test_download_malta_osm_pbf_file(self):
+        """
+        Test the download of geofabrik file via URL
+        """
+
+        country = 'malta'
+
+        path = os.path.join(fd_fct.COMMON_DL_DIR, 'maps',
+                            f'{country}' + '-latest.osm.pbf')
+
+        if os.path.exists(path):
+            os.remove(path)
+
+        map_file_path, url = get_osm_pbf_filepath_url(country)
+        download_file(map_file_path, url, False)
+
+        self.assertTrue(os.path.exists(path))
+
+    def test_delete_not_existing_file(self):
+        """
+        Test if the removal of a not existing file raises a exception
+        """
+
+        path = os.path.join(fd_fct.COMMON_DL_DIR, 'maps',
+                            'malta' + '-latest.osm.pbf')
+
+        if os.path.exists(path):
+            os.remove(path)
+
+        with self.assertRaises(FileNotFoundError):
+            os.remove(path)
+
 
 if __name__ == '__main__':
     unittest.main()