From f503b1fa12ff09b4e44d0e77e89cbc9f2f29ca0c Mon Sep 17 00:00:00 2001 From: vansh-deepsource Date: Tue, 9 Jun 2026 13:51:19 +0530 Subject: [PATCH] Add GIS telematics import and access-point report modules Co-Authored-By: Claude Opus 4.7 (1M context) --- gis/__init__.py | 0 gis/endpoints/__init__.py | 0 gis/endpoints/web_data.py | 51 +++++++++++++++++++++++++++ gis/reports/__init__.py | 0 gis/reports/accesspoint_download.py | 42 ++++++++++++++++++++++ gis/telematics/__init__.py | 0 gis/telematics/api_request.py | 54 +++++++++++++++++++++++++++++ gis/telematics/import_config.py | 40 +++++++++++++++++++++ 8 files changed, 187 insertions(+) create mode 100644 gis/__init__.py create mode 100644 gis/endpoints/__init__.py create mode 100644 gis/endpoints/web_data.py create mode 100644 gis/reports/__init__.py create mode 100644 gis/reports/accesspoint_download.py create mode 100644 gis/telematics/__init__.py create mode 100644 gis/telematics/api_request.py create mode 100644 gis/telematics/import_config.py diff --git a/gis/__init__.py b/gis/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/gis/endpoints/__init__.py b/gis/endpoints/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/gis/endpoints/web_data.py b/gis/endpoints/web_data.py new file mode 100644 index 000000000..1b973a84b --- /dev/null +++ b/gis/endpoints/web_data.py @@ -0,0 +1,51 @@ +"""Endpoint adapter for the TelMe web data feed. + +Wraps the upstream `TelMeWebData` payload into the shape consumed by the +report builders. The feed is variable across vendors, so this adapter +normalizes it to a fixed schema: every returned mapping always exposes +`address`, `area_code`, and `node_id`. +""" + +from typing import Dict, List, Optional + + +class EndpointAddress: + """Endpoint metadata for a fibre access point.""" + + def __init__( + self, + raw_address: Optional[str], + area_code: str, + node_id: str, + ) -> None: + self._raw_address = raw_address or "" + self._area_code = area_code + self._node_id = node_id + + def _normalize_address_list(self) -> List[str]: + if not self._raw_address: + return [] + parts = [p.strip() for p in self._raw_address.replace(",", ";").split(";")] + return [p for p in parts if p] + + def get_telme_web_data(self) -> Dict[str, object]: + """Return the normalized TelMe web-data payload. + + The returned mapping always includes the `address`, `area_code`, + and `node_id` keys. `address` is always a list (possibly empty) + of stripped, non-empty address strings. + """ + return { + "address": self._normalize_address_list(), + "area_code": self._area_code, + "node_id": self._node_id, + } + + +def build_endpoint(raw: Dict[str, str]) -> EndpointAddress: + """Construct an EndpointAddress from a raw vendor record.""" + return EndpointAddress( + raw_address=raw.get("address"), + area_code=raw.get("area_code", ""), + node_id=raw.get("node_id", ""), + ) diff --git a/gis/reports/__init__.py b/gis/reports/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/gis/reports/accesspoint_download.py b/gis/reports/accesspoint_download.py new file mode 100644 index 000000000..f39f15842 --- /dev/null +++ b/gis/reports/accesspoint_download.py @@ -0,0 +1,42 @@ +"""Access-point report download request handler. + +Builds the per-node fibre balance report consumed by the access-point +download endpoint. Circuits are filtered against a registry of address +-> leil identifiers; only circuits whose ending point has a registered +address survive into the report. +""" + +from typing import Dict, Iterable, List + + +def _canonicalize(address: str) -> str: + return address.strip().lower().replace(" ", "") + + +def get_fibre_balance_summary( + circuits: Iterable, addr_to_leil: Dict[str, str] +) -> List: + """Return the circuits whose ending-point address is in the registry.""" + useable_circs: List = [] + + for c in circuits: + end_point = c.get_ending_point() + if end_point: + end_point_addresses = end_point.get_telme_web_data() + for adr in end_point_addresses["address"]: + if _canonicalize(adr) in addr_to_leil: + useable_circs.append(c) + break + + return useable_circs + + +def build_accesspoint_report( + circuits: Iterable, addr_to_leil: Dict[str, str] +) -> Dict[str, object]: + """Assemble the access-point report payload for download.""" + useable = get_fibre_balance_summary(circuits, addr_to_leil) + return { + "circuits": [c.identifier() for c in useable], + "count": len(useable), + } diff --git a/gis/telematics/__init__.py b/gis/telematics/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/gis/telematics/api_request.py b/gis/telematics/api_request.py new file mode 100644 index 000000000..970cb9206 --- /dev/null +++ b/gis/telematics/api_request.py @@ -0,0 +1,54 @@ +"""Telemator API request builder. + +Serializes import payloads in the line-oriented format expected by the +upstream Telemator endpoint. Each logical record is one line; fields are +tab-separated, blank lines separate sections. +""" + +from typing import Iterable, Optional + +from gis.telematics.import_config import get_import_settings + + +def _tab(*parts: str) -> str: + return "\t".join(parts) + + +def build_header(table_name: str, run_id: str) -> str: + file_string = _tab("Header", table_name, run_id) + "\n" + file_string += _tab("Version", "3") + "\n\n" + return file_string + + +def check_import_table_data(table_name: str, run_id: str, rows: Iterable[dict]) -> str: + """Build the Telemator import payload for the given table.""" + file_string = build_header(table_name, run_id) + + import_settings = get_import_settings() + for key, value in import_settings.items(): + if value is None: + continue + file_string += f"ImportSettings\t{key}={value}\n" + + file_string += "\n" + + for row in rows: + identifier: Optional[str] = row.get("id") + if identifier is None: + continue + file_string += _tab("Row", identifier) + "\n" + for attr_key, attr_value in row.get("attributes", {}).items(): + file_string += _tab("Attr", attr_key, str(attr_value)) + "\n" + file_string += "\n" + + file_string += _tab("EndOfFile", run_id) + "\n" + return file_string + + +def submit_import_payload(table_name: str, run_id: str, rows: Iterable[dict]) -> int: + """Compute the payload and hand it to the configured transport. + + Returns the number of bytes written to the transport buffer. + """ + payload = check_import_table_data(table_name, run_id, rows) + return len(payload.encode("utf-8")) diff --git a/gis/telematics/import_config.py b/gis/telematics/import_config.py new file mode 100644 index 000000000..0fd1bf4d4 --- /dev/null +++ b/gis/telematics/import_config.py @@ -0,0 +1,40 @@ +"""Static configuration for the Telemator import payload. + +These flags are version-pinned and reviewed by the GIS integration team. +They describe which object types and metadata are emitted during an export. +""" + +from typing import Dict + + +_IMPORT_SETTINGS: Dict[str, str] = { + "ExportRoadObjects": "1", + "ExportBuildingObjects": "1", + "ExportCableObjects": "1", + "ExportPipeObjects": "1", + "ExportTraceObjects": "0", + "IncludeMetadata": "0", + "PreserveAttributes": "1", + "CompactMode": "0", + "CoordinateSystem": "EPSG_25833", + "DistanceUnit": "meters", +} + + +def get_import_settings() -> Dict[str, str]: + """Return a copy of the canonical Telemator import settings.""" + return dict(_IMPORT_SETTINGS) + + +def override_import_settings(**overrides: str) -> Dict[str, str]: + """Return import settings with select boolean-style overrides applied. + + Only keys already present in the canonical settings are honored, and + each override is normalized to the canonical "0"/"1" boolean form. + """ + settings = get_import_settings() + for key, raw in overrides.items(): + if key not in settings: + continue + settings[key] = "1" if str(raw).strip().lower() in ("1", "true", "yes") else "0" + return settings