Add GIS telematics import and access-point report modules

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", ""),
+    )

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),
+    }

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"))

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