""" Provides functionality for maintaining information about known links for a session. """ import logging from collections.abc import ValuesView from dataclasses import dataclass from typing import Optional from core.emulator.data import LinkData, LinkOptions from core.emulator.enumerations import LinkTypes, MessageFlags from core.errors import CoreError from core.nodes.base import NodeBase from core.nodes.interface import CoreInterface from core.nodes.network import PtpNet logger = logging.getLogger(__name__) LinkKeyType = tuple[int, Optional[int], int, Optional[int]] def create_key( node1: NodeBase, iface1: Optional[CoreInterface], node2: NodeBase, iface2: Optional[CoreInterface], ) -> LinkKeyType: """ Creates a unique key for tracking links. :param node1: first node in link :param iface1: node1 interface :param node2: second node in link :param iface2: node2 interface :return: link key """ iface1_id = iface1.id if iface1 else None iface2_id = iface2.id if iface2 else None if node1.id < node2.id: return node1.id, iface1_id, node2.id, iface2_id else: return node2.id, iface2_id, node1.id, iface1_id @dataclass class CoreLink: """ Provides a core link data structure. """ node1: NodeBase iface1: Optional[CoreInterface] node2: NodeBase iface2: Optional[CoreInterface] ptp: PtpNet = None label: str = None color: str = None def key(self) -> LinkKeyType: """ Retrieve the key for this link. :return: link key """ return create_key(self.node1, self.iface1, self.node2, self.iface2) def is_unidirectional(self) -> bool: """ Checks if this link is considered unidirectional, due to current iface configurations. :return: True if unidirectional, False otherwise """ unidirectional = False if self.iface1 and self.iface2: unidirectional = self.iface1.options != self.iface2.options return unidirectional def options(self) -> LinkOptions: """ Retrieve the options for this link. :return: options for this link """ if self.is_unidirectional(): options = self.iface1.options else: if self.iface1: options = self.iface1.options else: options = self.iface2.options return options def get_data(self, message_type: MessageFlags, source: str = None) -> LinkData: """ Create link data for this link. :param message_type: link data message type :param source: source for this data :return: link data """ iface1_data = self.iface1.get_data() if self.iface1 else None iface2_data = self.iface2.get_data() if self.iface2 else None return LinkData( message_type=message_type, type=LinkTypes.WIRED, node1_id=self.node1.id, node2_id=self.node2.id, iface1=iface1_data, iface2=iface2_data, options=self.options(), label=self.label, color=self.color, source=source, ) def get_data_unidirectional(self, source: str = None) -> LinkData: """ Create other unidirectional link data. :param source: source for this data :return: unidirectional link data """ iface1_data = self.iface1.get_data() if self.iface1 else None iface2_data = self.iface2.get_data() if self.iface2 else None return LinkData( message_type=MessageFlags.NONE, type=LinkTypes.WIRED, node1_id=self.node2.id, node2_id=self.node1.id, iface1=iface2_data, iface2=iface1_data, options=self.iface2.options, label=self.label, color=self.color, source=source, ) class LinkManager: """ Provides core link management. """ def __init__(self) -> None: """ Create a LinkManager instance. """ self._links: dict[LinkKeyType, CoreLink] = {} self._node_links: dict[int, dict[LinkKeyType, CoreLink]] = {} def add(self, core_link: CoreLink) -> None: """ Add a core link to be tracked. :param core_link: link to track :return: nothing """ node1, iface1 = core_link.node1, core_link.iface1 node2, iface2 = core_link.node2, core_link.iface2 if core_link.key() in self._links: raise CoreError( f"node1({node1.name}) iface1({iface1.id}) " f"node2({node2.name}) iface2({iface2.id}) link already exists" ) logger.info( "adding link from node(%s:%s) to node(%s:%s)", node1.name, iface1.name if iface1 else None, node2.name, iface2.name if iface2 else None, ) self._links[core_link.key()] = core_link node1_links = self._node_links.setdefault(node1.id, {}) node1_links[core_link.key()] = core_link node2_links = self._node_links.setdefault(node2.id, {}) node2_links[core_link.key()] = core_link def delete( self, node1: NodeBase, iface1: Optional[CoreInterface], node2: NodeBase, iface2: Optional[CoreInterface], ) -> CoreLink: """ Remove a link from being tracked. :param node1: first node in link :param iface1: node1 interface :param node2: second node in link :param iface2: node2 interface :return: removed core link """ key = create_key(node1, iface1, node2, iface2) if key not in self._links: raise CoreError( f"node1({node1.name}) iface1({iface1.id}) " f"node2({node2.name}) iface2({iface2.id}) is not linked" ) logger.info( "deleting link from node(%s:%s) to node(%s:%s)", node1.name, iface1.name if iface1 else None, node2.name, iface2.name if iface2 else None, ) node1_links = self._node_links[node1.id] node1_links.pop(key) node2_links = self._node_links[node2.id] node2_links.pop(key) return self._links.pop(key) def reset(self) -> None: """ Resets and clears all tracking information. :return: nothing """ self._links.clear() self._node_links.clear() def get_link( self, node1: NodeBase, iface1: Optional[CoreInterface], node2: NodeBase, iface2: Optional[CoreInterface], ) -> Optional[CoreLink]: """ Retrieve a link for provided values. :param node1: first node in link :param iface1: interface for node1 :param node2: second node in link :param iface2: interface for node2 :return: core link if present, None otherwise """ key = create_key(node1, iface1, node2, iface2) return self._links.get(key) def links(self) -> ValuesView[CoreLink]: """ Retrieve all known links :return: iterator for all known links """ return self._links.values() def node_links(self, node: NodeBase) -> ValuesView[CoreLink]: """ Retrieve all links for a given node. :param node: node to get links for :return: node links """ return self._node_links.get(node.id, {}).values()