Add an improved version of maps

Object oriented now, I hope it will be much simpler to use. Also fancy
design with annotators.

Keeping the old maps for now, since this is not tested at all. When the
snippet gets updated to use this, this will get renamed to maps.py
older_libs
LEdoian 2 years ago
parent c500a97fa6
commit 367bd82a08

@ -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
Loading…
Cancel
Save