daemon: added class variable type hinting for core.xml
This commit is contained in:
parent
cfaa9397ad
commit
ef3cf5697d
3 changed files with 18 additions and 23 deletions
|
@ -4,7 +4,7 @@ Common support for configurable CORE objects.
|
|||
|
||||
import logging
|
||||
from collections import OrderedDict
|
||||
from typing import TYPE_CHECKING, Dict, List, Tuple, Type, Union
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Tuple, Type, Union
|
||||
|
||||
from core.emane.nodes import EmaneNet
|
||||
from core.emulator.enumerations import ConfigDataTypes
|
||||
|
@ -136,7 +136,8 @@ class ConfigurableManager:
|
|||
"""
|
||||
Clears all configurations or configuration for a specific node.
|
||||
|
||||
:param node_id: node id to clear configurations for, default is None and clears all configurations
|
||||
:param node_id: node id to clear configurations for, default is None and clears
|
||||
all configurations
|
||||
:return: nothing
|
||||
"""
|
||||
if not node_id:
|
||||
|
@ -222,7 +223,7 @@ class ConfigurableManager:
|
|||
result = node_configs.get(config_type)
|
||||
return result
|
||||
|
||||
def get_all_configs(self, node_id: int = _default_node) -> List[Dict[str, str]]:
|
||||
def get_all_configs(self, node_id: int = _default_node) -> Dict[str, Any]:
|
||||
"""
|
||||
Retrieve all current configuration types for a node.
|
||||
|
||||
|
|
|
@ -124,9 +124,9 @@ def add_configuration(parent: etree.Element, name: str, value: str) -> None:
|
|||
|
||||
class NodeElement:
|
||||
def __init__(self, session: "Session", node: NodeBase, element_name: str) -> None:
|
||||
self.session = session
|
||||
self.node = node
|
||||
self.element = etree.Element(element_name)
|
||||
self.session: "Session" = session
|
||||
self.node: NodeBase = node
|
||||
self.element: etree.Element = etree.Element(element_name)
|
||||
add_attribute(self.element, "id", node.id)
|
||||
add_attribute(self.element, "name", node.name)
|
||||
add_attribute(self.element, "icon", node.icon)
|
||||
|
@ -151,8 +151,8 @@ class NodeElement:
|
|||
|
||||
class ServiceElement:
|
||||
def __init__(self, service: Type[CoreService]) -> None:
|
||||
self.service = service
|
||||
self.element = etree.Element("service")
|
||||
self.service: Type[CoreService] = service
|
||||
self.element: etree.Element = etree.Element("service")
|
||||
add_attribute(self.element, "name", service.name)
|
||||
self.add_directories()
|
||||
self.add_startup()
|
||||
|
@ -268,10 +268,10 @@ class NetworkElement(NodeElement):
|
|||
|
||||
class CoreXmlWriter:
|
||||
def __init__(self, session: "Session") -> None:
|
||||
self.session = session
|
||||
self.scenario = etree.Element("scenario")
|
||||
self.networks = None
|
||||
self.devices = None
|
||||
self.session: "Session" = session
|
||||
self.scenario: etree.Element = etree.Element("scenario")
|
||||
self.networks: etree.SubElement = etree.SubElement(self.scenario, "networks")
|
||||
self.devices: etree.SubElement = etree.SubElement(self.scenario, "devices")
|
||||
self.write_session()
|
||||
|
||||
def write_session(self) -> None:
|
||||
|
@ -362,13 +362,11 @@ class CoreXmlWriter:
|
|||
def write_emane_configs(self) -> None:
|
||||
emane_global_configuration = create_emane_config(self.session)
|
||||
self.scenario.append(emane_global_configuration)
|
||||
|
||||
emane_configurations = etree.Element("emane_configurations")
|
||||
for node_id in self.session.emane.nodes():
|
||||
all_configs = self.session.emane.get_all_configs(node_id)
|
||||
if not all_configs:
|
||||
continue
|
||||
|
||||
for model_name in all_configs:
|
||||
config = all_configs[model_name]
|
||||
logging.debug(
|
||||
|
@ -453,9 +451,6 @@ class CoreXmlWriter:
|
|||
self.scenario.append(node_types)
|
||||
|
||||
def write_nodes(self) -> List[LinkData]:
|
||||
self.networks = etree.SubElement(self.scenario, "networks")
|
||||
self.devices = etree.SubElement(self.scenario, "devices")
|
||||
|
||||
links = []
|
||||
for node_id in self.session.nodes:
|
||||
node = self.session.nodes[node_id]
|
||||
|
@ -472,7 +467,6 @@ class CoreXmlWriter:
|
|||
|
||||
# add known links
|
||||
links.extend(node.all_link_data())
|
||||
|
||||
return links
|
||||
|
||||
def write_network(self, node: NodeBase) -> None:
|
||||
|
@ -597,8 +591,8 @@ class CoreXmlWriter:
|
|||
|
||||
class CoreXmlReader:
|
||||
def __init__(self, session: "Session") -> None:
|
||||
self.session = session
|
||||
self.scenario = None
|
||||
self.session: "Session" = session
|
||||
self.scenario: Optional[etree.ElementTree] = None
|
||||
|
||||
def read(self, file_name: str) -> None:
|
||||
xml_tree = etree.parse(file_name)
|
||||
|
|
|
@ -101,9 +101,9 @@ def get_ipv4_addresses(hostname: str) -> List[Tuple[str, str]]:
|
|||
|
||||
class CoreXmlDeployment:
|
||||
def __init__(self, session: "Session", scenario: etree.Element) -> None:
|
||||
self.session = session
|
||||
self.scenario = scenario
|
||||
self.root = etree.SubElement(
|
||||
self.session: "Session" = session
|
||||
self.scenario: etree.Element = scenario
|
||||
self.root: etree.SubElement = etree.SubElement(
|
||||
scenario, "container", id="TestBed", name="TestBed"
|
||||
)
|
||||
self.add_deployment()
|
||||
|
|
Loading…
Reference in a new issue