import os
import sys
-
from network_generation.base import NetworkGenerator
from network_generation.parameter_validator import ParameterValidator
from network_generation.view.network_viewer import NetworkViewer
def save_viewer_output(
- viewer: NetworkViewer,
- filename: str,
- task: dict[str, str] | dict[str, int],
- method_name: str,
-) -> None:
+ viewer: NetworkViewer, filename: str,
+ task: dict[str, str] | dict[str, int],
+ method_name: str) -> None:
"""
Save the output using the specified method of NetworkViewer.
"""
`python -m network_generation`.
"""
validator = ParameterValidator(sys.argv)
+
if not validator.is_valid():
print(validator.error_message())
return
-# Copyright 2023 highstreet technologies USA CORP.
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# returns a list of geo-locations as cell polygon
def get_cell_polygons(self) -> list[GeoLocation]:
- points: list[Point] = Hexagon.polygon_corners(
- self.layout, self.position
- )
+ points: list[Point] = Hexagon.polygon_corners(self.layout, self.position)
method = (
- self.parent.parent.parent.parent.parent.parent
- .geo_location.point_to_geo_location
+ self.parent.parent.parent.parent.parent.parent.geo_location.point_to_geo_location
)
geo_locations: list[GeoLocation] = list(map(method, points))
index: int = 1 + int(self._azimuth / self._cell_angle)
(points[p1].x + points[p2].x) / 2,
(points[p1].y + points[p2].y) / 2,
)
- intersect_gl1: GeoLocation = network_center.point_to_geo_location(
- intersect1
- )
+ intersect_gl1: GeoLocation = network_center.point_to_geo_location(intersect1)
p3: int = (2 * index + 3) % 6
p4: int = (2 * index + 4) % 6
(points[p3].x + points[p4].x) / 2,
(points[p3].y + points[p4].y) / 2,
)
- intersect_gl2: GeoLocation = network_center.point_to_geo_location(
- intersect2
- )
+ intersect_gl2: GeoLocation = network_center.point_to_geo_location(intersect2)
tower = self.geo_location
scaled_cell_polygon: list[GeoLocation] = []
arc: float = self.azimuth * math.pi / 180
- meterToDegree: float = (
- 2 * math.pi * GeoLocation().equatorialRadius / 360
- )
+ meterToDegree: float = 2 * math.pi * GeoLocation().equatorialRadius / 360
centerX: float = self.layout.size.x * 0.5 * math.sin(arc)
centerY: float = self.layout.size.y * 0.5 * math.cos(arc)
cell_center: GeoLocation = GeoLocation(
def to_directory(self, parent_dir: str) -> None:
pass
+ def get_geojson_href(self) -> str:
+ return f"https://{self.network.host}/area/o-ran-network.geo.json/features?properties.name={self.name}"
+
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ cell_polygon: list[GeoLocation] = self.get_cell_polygons()
+ coordinates: list = []
+ for gl in cell_polygon:
+ cord: list[float] = [gl.longitude, gl.latitude]
+ coordinates.append(cord)
+ return [
+ {
+ "type": "Feature",
+ "properties": {
+ "type": "PropertiesOdu",
+ "name": self.name,
+ "uuid": self.id,
+ "function": "o-ran-sc-network:cell",
+ "newRadioCellGlobalIdentity": {
+ "publicLandMobileNetworkIdentifier": "123-45",
+ "newRadioCellIdentity": "0x0FFFFFFFFF",
+ }
+ },
+ "geometry": {
+ "type": "Polygon",
+ "coordinates": [coordinates],
+ },
+ }
+ ]
+
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ # a cell is not a node it is a Termination Point
+ result: list[dict[str, Any]] = [] # super().to_topology_nodes()
+ return result
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ # as a cell is not a node, it does not have links
+ result: list[dict[str, Any]] = [] # super().to_topology_links()
+ return result
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return super().to_tmf633_service_candidate_references()
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return super().to_tmf633_service_candidates()
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return super().to_tmf633_service_specifications()
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_candidate_references()
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_specification_references()
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_candidates()
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_specifications()
+
def add_teiv_data_entities(
self,
entity_type: str = "o-ran-smo-teiv-ran:NRCellDU",
-# Copyright 2024 highstreet technologies USA CORP.
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
"to_topology_links",
)
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf686_vertex,
+ "to_tmf686_vertex",
+ )
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf686_edge,
+ "to_tmf686_edge",
+ )
+
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_geojson_feature,
+ "to_geojson_feature",
+ )
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf633_service_candidate_references,
+ "to_tmf633_service_candidate_references",
+ )
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf633_service_candidates,
+ "to_tmf633_service_candidates",
+ )
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf633_service_specifications,
+ "to_tmf633_service_specifications",
+ )
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf634_resource_candidate_references,
+ "to_tmf634_resource_candidate_references",
+ )
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf634_resource_specification_references,
+ "to_tmf634_resource_specification_references",
+ )
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf634_resource_candidates,
+ "to_tmf634_resource_candidates"
+ )
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_tower_references(
+ super().to_tmf634_resource_specifications,
+ "to_tmf634_resource_specifications",
+ )
+
def _extend_teiv_data_with_tower_references(
self: Any, tower_method_name: str
) -> dict[str, list[dict[str, Any]]]:
-# Copyright 2024 highstreet technologies
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
)
return result
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf686_vertex,
+ "to_tmf686_vertex",
+ )
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf686_edge,
+ "to_tmf686_edge",
+ )
+
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_geojson_feature,
+ "to_geojson_feature",
+ )
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf633_service_candidate_references,
+ "to_tmf633_service_candidate_references",
+ )
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf633_service_candidates,
+ "to_tmf633_service_candidates",
+ )
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf633_service_specifications,
+ "to_tmf633_service_specifications",
+ )
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf634_resource_candidate_references,
+ "to_tmf634_resource_candidate_references",
+ )
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf634_resource_specification_references,
+ "to_tmf634_resource_specification_references",
+ )
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf634_resource_candidates,
+ "to_tmf634_resource_candidates"
+ )
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cloud_du_references(
+ super().to_tmf634_resource_specifications,
+ "to_tmf634_resource_specifications",
+ )
+
def _extend_teiv_data_with_o_ran_cloud_du_references(
self: Any,
teiv_data: dict[str, list[dict[str, Any]]],
super().add_teiv_data_relationships(id, aside, bside, rel_type),
"add_teiv_data_relationships",
)
+
def __init__(
self,
data: dict[str, Any] = cast(dict[str, Any], default_value),
- **kwargs: dict[str, Any],
+ **kwargs: dict[str, Any]
) -> None:
o_ran_du_data: IORanDu = self._to_o_ran_du_data(data)
super().__init__(cast(dict[str, Any], o_ran_du_data), **kwargs)
if not os.path.exists(path):
os.mkdir(path)
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return super().to_geojson_feature()
+
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = super().to_tmf686_vertex()
+ return result
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = super().to_tmf686_edge()
+ for interface in ["e2", "o1"]:
+ link_id: str = "".join(
+ [interface, ":", self.name, "<->", self.parent.name]
+ )
+ source_tp: str = "-".join([self.name, interface.upper()])
+ dest_tp: str = "-".join([self.parent.name, interface.upper()])
+ result.append(
+ {
+ "id": link_id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}/edge/{link_id}",
+ "bidirectional": True,
+ "description": "Description of an edge object",
+ "name": self.name,
+ "edgeCharacteristic": [
+ # {
+ # "id": "string",
+ # "name": "string",
+ # "valueType": "string",
+ # "characteristicRelationship": [
+ # {
+ # "id": "string",
+ # "href": "string",
+ # "relationshipType": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ # ],
+ # "value": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ ],
+ "edgeSpecification": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "version": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "entity": {
+ "id": "indigo",
+ "href": "https://indigo.cosmos-lab.org",
+ "name": "INDIGO",
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "EntityRef",
+ "@referredType": "Individual",
+ },
+ "graph": {
+ "id": self.network.id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}",
+ "name": self.network.name,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "GraphRef",
+ "@referredType": "Graph",
+ },
+ "subGraph": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "vertex": [
+ {
+ "id": source_tp,
+ "href": (
+ f'https://{self.host}/tmf-api/topologyDiscovery'
+ f'/v4/graph/{self.network.id}/vertex/{source_tp}'
+ ),
+ "name": source_tp,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "VertexRef",
+ "@referredType": "Vertex",
+ },
+ {
+ "id": dest_tp,
+ "href": (
+ f'https://{self.host}/tmf-api/topologyDiscovery'
+ f'/v4/graph/{self.network.id}/vertex/{dest_tp}'
+ ),
+ "name": dest_tp,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "VertexRef",
+ "@referredType": "Vertex",
+ },
+ ],
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "Edge",
+ }
+ )
+ return result
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return super().to_tmf633_service_candidate_references()
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return super().to_tmf633_service_candidates()
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return super().to_tmf633_service_specifications()
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_candidate_references()
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_specification_references()
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_candidates()
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return super().to_tmf634_resource_specifications()
+
def add_teiv_data_entities(
self,
entity_type: str = "o-ran-smo-teiv-ran:ODUFunction",
-# Copyright 2024 highstreet technologies
+# Copyright 2024 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
)
return result
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf686_vertex, "to_tmf686_vertex"
+ )
+
+
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_geojson_feature, "to_geojson_feature"
+ )
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf686_edge, "to_tmf686_edge"
+ )
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf633_service_candidate_references,
+ "to_tmf633_service_candidate_references",
+ )
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf633_service_candidates,
+ "to_tmf633_service_candidates"
+ )
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf633_service_specifications,
+ "to_tmf633_service_specifications"
+ )
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf634_resource_candidate_references,
+ "to_tmf634_resource_candidate_references",
+ )
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf634_resource_specification_references,
+ "to_tmf634_resource_specification_references",
+ )
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf634_resource_candidates,
+ "to_tmf634_resource_candidates"
+ )
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_cu_references(
+ super().to_tmf634_resource_specifications,
+ "to_tmf634_resource_specifications",
+ )
+
def _extend_teiv_data_with_o_ran_cu_references(
self: Any,
teiv_data: dict[str, list[dict[str, Any]]],
"""
Module for a class representing a O-RAN Network
"""
-
-import os
import uuid
import xml.etree.ElementTree as ET
+import os
+from typing import Any, Dict, cast
from datetime import datetime, timezone
-from typing import Any, cast, Dict, List, Union
import network_generation.model.python.hexagon as Hexagon
-from network_generation.model.python.geo_location import GeoLocation, IGeoLocation
+from network_generation.model.python.geo_location import (
+ GeoLocation,
+ IGeoLocation,
+)
from network_generation.model.python.hexagon import Layout
-from network_generation.model.python.o_ran_object import IORanObject, ORanObject
+from network_generation.model.python.o_ran_object import (
+ IORanObject,
+ ORanObject,
+)
+from network_generation.model.python.o_ran_ru import ORanRu
from network_generation.model.python.o_ran_smo import ORanSmo
-from network_generation.model.python.o_ran_spiral_radius_profile import SpiralRadiusProfile
+from network_generation.model.python.o_ran_spiral_radius_profile import (
+ SpiralRadiusProfile,
+)
from network_generation.model.python.point import Point
self.name = str(configuration.get("name", "WhiteNetwork"))
self.operationalState = str(configuration.get("operationalState", "disabled"))
- self._center: IGeoLocation = cast(
- IGeoLocation, configuration["center"]
+ self.host: str = str(configuration.get("host", "test.operator.io"))
+ self.description = (
+ f'The root service category of 5G services including RAN, core '
+ f'network and IoT services for the year '
+ f'{self.__current_time.strftime("%Y")}.'
)
-
+ self._center: IGeoLocation = cast(IGeoLocation, configuration["center"])
# Calculate layout size using the configuration values.
nr_cell_du = configuration["pattern"]["nrCellDu"]
size = int(
"""
return self.__configuration
+ @property
+ def version(self) -> dict[str, Any]:
+ """
+ Getter for a version value of O-RAN Network.
+ :return A string with the version.
+ """
+ return self.configuration["version"]
+
+ @property
+ def valid_for(self) -> dict[str, Any]:
+ return self.__my_valid_for
+
+ @property
+ def current_time(self) -> datetime:
+ return self.__current_time
+
+ @property
+ def time_string(self) -> str:
+ return self.__time_string
+
+ @property
+ def network(
+ self,
+ ) -> Any: # expected is ORanNetwork
+ return self._network
+
+ @network.setter
+ def network(self, value: Any) -> None:
+ self._network = value
+
def __update_value_by_uuid(self, data: dict[str, Any], target_uuid: str, param_name: str, new_value: str) -> bool:
"""
Recursively searches for an object with the target UUID and updates its parameter value.
# root.append(self.__context.svg(x, y))
return root
+ def to_geojson(self) -> dict[str, Any]:
+ features: list[dict[str, Any]] = self._o_ran_smo.to_geojson_feature()
+ # links: list[dict[str, Any]] = self._o_ran_smo.to_topology_links()
+ return {
+ "type": "FeatureCollection",
+ "features": features,
+ }
+
+ def to_tmf686(self) -> dict[str, Any]:
+ vertex: list[dict[str, Any]] = self._o_ran_smo.to_tmf686_vertex()
+ edge: list[dict[str, Any]] = self._o_ran_smo.to_tmf686_edge()
+
+ return {
+ "id": self.id,
+ "type": "Graph",
+ "name": self.name,
+ "description": "A Network Topology in the context of INDIGO.",
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.id}",
+ "dateCreated": self.__time_string,
+ "lastUpdated": self.__time_string,
+ "status": "active",
+ "graphRelationship": [],
+ "edge": edge,
+ "vertex": vertex,
+ "@base": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "Graph",
+ }
+
+ def to_tmf632_party_organization(self) -> list[dict[str, Any]]:
+ return [self._party_organization.get_organization()]
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._o_ran_smo.to_tmf633_service_candidates()
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._o_ran_smo.to_tmf633_service_specifications()
+
+ def to_tmf634_resource_catalog(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceCatalog/{self.id}",
+ "name": self.name,
+ "description": f'Comprehensive catalog of 5G related resource for the year {self.__current_time.strftime("%Y")}.',
+ "lastUpdate": self.__time_string,
+ "lifecycleStatus": "Active",
+ "version": self.version,
+ "category": [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceCategory/{self.id}",
+ "name": self.name,
+ "version": self.version,
+ "@type": "ResourceCategoryRef",
+ }
+ ],
+ "relatedParty": self.related_party,
+ "validFor": self.valid_for,
+ "@type": "ResourceCatalog",
+ }
+ ]
+
+ def to_tmf634_resource_category(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceCategory/{self.id}",
+ "name": self.name,
+ "description": f'The root resource category of 5G RAN related resources for the year {self.__current_time.strftime("%Y")}.',
+ "lastUpdate": self.__time_string,
+ "lifecycleStatus": "Active",
+ "version": self.version,
+ "isRoot": True,
+ "category": [],
+ "resourceCandidate": self._o_ran_smo.to_tmf634_resource_candidate_references(),
+ "resourceSpecification": self._o_ran_smo.to_tmf634_resource_specification_references(),
+ "validFor": self.valid_for,
+ "relatedParty": self.related_party,
+ "@type": "ResourceCategory",
+ }
+ ]
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._o_ran_smo.to_tmf634_resource_candidates()
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._o_ran_smo.to_tmf634_resource_specifications()
+
+ # Get from network a list of the generated cells
+ def get_list_cells(self) -> list[ORanRu]:
+ return self._o_ran_smo.rus
+
def json(self) -> Dict[str, Any]:
"""
Return a JSON representation of the network.
-# Copyright 2024 highstreet technologies
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
result[key] = data[key] # type: ignore
return result
+ @property
+ def host(self) -> str:
+ return self.parent.host
+
@property
def address(self) -> AddressType:
return self._address
result.append(
{
"node-id": self.name,
- "o-ran-sc-network:uuid": str(
- uuid.uuid5(
- uuid.NAMESPACE_DNS,
- "-".join([self.network.name, self.name]),
- )
- ),
+ "o-ran-sc-network:uuid": str(uuid.uuid5(uuid.NAMESPACE_DNS, "-".join([self.network.name, self.name]))),
"o-ran-sc-network:type": self.type,
"o-ran-sc-network:operational-state": self.operationalState,
"ietf-network-topology:termination-point": tps,
def to_directory(self, parent_dir: str) -> None:
pass
+ def get_coordinates(self) -> list[float]:
+ lng: float = 0.0
+ lat: float = 0.0
+
+ gl = self._geo_location
+ # TODO: Why a is gl sometimes a dict and not a GeoLoaction???
+ if isinstance(gl, GeoLocation):
+ lng = gl.longitude
+ lat = gl.latitude
+ elif isinstance(gl, dict):
+ lng = gl["longitude"]
+ lat = gl["latitude"]
+
+ return [lng, lat]
+
+ @abstractmethod
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = []
+ tps: list[dict[str, Any]] = []
+ for tp in self.termination_points():
+ new_tp = tp.to_topology()
+ if any(existing_tp['tp-id'] == new_tp['tp-id'] for existing_tp in tps):
+ pass
+ else:
+ tps.append(new_tp)
+
+ result.append(
+ {
+ "type": "Feature",
+ "properties": {
+ "type": "PropertiesNode",
+ "node-uuid": str(uuid.uuid5(uuid.NAMESPACE_DNS, "-".join([self.network.name, self.name]))),
+ "node-id": self.name,
+ "function": self.type
+ },
+ "geometry": {
+ "type": "Point",
+ "coordinates": self.get_coordinates(),
+ },
+ }
+ )
+ return result
+
+ @abstractmethod
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = []
+ result.append(
+ {
+ "id": self.id,
+ "name": self.name,
+ "description": f"Description of a vertex object of type {type(self)}",
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}/vertex/{self.id}",
+ # optional "edge": [],
+ "entity": {
+ "id": "indigo",
+ "href": "https://indigo.cosmos-lab.org",
+ "name": "INDIGO",
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "EntityRef",
+ "@referredType": "Individual",
+ },
+ "graph": {
+ "id": self.network.id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}",
+ "name": self.network.name,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "GraphRef",
+ "@referredType": "Graph",
+ },
+ "subGraph": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": F'https://{self.host}/schema/tmf686-schema.json',
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "vertexCharacteristic": [
+ # {
+ # "id": "string",
+ # "name": "string",
+ # "valueType": "string",
+ # "characteristicRelationship": [
+ # {
+ # "id": "string",
+ # "href": "string",
+ # "relationshipType": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ # ],
+ # "value": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": F'https://{self.host}/schema/tmf686-schema.json',
+ # "@type": "string",
+ # }
+ ],
+ "vertexSpecification": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "version": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "Vertex",
+ }
+ )
+ for tp in self.termination_points():
+ result.append(tp.to_tmf686_vertex())
+ return result
+
+ @abstractmethod
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = []
+ source_tp: str = "-".join([self.name, "phy".upper()])
+ dest_tp: str = "-".join([self.parent.name, "phy".upper()])
+ if self.parent and "Tower" not in source_tp and "Tower" not in dest_tp:
+ link_id: str = "".join(["phy", ":", self.name, "<->", self.parent.name])
+ link: dict[str, Any] = {
+ "link-id": link_id,
+ "source": {"source-node": self.name, "source-tp": source_tp},
+ "destination": {
+ "dest-node": self.parent.name,
+ "dest-tp": dest_tp,
+ },
+ }
+ link = {
+ "id": link_id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}/edge/{link_id}",
+ "bidirectional": True,
+ "description": "Description of an edge object",
+ "name": self.name,
+ "edgeCharacteristic": [
+ # {
+ # "id": "string",
+ # "name": "string",
+ # "valueType": "string",
+ # "characteristicRelationship": [
+ # {
+ # "id": "string",
+ # "href": "string",
+ # "relationshipType": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ # ],
+ # "value": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ ],
+ "edgeSpecification": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "version": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "entity": {
+ "id": "indigo",
+ "href": "https://indigo.cosmos-lab.org",
+ "name": "INDIGO",
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "EntityRef",
+ "@referredType": "Individual",
+ },
+ "graph": {
+ "id": self.network.id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}",
+ "name": self.network.name,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "GraphRef",
+ "@referredType": "Graph",
+ },
+ "subGraph": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "vertex": [
+ {
+ "id": source_tp,
+ "href": (
+ f'https://{self.host}/tmf-api/topologyDiscovery/v4'
+ f'/graph/{self.network.id}/vertex/{source_tp}'
+ ),
+ "name": source_tp,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "VertexRef",
+ "@referredType": "Vertex",
+ },
+ {
+ "id": dest_tp,
+ "href": (
+ f'https://{self.host}/tmf-api/topologyDiscovery/v4'
+ f'/graph/{self.network.id}/vertex/{dest_tp}'
+ ),
+ "name": dest_tp,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "VertexRef",
+ "@referredType": "Vertex",
+ },
+ ],
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "Edge",
+ }
+ result.append(link)
+ return result
+
+ @abstractmethod
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/serviceCatalogManagement/v4/serviceCandidate/{self.id}",
+ "name": self.name,
+ "version": self.network.version,
+ }
+ ]
+
+ @abstractmethod
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/serviceCatalogManagement/v4/serviceCandidate/{self.id}",
+ "name": self.name,
+ "description": (
+ "The service candidate of 5G services including RAN, core "
+ "network and IoT services for the year "
+ f'{self.__current_time.strftime("%Y")}.'),
+ "lastUpdate": self.__time_string,
+ "lifecycleStatus": "Active",
+ "version": self.network.version,
+ "category": [
+ {
+ "id": self.network.id,
+ "href": (
+ f"https://{self.host}/tmf-api/serviceCatalogManagement/v4/serviceCategory/{self.network.id}"
+ ),
+ "name": self.network.name,
+ "version": self.network.version,
+ }
+ ],
+ "serviceSpecification": {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/serviceCatalogManagement/v4/serviceSpecification/{self.id}",
+ "name": self.name,
+ "version": self.network.version,
+ },
+ "validFor": self.network.valid_for,
+ }
+ ]
+
+ def get_oran_network_id_property(self) -> str:
+ if self.__class__.__name__ == "NrCellDu":
+ return "properties.name"
+ else:
+ return "properties.node-id"
+
+ @abstractmethod
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = []
+ result.append(
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/serviceCatalogManagement/v4/serviceSpecification/{self.id}",
+ "name": self.name,
+ "description": "INDIGO Service Specification PublicSafety",
+ "isBundle": False,
+ "version": self.network.version,
+ "lastUpdate": self.__time_string,
+ "lifecycleStatus": "Active",
+ "relatedParty": self.network.related_party,
+ "validFor": self.network.valid_for,
+ "attachment": [],
+ "entitySpecRelationship": [],
+ "featureSpecification": [],
+ "resourceSpecification": self.to_tmf634_resource_specification_references(),
+ "serviceLevelSpecification": [],
+ "serviceSpecRelationship": [],
+ "specCharacteristic": [
+ {
+ "id": "resilienceLevel",
+ "name": "Resilience Level",
+ "description": "The resilience level of this service.",
+ "valueType": "integer",
+ "configurable": False,
+ "characteristicValueSpecification": [
+ {"value": 0, "isDefault": True}
+ ],
+ },
+ {
+ "id": "securityLevel",
+ "description": "The security level of this service.",
+ "valueType": "integer",
+ "configurable": False,
+ "characteristicValueSpecification": [
+ {"value": 1, "isDefault": True}
+ ],
+ },
+ {
+ "id": "sNssai",
+ "name": "Single Network Slice Selection Assistance Information",
+ "description": "The Single Network Slice Selection Assistance Information of this service.",
+ "configurable": True,
+ "valueType": "Integer32",
+ "isUnique": True,
+ },
+ {
+ "id": "maxBandwidthUE",
+ "name": "Maximum Bandwidth UE",
+ "description": "Maximum bandwidth in [MBps] per User Equipment (UE)",
+ "configurable": True,
+ "valueType": "Mbps",
+ "minCardinality": 5,
+ "maxCardinality": 1000,
+ },
+ {
+ "id": "coverageArea",
+ "name": "Coverage Area",
+ "description": "The geographical area covered by this entity.",
+ "configurable": False,
+ "extensible": False,
+ "isUnique": True,
+ "characteristicValueSpecification": [
+ {
+ "valueType": "href",
+ "value": (
+ f'https://{self.host}/area/o-ran-network.'
+ "geo.json/features?"
+ f'{self.get_oran_network_id_property()}='
+ f'{self.name}'),
+ }
+ ],
+ },
+ {
+ "id": "maxLatency",
+ "name": "Maximal Latency",
+ "description": (
+ "Maximum tolerable delay in milliseconds [ms]"),
+ "configurable": False,
+ "valueType": "time",
+ "characteristicValueSpecification": [
+ {"valueType": "integer", "value": 10}
+ ],
+ },
+ {
+ "name": "availability",
+ "description": "Guaranteed service uptime",
+ "configurable": False,
+ "valueType": "Propability",
+ "characteristicValueSpecification": [
+ {"valueType": "Propability", "value": "99.9%"}
+ ],
+ },
+ {
+ "id": "mttr",
+ "name": "Mean Time to Repair",
+ "description": "Maximal Mean Time To Repair a failure",
+ "configurable": False,
+ "valueType": "time",
+ "characteristicValueSpecification": [
+ {"valueType": "maximum time", "value": "4 hours"}
+ ],
+ },
+ {
+ "id": "mtbf",
+ "name": "Mean Time Between Failures",
+ "description": "Target for Mean Time Between Failures",
+ "configurable": False,
+ "valueType": "time",
+ "characteristicValueSpecification": [
+ {"valueType": "minimum time", "value": "1 year"}
+ ],
+ },
+ ],
+ }
+ )
+ return result
+
+ @abstractmethod
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceCandidate/{self.id}",
+ "name": self.name,
+ "version": self.network.version,
+ "@type": "ResourceCandidateRef",
+ "@referredType": "ResourceCandidate",
+ }
+ ]
+
+ @abstractmethod
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceSpecification/{self.id}",
+ "name": self.name,
+ "version": self.network.version,
+ "@type": "ResourceSpecificationRef",
+ "@referredType": "ResourceSpecification",
+ }
+ ]
+
+ @abstractmethod
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceCandidate/{self.id}",
+ "name": self.name,
+ "description": f'The resource candidate of 5G RAN related resources for year {self.__current_time.strftime("%Y")}.',
+ "lastUpdate": self.__time_string,
+ "lifecycleStatus": "Active",
+ "version": self.network.version,
+ "category": [
+ {
+ "id": self.network.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceCategory/{self.network.id}",
+ "name": self.network.name,
+ "version": self.network.version,
+ }
+ ],
+ "resourceSpecification": {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceSpecification/{self.id}",
+ "name": self.name,
+ "version": self.network.version,
+ "@type": "ResourceSpecificationRef",
+ "@referredType": "ResourceSpecification",
+ },
+ "validFor": self.network.valid_for,
+ "@type": "ResourceCandidate",
+ }
+ ]
+
+ @abstractmethod
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = []
+ result.append(
+ {
+ "id": self.id,
+ "href": f"https://{self.host}/tmf-api/resourceCatalog/v5/resourceSpecification/{self.id}",
+ "name": self.name,
+ "description": f"INDIGO Resource Specification for type {type(self)}",
+ "isBundle": False,
+ "version": self.network.version,
+ "lastUpdate": self.__time_string,
+ "lifecycleStatus": "Active",
+ "relatedParty": self.network.related_party,
+ "validFor": self.network.valid_for,
+ "category": "RAN resource",
+ "attachment": [],
+ "featureSpecification": [],
+ "resourceSpecCharacteristic": [
+ {
+ "name": "nRCellName",
+ "description": "Cell identifier or name of the cell.",
+ "configurable": True,
+ "isUnique": True,
+ "valueType": "string",
+ },
+ {
+ "name": "nRCellId",
+ "description": "Uniquely identifies a cell within a PLMN. It is often constructed from gNodeB ID + Physical Cell ID.",
+ "valueType": "string",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "nRCellState",
+ "description": (
+ "Represents the active state of the cell. Takes "
+ "one of the following values: "
+ "IDLE ACTIVE INACTIVE UNKNOWN"),
+ "valueType": "string",
+ "configurable": False,
+ "isUnique": True,
+ },
+ {
+ "name": "TAI",
+ "description": (
+ "Tracking Area Identifier (TAI). This is a "
+ "globally unique tracking area identifier, "
+ "made up of the PLMN ID and the TAC."),
+ "valueType": "string",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "channelBandwidthUl",
+ "description": "Uplink channel bandwidth.",
+ "valueType": "float",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "channelBandwidthDl",
+ "description": "Downlink channel bandwidth.",
+ "valueType": "float",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "maximumOutputPower",
+ "description": (
+ "Maximum power in Watts for the sum of all downlink"
+ " channels that are allowed to be used "
+ "simultaneously in a cell."),
+ "valueType": "float",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "userCapacity",
+ "description": (
+ "Maximum number of pieces of user equipment (UEs) "
+ "that can connect to this nrCellDU simultaneously."
+ ),
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "physicalCellID",
+ "description": (
+ "Physical cell identifier. Takes a value in the "
+ "range 0 to 503. The physical cell id is used by "
+ "the cell to encode and decode the data that it "
+ "transmits. It is used in a similar way to the "
+ "UMTS scrambling code. To avoid interference, "
+ "neighboring cells should have different physical "
+ "cell identifiers. The physical cell id is derived "
+ "from the primary and secondary synchronization "
+ "signals (PSS and SSS). The PSS takes a value from "
+ "0 to 2, the SSS takes a value from 0 to 167, and "
+ "the physical cell id is determined based on the "
+ "following formula: PSS + 3*SSS. The result of "
+ "this calculation equates to a value of between "
+ "0 and 503."),
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "localCellId",
+ "description": "Local cell id unique within the nrCellDU.",
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "arfcnDl",
+ "description": (
+ "Absolute Radio Frequency Channel Number "
+ "(downlink). An integer value which identifies the "
+ "downlink carrier frequency of the cell."),
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "arfcnUl",
+ "description": (
+ "Absolute Radio Frequency Channel Number (uplink). "
+ "An integer value which identifies the uplink "
+ "carrier frequency of the cell."),
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "nRPCI",
+ "description": (
+ "Holds the Physical Cell Identity (PCI) of the "
+ "NR cell."),
+ "valueType": "String",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "ssbFreq",
+ "description": (
+ "Indicates cell defining SSB frequency domain "
+ "position. Frequency of the cell defining SSB "
+ "transmission. The frequency provided in this "
+ "attribute identifies the position of resource "
+ "element. The frequency shall be positioned on the "
+ "NR global frequency raster, and within "
+ "bSChannelBwDL. Allowed values: 0..3279165"),
+ "valueType": "Float",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "ssbPeriodicity",
+ "description": (
+ "Indicates cell defined SSB periodicity in number "
+ "of subframes(ms). The SSB periodicity in msec is "
+ "used for the rate matching purpose. "
+ "Allowed values: 5, 10, 20, 40, 80, 160"),
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "ssbSubCarrierSpacing",
+ "description": "This SSB is used for synchronization. Its units are in kHz. Allowed values: {15, 30, 120, 240}",
+ "valueType": "Integer",
+ "configurable": True,
+ "isUnique": True,
+ },
+ {
+ "name": "operationalState",
+ "description": (
+ "Operational state of the nrCellDU. Takes one of "
+ "the following values: Enabled Disabled Other "
+ "Unknown"),
+ "valueType": "String",
+ "configurable": False,
+ "isUnique": True,
+ },
+ {
+ "name": "administrativeState",
+ "description": (
+ "Administrative state of the nrCellDU. Takes one of"
+ " the following values: Unlocked Locked Shutting "
+ "Down Other Unknown"),
+ "valueType": "String",
+ "configurable": False,
+ "isUnique": True,
+ },
+ ],
+ "resourceSpecRelationship": [], # TODO self.parent.to_tmf634_resource_specifications(),
+ "@type": "ResourceSpecification",
+ }
+ )
+ return result
+
@abstractmethod
def add_teiv_data_entities(
self, entity_type: str, attributes: dict[str, Any] = {}
-# Copyright 2024 highstreet technologies
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
result.extend(self.flatten_list(getattr(cell, cell_method_name)()))
return result
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_geojson_feature,
+ "to_geojson_feature",
+ )
+
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = super().to_tmf686_vertex()
+ result.extend(self.oRanDu.to_tmf686_vertex())
+ return result
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ result: list[dict[str, Any]] = super().to_topology_links()
+ result.extend(self.oRanDu.to_topology_links())
+ for interface in ["phy", "ofhm", "ofhc", "ofhu", "ofhs"]:
+ link_id: str = "".join([interface, ":", self.name, "<->", self.oRanDu.name])
+ source_tp: str = "-".join([self.name, interface.upper()])
+ dest_tp: str = "-".join([self.oRanDu.name, interface.upper()])
+ result.append(
+ {
+ "id": link_id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}/edge/{link_id}",
+ "bidirectional": True,
+ "description": "Description of an edge object",
+ "name": self.name,
+ "edgeCharacteristic": [
+ # {
+ # "id": "string",
+ # "name": "string",
+ # "valueType": "string",
+ # "characteristicRelationship": [
+ # {
+ # "id": "string",
+ # "href": "string",
+ # "relationshipType": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ # ],
+ # "value": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ ],
+ "edgeSpecification": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "version": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "entity": {
+ "id": "indigo",
+ "href": "https://indigo.cosmos-lab.org",
+ "name": "INDIGO",
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "EntityRef",
+ "@referredType": "Individual",
+ },
+ "graph": {
+ "id": self.network.id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}",
+ "name": self.network.name,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "GraphRef",
+ "@referredType": "Graph",
+ },
+ "subGraph": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "vertex": [
+ {
+ "id": source_tp,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}/vertex/{source_tp}",
+ "name": source_tp,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "VertexRef",
+ "@referredType": "Vertex",
+ },
+ {
+ "id": dest_tp,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}/vertex/{dest_tp}",
+ "name": dest_tp,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "VertexRef",
+ "@referredType": "Vertex",
+ },
+ ],
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.host}/schema/tmf686-schema.json",
+ "@type": "Edge",
+ }
+ )
+ return result
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf633_service_candidate_references,
+ "to_tmf633_service_candidate_references",
+ )
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf633_service_candidates,
+ "to_tmf633_service_candidates",
+ )
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf633_service_specifications,
+ "to_tmf633_service_specifications",
+ )
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf634_resource_candidate_references,
+ "to_tmf634_resource_candidate_references",
+ )
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf634_resource_specification_references,
+ "to_tmf634_resource_specification_references",
+ )
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf634_resource_candidates, "to_tmf634_resource_candidates"
+ )
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_cell_references(
+ super().to_tmf634_resource_specifications,
+ "to_tmf634_resource_specifications",
+ )
+
def add_teiv_data_entities(
self,
entity_type: str = "o-ran-smo-teiv-ran:ORUFunction",
-# Copyright 2024 highstreet technologies
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
ORanNode,
default_value,
)
+from network_generation.model.python.o_ran_ru import ORanRu
from network_generation.model.python.tower import Tower
# Define the "IORanSmo" interface
"""
Class representing an O-RAN Service Management and Operation object.
"""
-
_interfaces = ["a1", "o1", "ofhm", "o2"]
def __init__(
result.append(tower)
return result
+ @property
+ def rus(self) -> list[ORanRu]:
+ result: list[ORanRu] = []
+ for ric in self.o_ran_near_rt_rics:
+ for tower in ric.towers:
+ result.extend(tower.o_ran_rus)
+ return result
+
def toKml(self) -> ET.Element:
smo = super().toKml()
for ric in self.o_ran_near_rt_rics:
super().to_topology_links, "to_topology_links"
)
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf686_vertex, "to_tmf686_vertex")
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf686_edge, "to_tmf686_edge")
+
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_geojson_feature, "to_geojson_feature")
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf633_service_candidate_references,
+ "to_tmf633_service_candidate_references")
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf633_service_candidates,
+ "to_tmf633_service_candidates")
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf633_service_specifications,
+ "to_tmf633_service_specifications")
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf634_resource_candidate_references,
+ "to_tmf634_resource_candidate_references")
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf634_resource_specification_references,
+ "to_tmf634_resource_specification_references")
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf634_resource_candidates,
+ "to_tmf634_resource_candidates")
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_ric_references(
+ super().to_tmf634_resource_specifications,
+ "to_tmf634_resource_specifications")
+
def _extend_teiv_data_with_ric_references(
self: Any,
teiv_data: dict[str, list[dict[str, Any]]],
-# Copyright 2024 highstreet technologies
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
result[key] = data[key] # type: ignore
return result
+ @property
+ def host(self) -> str:
+ return self.network.host
+
@property
def supporter(self) -> dict[str, str]:
return self._supporter
if not (self.type == "o-ran-sc-network:phy"):
result["supporting-termination-point"] = [self.supporter]
return result
+
+ def to_tmf686_vertex(self) -> dict[str, Any]:
+ result: dict[str, Any] = {
+ "id": self.id,
+ "name": self.name,
+ "description": f"Description of a vertex object of type {type(self)}",
+ "@type": "Vertex",
+ }
+ if (self.supporter and self.network):
+ # TODO self.network!!! network.host
+ result = {
+ "id": self.id,
+ "name": self.name,
+ "description": (
+ f'Description of a vertex object of type {type(self)}'),
+ "href": (
+ f'https://{self.network.host}/tmf-api/topologyDiscovery/v4/'
+ f'graph/{self.network.id}/vertex/{self.id}'),
+ # optional "edge": [],
+ "entity": {
+ "id": "indigo",
+ "href": "https://indigo.cosmos-lab.org",
+ "name": "INDIGO",
+ "@baseType": "object",
+ "@schemaLocation": (
+ f'https://{self.network.host}/schema/tmf686-schema.json'
+ ),
+ "@type": "EntityRef",
+ "@referredType": "Individual",
+ },
+ "graph": {
+ "id": self.network.id,
+ "href": f"https://{self.host}/tmf-api/topologyDiscovery/v4/graph/{self.network.id}",
+ "name": self.network.name,
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.network.host}/schema/tmf686-schema.json",
+ "@type": "GraphRef",
+ "@referredType": "Graph",
+ },
+ "subGraph": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": f'https://{self.network.host}/schema/tmf686-schema.json',
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "vertexCharacteristic": [
+ # {
+ # "id": "string",
+ # "name": "string",
+ # "valueType": "string",
+ # "characteristicRelationship": [
+ # {
+ # "id": "string",
+ # "href": "string",
+ # "relationshipType": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # }
+ # ],
+ # "value": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": f'https://{self.network.host}/schema/tmf686-schema.json',
+ # "@type": "string",
+ # }
+ ],
+ "vertexSpecification": {
+ # "id": "string",
+ # "href": "string",
+ # "name": "string",
+ # "version": "string",
+ # "@baseType": "string",
+ # "@schemaLocation": "string",
+ # "@type": "string",
+ # "@referredType": "string",
+ },
+ "@baseType": "object",
+ "@schemaLocation": f"https://{self.network.host}/schema/tmf686-schema.json",
+ "@type": "Vertex",
+ }
+ return result
"to_topology_links",
)
+ def to_tmf686_vertex(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf686_vertex,
+ "to_tmf686_vertex",
+ )
+
+ def to_tmf686_edge(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf686_edge,
+ "to_tmf686_edge",
+ )
+
+ def to_geojson_feature(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_geojson_feature,
+ "to_geojson_feature",
+ )
+
+ def to_tmf633_service_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf633_service_candidate_references,
+ "to_tmf633_service_candidate_references",
+ )
+
+ def to_tmf633_service_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf633_service_candidates,
+ "to_tmf633_service_candidates",
+ )
+
+ def to_tmf633_service_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf633_service_specifications,
+ "to_tmf633_service_specifications",
+ )
+
+ def to_tmf634_resource_candidate_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf634_resource_candidate_references,
+ "to_tmf634_resource_candidate_references",
+ )
+
+ def to_tmf634_resource_specification_references(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf634_resource_specification_references,
+ "to_tmf634_resource_specification_references",
+ )
+
+ def to_tmf634_resource_candidates(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf634_resource_candidates, "to_tmf634_resource_candidates"
+ )
+
+ def to_tmf634_resource_specifications(self) -> list[dict[str, Any]]:
+ return self._extend_with_o_ran_ru_references(
+ super().to_tmf634_resource_specifications,
+ "to_tmf634_resource_specifications",
+ )
+
def _extend_teiv_data_with_o_ran_ru_references(
self: Any, o_ran_ru_method_name: str
) -> dict[str, list[dict[str, Any]]]:
-# Copyright 2024 highstreet technologies GmbH
+# Copyright 2025 highstreet technologies USA Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
import json
import xml.etree.ElementTree as ET
import zipfile
-from typing import Any
-
+from typing import Any, Callable
from typing_extensions import Buffer
from network_generation.model.python.o_ran_network import ORanNetwork
kml_file.write(kml)
kml_file.close()
print(f'File "{filename}.kml" saved!')
+
+ def rfc7946(self, filename: str, compressed: bool = True) -> None:
+ """
+ Method saving the class content to a file in geojson format.
+
+ :param filename: A valid path to a file on the system.
+ :param compressed: if True, kml is stored as kmz format.
+ :type filename: string
+ """
+ output: dict[str, Any] = self.__network.to_geojson()
+ file_extension: str = ".geo.json"
+ self.__save_on_disc(f"{filename}{file_extension}", compressed, output)
+
+ def tmf686(self, filename: str, compressed: bool = True) -> None:
+ """
+ Method saving the class content to a file in json format.
+ :param filename: A valid path to a file on the system.
+ :param compressed: if True, svg is stored as tmf686 format.
+ :type filename: string
+ """
+ output: dict[str, Any] = self.__network.to_tmf686()
+ file_extension: str = ".tmf686.json"
+ self.__save_on_disc(f"{filename}{file_extension}", compressed, output)
+
+
+ def tmf632(self, filename: str, compressed: bool = True) -> None:
+ """
+ Method saving the class content to a file in tmf632 format.
+ """
+
+ file_suffixes = {
+ "party-organization": self.__network.to_tmf632_party_organization,
+ }
+ for suffix, method in file_suffixes.items():
+ output: list[dict[str, Any]] = method()
+ file_extension: str = f".tmf632.{suffix}.json"
+ self.__save_on_disc(
+ f"{filename}{file_extension}", compressed, output)
+
+ def tmf633(self, filename: str, compressed: bool = True) -> None:
+ """
+ Method saving the class content to a file in tmf633 format.
+ It requires 4 json files:
+ 1. service-catalog
+ 2. service-category
+ 3. service-definition
+ 4. service-candidate
+
+ :param filename: A valid path to a file on the system.
+ :param compressed: if True, kml is stored as kmz format.
+ :type filename: string
+ """
+
+ file_suffixes = {
+ "service-catalogs": self.service_catalog.to_tmf633_service_catalog,
+ "service-categories": lambda: self.__network.to_tmf633_service_category() + self.__service2.to_tmf633_service_category() + self.__service3.to_tmf633_service_category(),
+ "service-candidates": lambda: self.__network.to_tmf633_service_candidates() + self.__service2.to_tmf633_service_candidates() + self.__service3.to_tmf633_service_candidates(),
+ "service-specifications": lambda: self.__network.to_tmf633_service_specifications() + self.__service2.to_tmf633_service_specifications() + self.__service3.to_tmf633_service_specifications(),
+ }
+ for suffix, method in file_suffixes.items():
+ output: list[dict[str, Any]] = method()
+ file_extension: str = f".tmf633.{suffix}.json"
+ self.__save_on_disc(
+ f"{filename}{file_extension}", compressed, output)
+
+ def tmf634(self, filename: str, compressed: bool = True) -> None:
+ """
+ Method saving the class content to a file in tmf634 format.
+ It requires 4 json files:
+ 1. resource-catalog
+ 2. resource-category
+ 3. resource-definition
+ 4. resource-candidate
+
+ :param filename: A valid path to a file on the system.
+ :param compressed: if True, kml is stored as kmz format.
+ :type filename: string
+ """
+
+ file_suffixes = {
+ "resource-catalogs": self.__network.to_tmf634_resource_catalog,
+ "resource-categories": self.__network.to_tmf634_resource_category,
+ "resource-candidates": self.__network.to_tmf634_resource_candidates,
+ "resource-specifications": self.__network.to_tmf634_resource_specifications,
+ }
+
+ for suffix, method in file_suffixes.items():
+ output: list[dict[str, Any]] = method()
+ file_extension: str = f".tmf634.{suffix}.json"
+ self.__save_on_disc(
+ f"{filename}{file_extension}", compressed, output)
+
+ # Extend process_data to handle a list of methods
+ @staticmethod
+ def chain(methods: list[Callable[[], list[dict[str, Any]]]]) -> list[dict[str, Any]]:
+ result = []
+
+ # Loop through each method, call it, and extend the result list
+ for method in methods:
+ data = method() # Call the method
+ # Process the data, e.g., by adding a new field to each dictionary
+ for item in data:
+ item['processed'] = True # Adding a new field 'processed'
+ result.extend(data)
+
+ return result
Code Review / oam.git / commitdiff