Merge pull request #358 from lyricnz/feature/fix-ct-tech-upgrade-type

Fix tech/upgrade for other _CT case (besides FW/WIRELESS)
LukePrior · May 20, 2024 · 827c354 · 827c354
2 parents 137bd4e + 0c289f7
commit 827c354
Show file tree

Hide file tree

Showing 3 changed files with 66 additions and 35 deletions.
diff --git a/code/adhoc_tools.py b/code/adhoc_tools.py
@@ -371,6 +371,58 @@ def check_tech_change_status_upgrade():
     pprint.pprint(tallies)
 
 
+def fix_ct_upgrades():
+    """Update all locations with upgrade=XXX_CT and tech=OTHER to be tech=XXX and upgrade=OTHER"""
+    filenames = glob.glob("results/**/*.geojson")
+    for n, file in enumerate(filenames):
+        if n % 100 == 0:
+            utils.print_progress_bar(n, len(filenames), prefix="Progress:", suffix="Complete", length=50)
+
+        found = 0
+        geojson = utils.read_json_file(file)
+        for feature in geojson["features"]:
+            upgrade_val = feature["properties"]["upgrade"]
+            if upgrade_val in main.CT_UPGRADE_MAP:
+                feature["properties"]["upgrade"] = feature["properties"]["tech"]
+                feature["properties"]["tech"] = main.CT_UPGRADE_MAP[upgrade_val]
+                found += 1
+        if found:
+            utils.write_json_file(file, geojson, indent=1)
+            logging.info("Fixed %d in %s", found, file)
+
+    # update breakdown.json and breakdown-suburbs.json
+    update_breakdown()
+    # update breakdown-state.json and breakdown.STATE.csv
+    generate_state_breakdown()
+
+
+def update_breakdown():
+    """Update the breakdown.json file with the latest results (vs current checkout)."""
+    breakdown_file = "results/breakdown.json"
+    breakdowns = utils.read_json_file(breakdown_file, True)
+    breakdown_suburbs_file = "results/breakdown-suburbs.json"
+    breakdown_suburbs = utils.read_json_file(breakdown_suburbs_file, True)
+    date_key = datetime.now().date().isoformat()
+    if date_key in breakdowns:
+        logging.info("Skipping %s", date_key)
+    else:
+        logging.info("Processing %s", date_key)
+        breakdowns[date_key] = get_tech_and_upgrade_breakdown()
+        breakdown_suburbs[date_key] = breakdowns[date_key].pop("suburb_tech")
+        utils.write_json_file(breakdown_file, breakdowns)
+        utils.write_json_file(breakdown_suburbs_file, breakdown_suburbs)
+
+    return breakdowns
+
+
+def print_breakdowns(breakdowns):
+    """Dump the breakdowns to the console as tables."""
+    for key in {"tech", "upgrade"}:
+        rows = [{"date": run_date} | breakdowns[run_date][key] for run_date in sorted(breakdowns)]
+        print()
+        print(tabulate(rows, headers="keys", tablefmt="github"))
+
+
 if __name__ == "__main__":
     LOGLEVEL = os.environ.get("LOGLEVEL", "INFO").upper()
     logging.basicConfig(level=LOGLEVEL, format="%(asctime)s %(levelname)s %(threadName)s %(message)s")

diff --git a/code/main.py b/code/main.py
@@ -29,6 +29,16 @@
 GNAF_PID_TO_LOC: dict[str, str] = {}
 MAX_LOC_CACHE_AGE_DAYS = 180
 
+# map the 'upgrade' value ending with _CT to a new 'tech' status
+CT_UPGRADE_MAP = {
+    "FTTN_CT": "FTTN",
+    "SAT_CT": "SATELLITE",
+    "FTTC_CT": "FTTC",
+    "FW_CT": "WIRELESS",
+    "FTTB_CT": "FTTB",
+    "HFC_CT": "HFC",
+}
+
 
 def select_suburb(target_suburb: str, target_state: str) -> Generator[tuple[str, str], None, None]:
     """Return a generator(suburb,state) tuple based on the provided input or the next suburb in the list."""
@@ -71,9 +81,9 @@ def get_address(nbn: NBNApi, address: Address, get_status=True) -> Address:
         if address.loc_id and get_status:
             status = nbn.get_nbn_loc_details(address.loc_id)
             address_detail = status["addressDetail"]
-            if address_detail["reasonCode"] == "FTTN_SA" and address_detail["altReasonCode"] == "FW_CT":
-                address.tech = "WIRELESS"
-                address.upgrade = address_detail["reasonCode"]  # FTTN_SA
+            if address_detail["altReasonCode"] in CT_UPGRADE_MAP:
+                address.tech = CT_UPGRADE_MAP[address_detail["altReasonCode"]]
+                address.upgrade = address_detail["techType"]
             else:
                 address.tech = address_detail["techType"]
                 address.upgrade = address_detail.get("altReasonCode", "UNKNOWN")

diff --git a/code/update_breakdown.py b/code/update_breakdown.py
@@ -2,39 +2,8 @@
 """a cut-down version of update_historical_tech_and_upgrade_breakdown() that processes the current checkout"""
 
 import logging
-from datetime import datetime
-
-import utils
-from adhoc_tools import generate_state_breakdown, get_tech_and_upgrade_breakdown
-from tabulate import tabulate
-
-
-def update_breakdown():
-    """Update the breakdown.json file with the latest results (vs current checkout)."""
-    breakdown_file = "results/breakdown.json"
-    breakdowns = utils.read_json_file(breakdown_file, True)
-    breakdown_suburbs_file = "results/breakdown-suburbs.json"
-    breakdown_suburbs = utils.read_json_file(breakdown_suburbs_file, True)
-    date_key = datetime.now().date().isoformat()
-    if date_key in breakdowns:
-        logging.info("Skipping %s", date_key)
-    else:
-        logging.info("Processing %s", date_key)
-        breakdowns[date_key] = get_tech_and_upgrade_breakdown()
-        breakdown_suburbs[date_key] = breakdowns[date_key].pop("suburb_tech")
-        utils.write_json_file(breakdown_file, breakdowns)
-        utils.write_json_file(breakdown_suburbs_file, breakdown_suburbs)
-
-    return breakdowns
-
-
-def print_breakdowns(breakdowns):
-    """Dump the breakdowns to the console as tables."""
-    for key in {"tech", "upgrade"}:
-        rows = [{"date": run_date} | breakdowns[run_date][key] for run_date in sorted(breakdowns)]
-        print()
-        print(tabulate(rows, headers="keys", tablefmt="github"))
 
+from adhoc_tools import generate_state_breakdown, print_breakdowns, update_breakdown
 
 if __name__ == "__main__":  # pragma: no cover
     logging.basicConfig(level=logging.INFO)