feat: externalized depthmap configuration #3

Makefile

@@ -12,6 +12,7 @@ install:
 
 install-mac:
 	brew install gdal proj geos --force
+	export DYLD_LIBRARY_PATH=/opt/homebrew/opt/geos/lib/
 	install
 
 test: test/*.py

config.py

@@ -1,10 +1,66 @@
+import enum
 from dataclasses import dataclass
 from logging import Logger
 from pathlib import Path
+from typing import List, Union
 from uuid import uuid4
 
-from download import create_workdir
+import jsonpickle
+
 from logger import configure_logger
+from utils import create_workdir
+
+Num = Union[int, str]
+
+
+@dataclass
+class AxialAnalysisConfiguration:
+    radii: List[Num]
+
+
+default_axial = AxialAnalysisConfiguration(radii=[3, "n"])
+
+
+class SegmentAnalysisType(enum.Enum):
+    ANGULAR_TULIP = ("tulip",)  # (Angular Tulip - Faster)
+    ANGULAR_TULIP_CHOICE = ("tulip -sic",)  # (Angular Tulip - Faster include choice )
+    ANGULAR_FULL = ("angular",)  # (Angular Full - Slower)
+    TOPOLOGICAL = ("topological",)
+    METRIC = "metric"
+
+
+class SegmentAnalysisTulipRadiusType(enum.Enum):
+    STEPS = ("steps",)
+    METRIC = ("metric",)
+    ANGULAR = ("angular",)
+
+
+@dataclass
+class SegmentAnalysisConfiguration:
+    type: SegmentAnalysisType
+    radius_type: SegmentAnalysisTulipRadiusType
+    # (4 to 1024, 1024 approximates full angular)
+    tulip_bins: int
+    radii: List[Num]
+
+
+default_segment = SegmentAnalysisConfiguration(
+    type=SegmentAnalysisType.ANGULAR_TULIP,
+    radius_type=SegmentAnalysisTulipRadiusType.METRIC,
+    tulip_bins=1024,
+    radii=[300, 1000, "n"],
+)
+
+
+@dataclass
+class AnalysisConfiguration:
+    axial: AxialAnalysisConfiguration
+    segment: SegmentAnalysisConfiguration
+
+
+DEFAULT_ANALYSIS_CONFIG = AnalysisConfiguration(
+    axial=default_axial, segment=default_segment
+)
 
 
 @dataclass
@@ -13,12 +69,22 @@ class Configuration:
     operation_id: str
     workdir: Path
     log: Logger
+    analysis: AnalysisConfiguration
 
 
 def configure(place: str, root_workdir: str = "./downloads/") -> Configuration:
     operation_id = place.split(",")[0] + "-" + uuid4().hex[:8]
     workdir = create_workdir(f"{root_workdir}/{operation_id}")
     log = configure_logger(workdir)
     return Configuration(
-        place=place, operation_id=operation_id, workdir=workdir, log=log
+        place=place,
+        operation_id=operation_id,
+        workdir=workdir,
+        log=log,
+        analysis=DEFAULT_ANALYSIS_CONFIG,
     )
+
+
+def dump_config_file(config: Configuration) -> None:
+    with open(Path.joinpath(config.workdir, "config.json"), "w", encoding="utf-8") as f:
+        f.write(jsonpickle.dumps(config))  # type: ignore

depthmap.py

@@ -9,6 +9,11 @@
 
 from convert import mif_to_shp
 from logger import default_logger
+from config import (
+    AnalysisConfiguration,
+    AxialAnalysisConfiguration,
+    SegmentAnalysisConfiguration,
+)
 
 log = default_logger()
 
@@ -59,7 +64,9 @@ async def run(cmd: str, description="Running command"):
     )
 
 
-async def axial(graph_file: str, depthmapx: DepthmapX):
+async def axial(
+    graph_file: str, depthmapx: DepthmapX, config: AxialAnalysisConfiguration
+):
     base_file, _ = path.splitext(graph_file)
     axial_map_file = f"{base_file}-axial-map.graph"
     await run(
@@ -68,7 +75,7 @@ async def axial(graph_file: str, depthmapx: DepthmapX):
     )
     axial_analysis_file = f"{base_file}-axial-analysis.graph"
     await run(
-        f"{depthmapx.executable} -m AXIAL -f {axial_map_file} -o {axial_analysis_file} -p -xa 3,n",
+        f"{depthmapx.executable} -m AXIAL -f {axial_map_file} -o {axial_analysis_file} -p -xa {','.join(str(radius) for radius in config.radii)}",
         "Performing axial analysis",
     )
     axial_shapefile_mif = f"{base_file}.axial.mif"
@@ -79,7 +86,9 @@ async def axial(graph_file: str, depthmapx: DepthmapX):
     return mif_to_shp(axial_shapefile_mif)
 
 
-async def segment(graph_file: str, depthmapx: DepthmapX):
+async def segment(
+    graph_file: str, depthmapx: DepthmapX, config: SegmentAnalysisConfiguration
+):
     base_file, _ = path.splitext(graph_file)
     log.info("This operation may take longer than 25 minutes to complete!!!")
     segment_map_file = f"{base_file}-segment-map.graph"
@@ -89,7 +98,7 @@ async def segment(graph_file: str, depthmapx: DepthmapX):
     )
     segment_analysis_file = f"{base_file}-segment-analysis.graph"
     await run(
-        f"{depthmapx.executable} -m SEGMENT -f {segment_map_file} -o {segment_analysis_file} -p  -st tulip -sic -srt metric -stb 1024 -sr 300,1000,n",
+        f"{depthmapx.executable} -m SEGMENT -f {segment_map_file} -o {segment_analysis_file} -p  -st {config.type} -srt {config.radius_type} -stb {config.tulip_bins} -sr {','.join(str(radius) for radius in config.radii)}",
         "Performing segment analysis",
     )
     segment_shapefile_mif = f"{base_file}.segment.mif"
@@ -100,7 +109,7 @@ async def segment(graph_file: str, depthmapx: DepthmapX):
     return mif_to_shp(segment_shapefile_mif)
 
 
-async def analyse(dxfFile: str):
+async def analyse(dxfFile: str, analysis_config: AnalysisConfiguration):
     base_file, _ = path.splitext(dxfFile)
     graph_file = base_file + ".graph"
     depthmapx = depthmapx_factory()
@@ -114,5 +123,6 @@ async def analyse(dxfFile: str):
         "Transform dxf to shape file",
     )
     return await asyncio.gather(
-        axial(graph_file, depthmapx), segment(graph_file, depthmapx)
+        axial(graph_file, depthmapx, analysis_config.axial),
+        segment(graph_file, depthmapx, analysis_config.segment),
     )

download.py

@@ -19,11 +19,6 @@
 log = default_logger()
 
 
-def create_workdir(workdir: str = "./downloads") -> Path:
-    path = Path(workdir)
-    path.mkdir(exist_ok=True, parents=True)
-    return path
-
 
 def download(place: str, operation_id=uuid.uuid4(), workdir="./downloads") -> str:
     log.info(f"Downloading map for: {place} ...")

main.py

@@ -13,14 +13,15 @@
 async def process(cfg: config.Configuration) -> None:
     start = time()
     utils.create_status_file(cfg.workdir, utils.Status.WORKING)
+    config.dump_config_file(cfg)
     cfg.log.info(f"Starting operation {cfg.operation_id}")
     cfg.log.info("Downloading osm files...")
     map = download(place, cfg.operation_id, cfg.workdir)
     cfg.log.info(
         f"Downloading osm files took {datetime.timedelta(seconds=(time() - start))}"
     )
     dxf = osm_to_dxf(map)
-    axial_analysis, segment_analysis = await analyse(dxf)
+    axial_analysis, segment_analysis = await analyse(dxf, cfg.analysis)
     cfg.log.info("Exported axial files: ", axial_analysis)
     cfg.log.info("Exported segment files: ", segment_analysis)
     cfg.log.info(

requirements.txt

@@ -1,3 +1,4 @@
-osmnx==1.1.2
-geopandas==0.10.2
-Fiona==1.8.21
+osmnx==1.2.1
+geopandas==0.11.0
+Fiona==1.8.21
+jsonpickle==2.2.0

test/test_download.py

@@ -1,26 +1,7 @@
 import json
 import os
-from pathlib import Path
 
-from download import (
-    create_workdir,
-    download_administrative_geojson,
-    download_drive_graph_from_place,
-)
-
-
-def test_create_workdir_default():
-    test_dir = create_workdir()
-    assert test_dir.exists()
-    assert test_dir.is_dir()
-    assert test_dir.stem == "downloads"
-
-
-def test_create_workdir_new(tmpdir):
-    test_dir = create_workdir(Path(tmpdir).joinpath("downloads"))
-    assert test_dir.exists()
-    assert test_dir.is_dir()
-    assert test_dir.stem == "downloads"
+from download import download_administrative_geojson, download_drive_graph_from_place
 
 
 def test_download_administrative_geojson():

test/test_utils.py

@@ -1,7 +1,20 @@
 import os
 import pathlib
 import py
-from utils import Status, create_status_file
+from utils import Status, create_status_file, create_workdir
+
+def test_create_workdir_default():
+    test_dir = create_workdir()
+    assert test_dir.exists()
+    assert test_dir.is_dir()
+    assert test_dir.stem == "downloads"
+
+
+def test_create_workdir_new(tmpdir):
+    test_dir = create_workdir(pathlib.Path(tmpdir).joinpath("downloads"))  # type: ignore
+    assert test_dir.exists()
+    assert test_dir.is_dir()
+    assert test_dir.stem == "downloads"
 
 
 def check_status_file_exists(tmpdir: py.path.local, status: Status):

utils.py

@@ -13,3 +13,9 @@ def create_status_file(directory: Path, status: Status) -> None:
         # delete WORKING status so that the status can only be in one of the final states
         Path.unlink(directory.joinpath(Status.WORKING.name), missing_ok=True)
         Path.touch(directory.joinpath(status.name))
+
+
+def create_workdir(workdir: str = "./downloads") -> Path:
+    path = Path(workdir)
+    path.mkdir(exist_ok=True, parents=True)
+    return path