diff --git a/birdvisu/maps_new.py b/birdvisu/maps_new.py new file mode 100644 index 0000000..316aa37 --- /dev/null +++ b/birdvisu/maps_new.py @@ -0,0 +1,199 @@ +from dataclasses import dataclass, field +from abc import ABC, abstractmethod +from typing import Sequence, Callable + +# The topologies provided by TopologyProviders are independent and do not have +# common routers, networks and links. We therefore join them on identifiers, +# which we expect to be canonical, somehow (i.e. the names BIRD has used) + +@dataclass +class Router: + ident: str + links: list['Link'] + # The full details in ospffile format. This allows extracting original + # details (e.g. visualisation details) from the router objects. + # FIXME: Topology handling does not need this, ideally this would be in + # some separate part of code. But everything else takes data from a + # CombinedTopology object, so we would still need a way of resolving this, + # and adding a separate object would complicate the code even more. + details: tuple[str,list] | None = None + # source objects for this Router in case of combined topology. Key is some + # source identifier + # FIXME: This allows for different identifiers in sources, which should be + # forbidden + # XXX: This is here, because Python mandates so: "TypeError: non-default + # argument 'links' follows default argument". I would rather have it right + # after `.ident`, but to no avail. + sources: dict[str, 'Router'] = field(default_factory=dict) + + @property + def all_details(self): + if self.details is not None: + yield self.details + for src in self.sources.values(): + for x in src.all_details: yield x + +@dataclass +class Network: + ident: str + links: list['Link'] + details: tuple[str,list] | None = None + sources: dict[str, 'Network'] = field(default_factory=dict) + + @property + def all_details(self): + if self.details is not None: + yield self.details + for src in self.sources.values(): + for x in src.all_details: yield x + +@dataclass +class Link: + """A connection from a Router to Network""" + router: Router + network: Network + metric: int + sources: dict[str, 'Link'] = field(default_factory=dict) + # Are there any other details that help describe a link? + +# This could be just a (named) tuple, dataclass is only used for consistency. +@dataclass +class Topology: + # Keys: idents + routers: dict[str, Router] = field(default_factory=dict) + networks: dict[str, Network] = field(default_factory=dict) + links: dict[tuple[str,str], Link] = field(default_factory=dict) # (router.ident, network.ident) -> Link + + # TODO: function for validating the topology, namely: + # - Routers and networks all know their links + # - Objects from this topology are referenced, not their sources nor other objects + # - Keys in dictionaries are same as identifiers of the values + # - maybe sth else… + +class TopologyProvider(ABC): + """Allows obtaining topology description from somewhere (usually OSPF + files)""" + + @abstractmethod + def get_topology(self) -> Topology: ... + +class TopologyCombiner: + """Takes multiple topologies (e.g. reference and actual one) and creates a + super-topology out of that. + + The rest of code relies mainly on such super-topologies""" + + def __init__(self): + self.topology = Topology() + + def add_topology(self, source_name: str, topo: Topology) -> None: + # We handle Routers and Networks pretty much the same way + for which_dict, which_class in [ + ('routers', Router), + ('networks', Network), + ]: + own_dict = getattr(self.topology, which_dict) + their_dict = getattr(topo, which_dict) + for nid, n in their_dict.items(): + assert nid == n.ident + if nid not in own_dict: + # FIXME: should only add complete objects. This will require + # different approach. + own_dict[nid] = which_class( + ident = nid, + links = [], + ) + assert source_name not in own_dict[nid].sources + own_dict[nid].sources[source_name] = n + # TODO: Can we avoid code duplication? + for lkey, l in topo.links.items(): + assert lkey == (l.router.ident, l.network.ident) + if lkey not in self.topology.links: + new_link = Link( + # Use own objects, not their children + router = self.topology.routers[l.router.ident], + network = self.topology.networks[l.network.ident], + metric = l.metric, + ) + self.topology.links[lkey] = new_link + # We add the link to our nodes + # TODO: This should ensure that only one link exists between + # this Router and Network, but currently it is based mostly on + # belief. + self.topology.routers[l.router.ident].links.append(new_link) + self.topology.networks[l.network.ident].links.append(new_link) + own_link = self.topology.links[lkey] + assert source_name not in own_link.sources + own_link.sources[source_name] = l + if own_link.metric != l.metric: + own_link.metric = -1 # Discrepant metrics! + + def get_complete_topology(self) -> Topology: + return self.topology + +@dataclass +class AnnotatedTopology: + topology: Topology + # Router.ident -> some tags. What they are is up to users / + # annotators. Similarly for other fields. + router_annotations: dict[str, list[object]] + network_annotations: dict[str, list[object]] + link_annotations: dict[tuple[str, str], list[object]] + # FIXME: this should rather be immutable, to allow saving different annotations. + +# TODO: maybe class to handle sets of annotators? +def annotate_topology(topology, + annotators: Sequence[Callable[[AnnotatedTopology], AnnotatedTopology]], + initial_annotation: AnnotatedTopology = None, + ) -> AnnotatedTopology: + """Runs all the annotators and assigns all the tags. + + The tags will change how the topology will be visualized. Examples: + - Which parts of network were expected and are missing + - Which paths are used in a routing tree of particular router + - Where should be nodes placed + + The annotators are run in order, therefore later annotators may depend on + pre-existing tags. This allows pipelining, for example to first analyse the + networks, then set styles, then layout the graph. An AnnotatedTopology + should therefore contain everything the visualisation (or other user) + needs. + + Initial topology contains all the dictionaries with all keys and empty lists. + + Annotators are not allowed to remove tags (by policy, code unfortunately + currently enables that.).""" + # This is maybe too general… + # The use of lists leaves namespacing to users, but can hinder performance + # in case of many tags. + # Removal of annotators is problematic, since tags are not tied to them. + # Unless users can distinguish tag ownership somehow (i.e. using own types + # for the tags), the only option is to recalculate the unannotated topology + # with a different set of annotators. + # TODO: We are going to abuse the fact that the annotators run in order and + # the tags are in lists to avoid searching through the whole list of tags – + # when we depend on previous annotator, its output is at the end. This + # should be more explicit (maybe an AnnotationPipeline object?), since this + # can easily be broken by misordering the annotators. + # - Idea: allow using `annotate_topology` itself as annotator. This way, a + # single call can hold the pipeline segment and can be managed + # separately (and possibly in more readable manner) + if initial_annotation is None: + initial_annotation = AnnotatedTopology( + topology = topology, + router_annotations = {k: [] for k in topology.routers.keys()}, + network_annotations = {k: [] for k in topology.networks.keys()}, + link_annotations = {k: [] for k in topology.links.keys()}, + ) + assert initial_annotation.topology is topology + # FIXME: probably shouldn't rely on topology having correct keys. + assert all(k in router_annotations for k in topology.routers.keys()) + assert all(k in network_annotations for k in topology.networks.keys()) + assert all(k in link_annotations for k in topology.links.keys()) + + for annotator in annotators: + new_annotation = annotator(initial_annotation) + assert isinstance(new_annotation, AnnotatedTopology) + initial_annotation = new_annotation + + return initial_annotation