"""
Provides an EMANE network node class, which has several attached NEMs that
share the same MAC+PHY model.
"""
import logging
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type
from core.emulator.distributed import DistributedServer
from core.emulator.enumerations import LinkTypes, NodeTypes, RegisterTlvs
from core.nodes.base import CoreNetworkBase
from core.nodes.interface import CoreInterface
if TYPE_CHECKING:
from core.emulator.session import Session
from core.location.mobility import WirelessModel
WirelessModelType = Type[WirelessModel]
try:
from emane.events import LocationEvent
except ImportError:
try:
from emanesh.events import LocationEvent
except ImportError:
logging.debug("compatible emane python bindings not installed")
class EmaneNet(CoreNetworkBase):
"""
EMANE node contains NEM configuration and causes connected nodes
to have TAP interfaces (instead of VEth). These are managed by the
Emane controller object that exists in a session.
"""
apitype = NodeTypes.EMANE
linktype = LinkTypes.WIRED
type = "wlan"
is_emane = True
def __init__(
self,
session: "Session",
_id: int = None,
name: str = None,
start: bool = True,
server: DistributedServer = None,
) -> None:
super().__init__(session, _id, name, start, server)
self.conf = ""
self.up = False
self.nemidmap = {}
self.model = None
self.mobility = None
def linkconfig(
self,
netif: CoreInterface,
bw: float = None,
delay: float = None,
loss: float = None,
duplicate: float = None,
jitter: float = None,
netif2: CoreInterface = None,
) -> None:
"""
The CommEffect model supports link configuration.
"""
if not self.model:
return
self.model.linkconfig(
netif=netif,
bw=bw,
delay=delay,
loss=loss,
duplicate=duplicate,
jitter=jitter,
netif2=netif2,
)
def config(self, conf: str) -> None:
self.conf = conf
def shutdown(self) -> None:
pass
def link(self, netif1: CoreInterface, netif2: CoreInterface) -> None:
pass
def unlink(self, netif1: CoreInterface, netif2: CoreInterface) -> None:
pass
def updatemodel(self, config: Dict[str, str]) -> None:
if not self.model:
raise ValueError("no model set to update for node(%s)", self.id)
logging.info(
"node(%s) updating model(%s): %s", self.id, self.model.name, config
)
self.model.set_configs(config, node_id=self.id)
def setmodel(self, model: "WirelessModelType", config: Dict[str, str]) -> None:
"""
set the EmaneModel associated with this node
"""
logging.info("adding model: %s", model.name)
if model.config_type == RegisterTlvs.WIRELESS:
# EmaneModel really uses values from ConfigurableManager
# when buildnemxml() is called, not during init()
self.model = model(session=self.session, _id=self.id)
self.model.update_config(config)
elif model.config_type == RegisterTlvs.MOBILITY:
self.mobility = model(session=self.session, _id=self.id)
self.mobility.update_config(config)
def setnemid(self, netif: CoreInterface, nemid: int) -> None:
"""
Record an interface to numerical ID mapping. The Emane controller
object manages and assigns these IDs for all NEMs.
"""
self.nemidmap[netif] = nemid
def getnemid(self, netif: CoreInterface) -> Optional[int]:
"""
Given an interface, return its numerical ID.
"""
if netif not in self.nemidmap:
return None
else:
return self.nemidmap[netif]
def getnemnetif(self, nemid: int) -> Optional[CoreInterface]:
"""
Given a numerical NEM ID, return its interface. This returns the
first interface that matches the given NEM ID.
"""
for netif in self.nemidmap:
if self.nemidmap[netif] == nemid:
return netif
return None
def netifs(self, sort: bool = True) -> List[CoreInterface]:
"""
Retrieve list of linked interfaces sorted by node number.
"""
return sorted(self._netif.values(), key=lambda ifc: ifc.node.id)
def installnetifs(self) -> None:
"""
Install TAP devices into their namespaces. This is done after
EMANE daemons have been started, because that is their only chance
to bind to the TAPs.
"""
if (
self.session.emane.genlocationevents()
and self.session.emane.service is None
):
warntxt = "unable to publish EMANE events because the eventservice "
warntxt += "Python bindings failed to load"
logging.error(warntxt)
for netif in self.netifs():
external = self.session.emane.get_config(
"external", self.id, self.model.name
)
if external == "0":
netif.setaddrs()
if not self.session.emane.genlocationevents():
netif.poshook = None
continue
# at this point we register location handlers for generating
# EMANE location events
netif.poshook = self.setnemposition
netif.setposition()
def deinstallnetifs(self) -> None:
"""
Uninstall TAP devices. This invokes their shutdown method for
any required cleanup; the device may be actually removed when
emanetransportd terminates.
"""
for netif in self.netifs():
if "virtual" in netif.transport_type.lower():
netif.shutdown()
netif.poshook = None
def _nem_position(
self, netif: CoreInterface
) -> Optional[Tuple[int, float, float, float]]:
"""
Creates nem position for emane event for a given interface.
:param netif: interface to get nem emane position for
:return: nem position tuple, None otherwise
"""
nemid = self.getnemid(netif)
ifname = netif.localname
if nemid is None:
logging.info("nemid for %s is unknown", ifname)
return
node = netif.node
x, y, z = node.getposition()
lat, lon, alt = self.session.location.getgeo(x, y, z)
if node.position.alt is not None:
alt = node.position.alt
# altitude must be an integer or warning is printed
alt = int(round(alt))
return nemid, lon, lat, alt
def setnemposition(self, netif: CoreInterface) -> None:
"""
Publish a NEM location change event using the EMANE event service.
:param netif: interface to set nem position for
"""
if self.session.emane.service is None:
logging.info("position service not available")
return
position = self._nem_position(netif)
if position:
nemid, lon, lat, alt = position
event = LocationEvent()
event.append(nemid, latitude=lat, longitude=lon, altitude=alt)
self.session.emane.service.publish(0, event)
def setnempositions(self, moved_netifs: List[CoreInterface]) -> None:
"""
Several NEMs have moved, from e.g. a WaypointMobilityModel
calculation. Generate an EMANE Location Event having several
entries for each netif that has moved.
"""
if len(moved_netifs) == 0:
return
if self.session.emane.service is None:
logging.info("position service not available")
return
event = LocationEvent()
for netif in moved_netifs:
position = self._nem_position(netif)
if position:
nemid, lon, lat, alt = position
event.append(nemid, latitude=lat, longitude=lon, altitude=alt)
self.session.emane.service.publish(0, event)