Really unsimplify this time

nshmdb/scripts/nshm_db_generator.py

@@ -23,6 +23,8 @@
     python generate_nshm2022_data.py data/cru_solutions.zip output/nshm2022.sqlite
 """
 
+import copy
+import difflib
 import zipfile
 from pathlib import Path
 from typing import Annotated
@@ -48,6 +50,57 @@
 MFDS_PATH = Path("ruptures") / "sub_seismo_on_fault_mfds.csv"
 
 
+def print_array_diff(arr1: list, arr2: list) -> None:
+    """Print the differences between two arrays in a readable format.
+
+    Parameters
+    ----------
+    arr1 : list
+        The first array to compare.
+    arr2 : list
+        The second array to compare.
+
+    Returns
+    -------
+    None
+
+    Example
+    -------
+    >>> arr1 = ["a", "b", "c"]
+    >>> arr2 = ["a", "c"]
+    >>> print_array_diff(arr1, arr2)
+        Old: a  b  c
+        New: a     c
+    """
+    seq_match = difflib.SequenceMatcher(a=arr1, b=arr2)
+    new_line = []
+    old_line = []
+
+    for tag, i1, i2, j1, j2 in seq_match.get_opcodes():
+        if tag == "equal":
+            for old, new in zip(arr1[i1:i2], arr2[j1:j2]):
+                new_line.append(f"{new}")
+                old_line.append(f"{old}")
+        elif tag == "replace":
+            max_len = max(i2 - i1, j2 - j1)
+            for k in range(max_len):
+                old = arr1[i1 + k] if k < i2 - i1 else " " * len(str(arr2[j1 + k]))
+                new = arr2[j1 + k] if k < j2 - j1 else " " * len(str(arr1[i1 + k]))
+                new_line.append(f"{new}")
+                old_line.append(f"{old}")
+        elif tag == "delete":
+            for old in arr1[i1:i2]:
+                new_line.append(" " * len(f"{old}"))
+                old_line.append(f"{old}")
+        elif tag == "insert":
+            for new in arr2[j1:j2]:
+                new_line.append(f"{new}")
+                old_line.append(" " * len(f"{new}"))
+
+    print("Old: " + "  ".join(old_line))
+    print("New: " + "  ".join(new_line))
+
+
 def extract_faults_from_info(
     fault_info_list: FeatureCollection,
 ) -> dict[str, Fault]:
@@ -72,11 +125,20 @@ def extract_faults_from_info(
                 np.array(list(geojson.utils.coords(fault_feature)))[:, ::-1]
             )
         )
+        fault_trace_old = copy.deepcopy(fault_trace)
+        fault_trace = shapely.remove_repeated_points(fault_trace, 0)
         trace_coords = np.array(fault_trace.coords)
         name = fault_feature.properties["FaultName"]
         bottom = fault_feature.properties["LowDepth"]
         dip_dir = fault_feature.properties["DipDir"]
         dip = fault_feature.properties["DipDeg"]
+
+        if not shapely.equals_exact(fault_trace, fault_trace_old):
+            old_trace = list(fault_trace_old.coords)
+            new_trace = list(fault_trace.coords)
+            print(f"Warning: Fault trace for {name} was altered.")
+            print_array_diff(old_trace, new_trace)
+
         planes = []
         for j in range(len(trace_coords) - 1):
             top_left = trace_coords[j]
@@ -240,7 +302,3 @@ def main(
                 rupture_fault_join_df.to_sql(
                     "rupture_faults", conn, index=False, if_exists="append"
                 )
-
-
-if __name__ == "__main__":
-    app()

tests/test_nshmdb_generator.py

@@ -1,9 +1,20 @@
 from pathlib import Path
 
+import pytest
+
 from nshmdb.nshmdb import NSHMDB
 from nshmdb.scripts import nshm_db_generator
 
-CRU_FAULT_SOLUTIONS = Path("tests") / "CRU_fault_system_solution.zip"
+CRU_FAULT_SOLUTIONS = Path("tests") / "CRU_fault_system_solution_small.zip"
+
+
+def test_sequence_differ(capsys: pytest.CaptureFixture):
+    nshm_db_generator.print_array_diff(
+        [1, 2, 3, 4],
+        [1, 2, 5, 6, 7],
+    )
+    captured = capsys.readouterr()
+    assert captured.out == ("Old: 1  2  3  4   \n" + "New: 1  2  5  6  7\n")
 
 
 def test_nshmdb_generator(tmp_path: Path):
@@ -17,3 +28,13 @@ def test_nshmdb_generator(tmp_path: Path):
     assert rupture.area == 1090332700
     assert rupture.length == 34817.69
     assert set(rupture.faults) == {"Acton"}
+
+
+CRU_FAULT_SOLUTIONS_FULL = Path("tests") / "CRU_fault_system_solution.zip"
+
+
+def test_nshmdb_generator_runs(tmp_path: Path):
+    nshmdb_path = tmp_path / "nhsmdb.db"
+    nshm_db_generator.main(CRU_FAULT_SOLUTIONS_FULL, nshmdb_path)
+    db = NSHMDB(nshmdb_path)
+    assert db is not None