Switch to using pathlib for paths

buildmap/exporter/geojson.py

@@ -1,5 +1,4 @@
 import json
-import os
 import time
 from sqlalchemy import text
 from . import Exporter
@@ -14,12 +13,11 @@ def export(self):
         self.source_tables = dict(
             (table, layer) for layer, table in self.buildmap.get_source_layers()
         )
-        self.output_dir = os.path.join(self.config["web_directory"], "vector")
+        self.output_dir = (
+            self.buildmap.resolve_path(self.config["web_directory"]) / "vector"
+        )
 
-        try:
-            os.mkdir(self.output_dir)
-        except OSError:
-            pass
+        self.output_dir.mkdir(parents=True, exist_ok=True)
 
         for layer in self.config["vector_layer"]:
             layer_name = layer["name"]
@@ -140,7 +138,9 @@ def generate_layer(self, name, source_layers):
             "features": result,
         }
 
-        with open(os.path.join(self.output_dir, "%s.json" % name), "w") as fp:
+        output_path = self.output_dir / ("%s.json" % name)
+
+        with output_path.open("w") as fp:
             json.dump(geojson, fp, indent=4)
 
     def generate_layer_index(self):
@@ -156,9 +156,13 @@ def generate_layer_index(self):
             )
 
         data = {"layers": vector_layers, "styles": self.generate_styles()}
-        with open(
-            os.path.join(self.config["web_directory"], "vector_layers.json"), "w"
-        ) as fp:
+
+        output_path = (
+            self.buildmap.resolve_path(self.config["web_directory"])
+            / "vector_layers.json"
+        )
+
+        with output_path.open("w") as fp:
             json.dump(data, fp, indent=4)
 
     def generate_styles(self):

buildmap/main.py

@@ -10,7 +10,9 @@
 import importlib
 from json.decoder import JSONDecodeError
 from collections import defaultdict
+from typing import Union
 from shapely.geometry import MultiPolygon, Polygon
+from pathlib import Path
 
 from .util import sanitise_layer
 from .mapdb import MapDB
@@ -61,15 +63,11 @@ def __init__(self):
         self.bbox = None
 
         # Resolve any relative paths with respect to the first config file
-        self.base_path = os.path.dirname(os.path.abspath(self.args.config[0]))
+        self.base_path = Path(self.args.config[0]).absolute().parent
         self.temp_dir = self.resolve_path(self.config["output_directory"])
         self.known_attributes = defaultdict(set)
         shutil.rmtree(self.temp_dir, True)
-        try:
-            os.makedirs(self.temp_dir)
-        except OSError as e:
-            if e.errno != errno.EEXIST:
-                raise
+        self.temp_dir.mkdir(parents=True, exist_ok=True)
 
     def load_config(self, config_files):
         config = {}
@@ -81,12 +79,12 @@ def load_config(self, config_files):
                 raise Exception("Error loading config file {}: {}".format(filename, e))
         return config
 
-    def resolve_path(self, path):
-        return os.path.normpath(os.path.join(self.base_path, path))
+    def resolve_path(self, path: Union[str, Path]) -> Path:
+        return self.base_path / path
 
-    def import_dxf(self, dxf, table_name):
+    def import_dxf(self, dxf: Path, table_name: str):
         """Import the DXF into Postgres into the specified table name, overwriting the existing table."""
-        if not os.path.isfile(dxf):
+        if not dxf.is_file():
             raise Exception("Source DXF file %s does not exist" % dxf)
 
         self.log.info("Importing %s into PostGIS table %s...", dxf, table_name)

buildmap/plugins/noc/__init__.py

@@ -1,6 +1,5 @@
 import logging
 import time
-import os.path
 import pydotplus as pydot  # type: ignore
 import csv
 import html
@@ -635,21 +634,18 @@ def run(self):
             return
         self.log.info("Plan generated in %.2f seconds", time.time() - start)
 
-        out_path = os.path.join(
-            self.buildmap.resolve_path(self.buildmap.config["web_directory"]), "noc"
+        out_path = (
+            self.buildmap.resolve_path(self.buildmap.config["web_directory"]) / "noc"
         )
+        out_path.mkdir(parents=True, exist_ok=True)
 
-        if not os.path.isdir(out_path):
-            os.makedirs(out_path)
-
-        with open(os.path.join(out_path, "locations.csv"), "w") as locations_file:
+        with (out_path / "locations.csv").open("w") as locations_file:
             writer = csv.writer(locations_file)
             writer.writerow(["Location-Name"])
             for location in sorted(self.locations.values()):
                 writer.writerow([location.name])
 
-        # links.csv
-        with open(os.path.join(out_path, "links.csv"), "w") as links_file:
+        with (out_path / "links.csv").open("w") as links_file:
             writer = csv.writer(links_file)
             writer.writerow(
                 ["From-Location", "To-Location", "Type", "Subtype", "Length", "Cores"]
@@ -666,8 +662,7 @@ def run(self):
                     ]
                 )
 
-        # links-logical.csv
-        with open(os.path.join(out_path, "links-logical.csv"), "w") as links_file:
+        with (out_path / "links-logical.csv").open("w") as links_file:
             writer = csv.writer(links_file)
             writer.writerow(
                 ["From-Location", "To-Location", "Type", "Total-Length", "Couplers"]
@@ -683,24 +678,21 @@ def run(self):
                     ]
                 )
 
-        # warnings.txt
-        with open(os.path.join(out_path, "warnings.txt"), "w") as warnings_file:
+        with (out_path / "warnings.txt").open("w") as warnings_file:
             warnings_file.writelines("\n".join(self.warnings))
 
-        # stats.txt
-        with open(os.path.join(out_path, "stats.txt"), "w") as stats_file:
+        with (out_path / "stats.txt").open("w") as stats_file:
             self._write_stats(stats_file)
 
-        # noc-physical.pdf
         physical_dot = self.create_physical_dot()
         if not physical_dot:
             return
-        with open(os.path.join(out_path, "noc-physical.pdf"), "wb") as f:
+        with (out_path / "noc-physical.pdf").open("wb") as f:
             f.write(physical_dot.create_pdf())
 
         # noc-logical.pdf
         logical_dot = self.create_logical_dot()
         if not logical_dot:
             return
-        with open(os.path.join(out_path, "noc-logical.pdf"), "wb") as f:
+        with (out_path / "noc-logical.pdf").open("wb") as f:
             f.write(logical_dot.create_pdf())

buildmap/plugins/search/__init__.py

@@ -1,13 +1,15 @@
-import os.path
 import json
 from sqlalchemy import text
 from shapely import wkt
+from pathlib import Path
+from ...main import BuildMap
+from ...mapdb import MapDB
 
 
 class SearchPlugin(object):
     """Generate a JSON search index file to power JS search"""
 
-    def __init__(self, buildmap, _config, opts, db):
+    def __init__(self, buildmap: BuildMap, _config, opts, db: MapDB):
         self.db = db
         self.buildmap = buildmap
         self.opts = opts
@@ -51,14 +53,13 @@ def get_data(self):
     def run(self):
         data = self.get_data()
 
-        out_path = os.path.join(
-            self.buildmap.resolve_path(self.buildmap.config["web_directory"]), "search"
+        out_path = self.buildmap.resolve_path(
+            self.opts.get(
+                "output_path",
+                Path(self.buildmap.config["web_directory"]) / "search" / "search.json",
+            )
         )
+        out_path.parent.mkdir(parents=True, exist_ok=True)
 
-        try:
-            os.makedirs(out_path)
-        except FileExistsError:
-            pass
-
-        with open(os.path.join(out_path, "search.json"), "w") as f:
+        with out_path.open("w") as f:
             json.dump(data, f)

buildmap/plugins/translate/__init__.py

@@ -45,16 +45,15 @@ def insert_translations(self, table, lang, translations):
         )
 
         for orig, translated in translations.items():
-            for layer in self.opts.get("layers", []):
-                self.db.execute(
-                    text(
-                        "UPDATE {table} SET {col} = :translated WHERE text = :orig".format(
-                            table=table, col=trans_col
-                        )
-                    ),
-                    translated=translated,
-                    orig=orig,
-                )
+            self.db.execute(
+                text(
+                    "UPDATE {table} SET {col} = :translated WHERE text = :orig".format(
+                        table=table, col=trans_col
+                    )
+                ),
+                translated=translated,
+                orig=orig,
+            )
 
         self.buildmap.known_attributes[table].add(trans_col)