Finished feature NGM-23-design-input-data.

mapping_provider/api/map.py

@@ -146,3 +146,17 @@ def get_trunks() -> services.ServiceList:
     handler for /trunks, same as /services/IP TRUNK
     """
     return get_services(service_type='IP TRUNK')
+
+
+@router.get("/map-info")
+@router.get("/map-info/{service_type}")
+def get_map_info(service_type: str | None = None) -> services.MapInfo:
+    """
+    handler for /map-info
+    """
+    pop_list = get_pops()
+    pops_dict = {_p.name: _p.abbreviation for _p in pop_list.pops}
+    return_value = services.build_map_info(pop_abbrevs=pops_dict, service_type=service_type)
+    if not return_value.nodes:
+        raise HTTPException(status_code=404, detail=f'unrecognized service type: {service_type}')
+    return return_value

mapping_provider/backends/services.py

@@ -38,6 +38,11 @@ class ServiceList(BaseModel):
     services: list[Service]
 
 
+class MapInfo(BaseModel):
+    nodes: list[str]
+    edges: list[str]
+
+
 def endpoint_equipment(endpoint: dict[str, Any]) -> str:
     """
     convert the correlator router hostname or optical equipment name
@@ -145,3 +150,45 @@ def build_service_info_list(service_type: str | None = None) -> ServiceList:
     return a list of mappable info about all operational services
     """  
     return ServiceList(services=list(_services(service_type)))
+
+
+def _service_label_in_map(service: Service) -> str | None:
+    """
+    return a label for the service in the map
+    """
+    if not service.pops:
+        return None
+
+    pops = sorted(service.pops)
+    label_components = pops + [service.scid[:8].upper()]
+    return '-'.join(label_components)
+
+
+def build_map_info(pop_abbrevs: dict[str, str], service_type: str | None = None) -> MapInfo:
+    """
+    return a list of mappable info about operational services
+    """
+    nodes = set()
+    edges = set()
+
+    def _get_service_pops(svc: Service) -> Generator[str]:
+        for _p in svc.pops:
+            abbrev = pop_abbrevs.get(_p)
+            if abbrev:
+                yield abbrev
+
+    for _s in _services(service_type):
+
+        _pops = set(_get_service_pops(_s))
+        if not _pops:
+            logger.warning(f'service {_s.sid} has no POPs')
+            continue
+
+        nodes |= set(_pops)
+
+        _lc = list(_pops) + [_s.scid[:8].upper()]
+        edges.add('-'.join(_lc))
+
+    return MapInfo(
+        nodes=sorted(nodes),
+        edges=sorted(edges))

test/test_map_endpoints.py

@@ -4,20 +4,22 @@ import pytest
 import responses
 
 from mapping_provider.api.map import EquipmentList, PopList
-from mapping_provider.backends.services import ServiceList
+from mapping_provider.backends.services import MapInfo, ServiceList
 
 from .common import load_test_data
 
 
-@responses.activate
-def test_get_pops(client):
-
+def _add_pops_response():
     responses.add(
         method=responses.GET,
         url=re.compile(r'.*/map/pops$'),
         json=load_test_data('inprov-pops.json')
     )
 
+@responses.activate
+def test_get_pops(client):
+    _add_pops_response()
+
     rv = client.get("/map/pops")
     assert rv.status_code == 200
     pop_list = PopList.model_validate(rv.json())
@@ -86,6 +88,7 @@ def test_get_all_services(client):
     service_list = ServiceList.model_validate(rv.json())
     assert service_list.services, 'test data should not be empty'
 
+
 @responses.activate
 def test_get_trunks(client):
     rv = client.get("/map/trunks")
@@ -95,3 +98,20 @@ def test_get_trunks(client):
     assert all(s.type == 'IP TRUNK' for s in service_list.services)
 
 
+@responses.activate
+def test_get_trunk_map_info(client):
+    _add_pops_response()
+
+    rv = client.get("/map/map-info/IP TRUNK")
+    assert rv.status_code == 200
+    map_info = MapInfo.model_validate(rv.json())
+    assert map_info.nodes, 'test data should not be empty'
+    assert map_info.edges, 'test data should not be empty'
+
+
+@responses.activate
+def test_get_unknown_service_type_map_info(client):
+    _add_pops_response()
+    rv = client.get("/map/map-info/BOGUS_SERVICE_TYPE")
+    assert rv.status_code == 404
+