From 9d9d14995e7c82e717ca15961a4b4de9bc15db1d Mon Sep 17 00:00:00 2001 From: Bob Lantz Date: Sun, 16 Oct 2022 15:09:29 -0700 Subject: [PATCH] Add node.Splitter() Splitter() is a simple splitter element that currently does not model coupling loss (although the split percents may be adjusted to model it as desired.) Note that Splitter() copies the signal to the split port if needed, creating a new signal that originates at the splitter. We do this in case this signal needs to propagate through one of the elements that the original signal propagates through. - add tests/splittertest.py - also remove dodgy intermediate signal state storage in OpticalSignal - also make assoc_loc_in/out more reliable by requiring at least one parameter - also remove unused include_optical_signal_in_roadm and references to it --- mnoptical/link.py | 9 ++- mnoptical/network.py | 73 ++++++++++--------- mnoptical/node.py | 146 ++++++++++++++++++++++++++------------ tests/splittertest.py | 158 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 304 insertions(+), 82 deletions(-) create mode 100755 tests/splittertest.py diff --git a/mnoptical/link.py b/mnoptical/link.py index f70396c9..3b4adc7a 100644 --- a/mnoptical/link.py +++ b/mnoptical/link.py @@ -41,9 +41,15 @@ def __init__(self, src_node, dst_node, src_out_port=-1, dst_in_port=-1, def connect(prev, component): "Connect previous component to component" if prev != src_node: + if prev.next_component: + raise ValueError( + f'{self}: {prev} already connected to {prev.next_component}') prev.next_component = component prev.set_output_port(component, self, output_port=0) if component != dst_node: + if component.link: + raise ValueError( + f'{self}: {component} already used in {component.link}') component.link = self component.prev_component = prev component.set_input_port(prev, self, input_port=0) @@ -145,7 +151,8 @@ def propagate(self, is_last_port=False, safe_switch=False): """ first_component = self.boost_amp or self.spans[0][0] for optical_signal in self.optical_signals: - first_component.include_optical_signal_in(optical_signal, in_port=0) + state = optical_signal.loc_in_to_state[self] + first_component.include_optical_signal_in(optical_signal, **state, in_port=0) first_component.propagate(optical_signals=self.optical_signals, is_last_port=is_last_port, safe_switch=safe_switch) diff --git a/mnoptical/network.py b/mnoptical/network.py index 8ae6f420..932ceefe 100644 --- a/mnoptical/network.py +++ b/mnoptical/network.py @@ -18,59 +18,56 @@ def __init__(self): self.name_to_node = {} - def add_lt(self, name, transceivers=None, **params): + def add_node(self, name, cls, *args, nodes=None, **params): + """ + Add node to network + :param name: name of node + :param cls: node class/constructor + :param nodes: list that node will be appended to + """ + if nodes is None: nodes = [] + if name in self.name_to_node: + raise ValueError(f"Network.add_node: node {name} already exists!") + node = cls(name, *args, **params) + self.name_to_node[name] = node + nodes.append(node) + self.topology[node] = [] + return node + + def add_lt(self, name, *args, cls=LineTerminal, **params): """ Add lt node :param name: name of lt :param transceivers: transceivers of LT for automated instantiation + :param cls: optional LineTerminal class/constructor :return: added lt """ - if name in self.name_to_node: - raise ValueError("Network.add_lt: lt with this name already exist!") - configs = {'name': name, - 'transceivers': transceivers} - configs.update(params) - lt = LineTerminal(**configs) - self.name_to_node[name] = lt - self.line_terminals.append(lt) - self.topology[lt] = [] - return lt + return self.add_node(name, cls, *args, nodes=self.line_terminals, + **params) - def add_roadm(self, name, **params): + def add_roadm(self, name, *args, cls=Roadm, **params): """ Add a ROADM node. :param name: name of ROADM + :cls: optional Roadm class/constructor :return: added ROADM """ - if name in self.name_to_node: - raise ValueError("Network.add_roadm: ROADM with this name already exist!!") - configs = {'name': name} - configs.update(params) - roadm = Roadm(**configs) - self.name_to_node[name] = roadm - self.roadms.append(roadm) - self.topology[roadm] = [] - return roadm + return self.add_node(name, cls, *args, nodes=self.roadms, + **params) - def add_amplifier(self, name, amplifier_type='EDFA', **params): + def add_amplifier(self, name, *args, cls=Amplifier, **params): """ Add an Amplifier node. :param name: name of Amplifier :param amplifier_type: amplifier type (currently supporting only EDFA) + :param cls: optional Amplifier class/constructor :return: added Amplifier """ - if name in self.name_to_node: - raise ValueError("Network.add_amplifier: Amplifier with this name already exist!! %s" % str(name)) - configs = {'name': name, - 'amplifier_type': amplifier_type} - configs.update(params) - amplifier = Amplifier(**configs) - self.name_to_node[name] = amplifier - self.amplifiers.append(amplifier) - return amplifier + return self.add_node(name, cls, *args, nodes=self.amplifiers, + **params) def add_link(self, src_node, dst_node, src_out_port=-1, - dst_in_port=-1, boost_amp=None, spans=None): + dst_in_port=-1, boost_amp=None, spans=None, cls=Link): """ Add a uni-directional link :param src_node: source node in link @@ -79,14 +76,14 @@ def add_link(self, src_node, dst_node, src_out_port=-1, :param dst_in_port: dst_node input port :param boost_amp: optional amplifier object for boost_amplification :param spans: + :param cls: optional Link class/constructor :return: created and added link """ - link = Link(src_node, dst_node, - src_out_port=src_out_port, - dst_in_port=dst_in_port, - boost_amp=boost_amp, - spans=spans) - + link = cls(src_node, dst_node, + src_out_port=src_out_port, + dst_in_port=dst_in_port, + boost_amp=boost_amp, + spans=spans) self.links.append(link) self.topology[src_node].append((dst_node, link)) return link diff --git a/mnoptical/node.py b/mnoptical/node.py index e224e46b..fe41cce8 100755 --- a/mnoptical/node.py +++ b/mnoptical/node.py @@ -5,6 +5,7 @@ from pprint import pprint import random from collections import namedtuple +from copy import copy from scipy.special import erfc from math import sqrt @@ -13,9 +14,9 @@ class Node(object): input_port_base = 0 output_port_base = 0 debugger = True # Print debugger messages by default - + def __init__(self, name, debugger=None): - + self.name = name if debugger is not None: self.debugger = debugger @@ -126,7 +127,8 @@ def remove_optical_signal(self, optical_signal): for out_port, optical_signals in port_to_optical_signal_out_copy.items(): if optical_signal in optical_signals: self.port_to_optical_signal_out[out_port].remove(optical_signal) - if not isinstance(self, Amplifier): + # Was: not isinstance(self, Amplifier) + if not hasattr(self, 'target_gain'): link = self.port_to_link_out[out_port] link.remove_optical_signal(optical_signal) @@ -288,18 +290,19 @@ def assoc_tx_to_channel(self, transceiver, channel, out_port=-1): def assoc_channel(self, transceiver, channel, out_port): # instantiate OpticalSignal object + power = transceiver.operation_power optical_signal = OpticalSignal(channel, transceiver.channel_spacing_H, transceiver.channel_spacing_nm, transceiver.modulation_format, transceiver.symbol_rate, transceiver.bits_per_symbol, - power=transceiver.operation_power) + power=power) # associate transceiver to optical_signal transceiver.assoc_optical_signal(optical_signal) dst_node = self.port_to_node_out[out_port] # the goal of this function - self.include_optical_signal_out(optical_signal, out_port=out_port) + self.include_optical_signal_out(optical_signal, power=power, out_port=out_port) self.tx_to_channel[out_port] = {'optical_signal': optical_signal, 'transceiver': transceiver} self.optical_signals_out += 1 @@ -352,15 +355,18 @@ def turn_on(self, safe_switch=False): for signal_count, out_port in enumerate(self.tx_to_channel, start=1): optical_signal = self.tx_to_channel[out_port]['optical_signal'] transceiver = self.tx_to_channel[out_port]['transceiver'] - transceiver.optical_signal.reset(component=self) + transceiver.optical_signal.reset() + # Originate signal here + power = transceiver.operation_power + self.include_optical_signal_out(optical_signal, power=power) if self.debugger: print("*** %s.turn_on %s on port %s" % (self, optical_signal, out_port)) - # pass signal info to link + # Pass signal info to link + # Note: power is always in watts link = self.port_to_link_out[out_port] - link.include_optical_signal_in(optical_signal) - + link.include_optical_signal_in(optical_signal, power=power) if signal_count == self.optical_signals_out: link.propagate(is_last_port=True, safe_switch=safe_switch) else: @@ -538,22 +544,19 @@ def assoc_loc_in(self, loc, power=None, ase_noise=None, nli_noise=None): Associate a location to signal performance values at the input interface of this point :param loc: location (i.e., node, span) - :param power: power levels [mW] (or None for default/launch state) - :param ase_noise: ase levels [mW] (or None for default/launch state) - :param nli_noise: nli levels [mW] (or None for default/launch state) - """ - if power is None: - power = self.power - if ase_noise is None: - ase_noise = self.ase_noise - if nli_noise is None: - nli_noise = self.nli_noise - # XXX: We probably shouldn't update the default/launch state to - # some random input state in the network, should we? - self.power = power - self.ase_noise = ase_noise - self.nli_noise = nli_noise - self.loc_in_to_state[loc] = {'power': power, 'ase_noise': ase_noise, 'nli_noise': nli_noise} + :param power: power levels [mW] + :param ase_noise: ase levels [mW] + :param nli_noise: nli levels [mW] + """ + assert (power, ase_noise, nli_noise) != (None, None, None) + state = self.loc_in_to_state.setdefault( + loc, dict(power=0, ase_noise=0, nli_noise=0)) + if power is not None: + state['power'] = power + if ase_noise is not None: + state['ase_noise'] = ase_noise + if nli_noise is not None: + state['nli_noise'] = nli_noise def assoc_loc_out(self, loc, power=None, ase_noise=None, nli_noise=None): """ @@ -564,18 +567,15 @@ def assoc_loc_out(self, loc, power=None, ase_noise=None, nli_noise=None): :param ase_noise: ase levels [mW] (or None for default/launch state) :param nli_noise: nli levels [mW] (or None for default/launch state) """ - if power is None: - power = self.power - if ase_noise is None: - ase_noise = self.ase_noise - if nli_noise is None: - nli_noise = self.nli_noise - # XXX: We probably shouldn't update the default/launch state to - # some random input state in the network, should we? - self.power = power - self.ase_noise = ase_noise - self.nli_noise = nli_noise - self.loc_out_to_state[loc] = {'power': power, 'ase_noise': ase_noise, 'nli_noise': nli_noise} + assert (power, ase_noise, nli_noise) != (None, None, None) + state = self.loc_out_to_state.setdefault( + loc, dict(power=0, ase_noise=0, nli_noise=0)) + if power is not None: + state['power'] = power + if ase_noise is not None: + state['ase_noise'] = ase_noise + if nli_noise is not None: + state['nli_noise'] = nli_noise def reset(self, component=None): """ @@ -587,7 +587,7 @@ def reset(self, component=None): self.ase_noise = self.ase_noise_start self.nli_noise = self.nli_noise_start # Reset signal state at all components, optionally - # presrving state at originating component + # preserving state at originating component self.loc_in_to_state = {} if component and component in self.loc_out_to_state: self.loc_out_to_state = {component: self.loc_out_to_state[component]} @@ -871,6 +871,7 @@ def can_switch(self, in_port, safe_switch): port_out_to_port_in_signals = {} # iterate through the optical signals that are currently at # in_port (if any) + self.port_to_optical_signal_in.setdefault(in_port, []) for optical_signal in self.port_to_optical_signal_in[in_port]: # check if there is a switching rule for a signal switch_rule = False @@ -890,7 +891,8 @@ def can_switch(self, in_port, safe_switch): port_out_to_port_in_signals[out_port][in_port].append(optical_signal) if not switch_rule: if self.debugger: - print(self, "Unable to find switch rule for signal:", optical_signal) + print(self, "Unable to find switch rule for signal:", + optical_signal, "input port:", in_port) port_to_optical_signal_out_copy = port_to_optical_signal_out.copy() # Check if there are other signals being switched at these output ports. @@ -900,7 +902,7 @@ def can_switch(self, in_port, safe_switch): # used by the switching rules found for the input port. for out_port, optical_signals in port_to_optical_signal_out_copy.items(): # iterate through all signals at an output port - for optical_signal in self.port_to_optical_signal_out[out_port]: + for optical_signal in self.port_to_optical_signal_out.get(out_port, []): # add the adjecent signals to the dictionaries if optical_signal not in optical_signals: port_to_optical_signal_out[out_port].append(optical_signal) @@ -916,6 +918,7 @@ def can_switch(self, in_port, safe_switch): if len(port_to_optical_signal_out) > 0: # iterate through the output ports for out_port, optical_signals in port_to_optical_signal_out_copy.items(): + if out_port not in self.port_to_optical_signal_out: continue # check if optical_signals == self.port_to_optical_signal_out[out_port] # check if the power levels have changed if all(optical_signal in optical_signals for optical_signal in @@ -924,8 +927,8 @@ def can_switch(self, in_port, safe_switch): self.port_check_range_out[out_port] += 1 if not self.power_divergence(optical_signals, in_port): if self.port_check_range_out[out_port] > self.check_range_th: - # these signals can be safely terminated at a LineTerminal - if not isinstance(self.port_to_node_out[out_port], LineTerminal): + # these signals can be safely received at a LineTerminal + if not hasattr(self.port_to_node_out[out_port], 'receiver'): if self.debugger: print('RoadmWarning:', self, "same signals already propagated on this output " "port. Stopping propagation.") @@ -986,7 +989,8 @@ def switch(self, in_port, src_node, safe_switch=False): Note: check for switch feasibility unless performing tasks independent of switching (i.e., EDFA gain configuration). """ - if isinstance(src_node, LineTerminal): + # Was: isinstance(src_node, LineTerminal) + if hasattr(src_node, 'transceivers'): # need to check for all (possible) input ports coming from LineTerminal port_to_optical_signal_out, port_out_to_port_in_signals = self.can_switch_from_lt(src_node, safe_switch) else: @@ -1491,6 +1495,62 @@ def propagate(self, optical_signals, is_last_port=False, safe_switch=False): self.next_component.propagate(is_last_port=is_last_port, safe_switch=safe_switch) +class Splitter(Node): + """ + Simple static splitter, for now without coupling loss. + Input at port 0 is split among output ports according + to the split array, e.g. {1: 99, 2:1} indicating + a 99%/1% split for output ports 1 and 2. + + There is some flexibility since the percents do not + need to add up to 100, though they should not exceed + 100 since this is really not supposed to be an amplifier! + + There is currently no error checking. + + We use the same ingress API/protocol as Roadm. + """ + + def __init__(self, name, split=None, monitor_mode='out'): + "split: {port:percent...}" + super().__init__(name) + self.split = split or {} + self.monitor = Monitor(name + "-monitor", component=self, mode=monitor_mode) + + def switch(self, in_port, src_node, safe_switch=False): + """Propagate splitter's signals to its output ports. + This is part of the Roadm protocol that we conform to.""" + siglists = self.port_to_optical_signal_in.values() + assert len(siglists) == 1, f"{self}: a Splitter can only have one input port" + signals = tuple(siglists)[0] + # Compute output signals + for signal in signals: + state = signal.loc_in_to_state[self] + power_in, ase_in, nli_in = state['power'], state['ase_noise'], state['nli_noise'] + for port in self.ports_out: + indices = {sig.index:sig for sig in self.port_to_optical_signal_out[port]} + if port != self.ports_out[0]: + # Reuse or copy signal as necessary + if signal.index in indices: + signal = indices[signal.index] + else: + signal = copy(signal) + signal.reset() + fraction = self.split.get(port, 0.0) / 100.0 + power_out = power_in * fraction + ase_out = ase_in * fraction + nli_out = nli_in * fraction + self.include_optical_signal_out( + signal, power=power_out, ase_noise=ase_out, nli_noise=nli_out, out_port=port) + link = self.port_to_link_out.get(port, None) + if link: + link.include_optical_signal_in( + signal, power=power_out, ase_noise=ase_out, nli_noise=nli_out) + # Propagate to output links + for port, link in self.port_to_link_out.items(): + link.propagate(is_last_port=True, safe_switch=safe_switch) + + class Monitor(Node): """ This implementation of Monitors could be used for ROADMs and Amplifiers. diff --git a/tests/splittertest.py b/tests/splittertest.py new file mode 100755 index 00000000..a876d0c8 --- /dev/null +++ b/tests/splittertest.py @@ -0,0 +1,158 @@ +#!/usr/bin/python3 +""" +splittertest.py: test splitter element + +We create a 99/1 power split and verify that it +is working as expected. +""" + +from mnoptical.network import Network +from mnoptical.link import Span as Fiber, SpanTuple as Segment +# Note: SpanTuple/Segment is a (Fiber(), Amplifier()) tuple +from mnoptical.node import ( + Transceiver, Roadm, LineTerminal, Splitter, Amplifier) +from mnoptical.units import abs_to_db + +# Units +km = dB = dBm = 1.0 +m = .001 + +# Terminal TX/RX port numbers (arbitrary for simulation) +TX, RX = 100, 200 + +# Splitter input and output ports +IN0, OUT1, OUT2 = 0, 1, 2 + +# Roadm line in/out and add/drop base +LINEIN, LINEOUT, ADD, DROP = 0, 1, RX, TX + +# Number of transceivers +txcount = 3 + +# Span helper function +def Span(net, length, amp='', **params): + if amp: amp = net.add_amplifier(ampname, **params) + return Segment(span=Fiber(length=length), amplifier=amp) + +# Network topology +def createnetwork(): + """Simple test network t1 -> r1 -> splitter(99/1) -> r3 -> t3 + -> r2 -> t2 """ + net = Network() + def span(*args, **params): return Span(net, *args, **params) + + # Nodes: terminals, ROADMs, splitter + for i in 1, 2, 3: + transceivers = [Transceiver(i,f'tx{i}',0*dBm) + for i in range(1, txcount+1)] + net.add_lt(f't{i}', transceivers, monitor_mode='in') + net.add_roadm(f'r{i}') + t1, t2, t3 = net.line_terminals + r1, r2, r3 = net.roadms + sp1 = net.add_node('sp1', cls=Splitter, split={OUT1:99, OUT2:1}) + + # Links: local add/drop(1m), WAN(25km), splitter outputs (1m) + L = net.add_link + for ch in range(1, txcount+1): + L(t1, r1, TX+ch, ADD+ch, spans=[span(1*m)]) + L(r2, t2, DROP+ch, RX+ch, spans=[span(1*m)]) + L(r3, t3, DROP+ch, RX+ch,spans=[span(1*m)]) + L(r1, sp1, LINEOUT, IN0, spans=[span(25*km)], + boost_amp=net.add_amplifier('boost1', boost=True, target_gain=0.0)) + L(sp1, r2, OUT1, LINEIN, spans=[span(1*m)]) + L(sp1, r3, OUT2, LINEIN, spans=[span(1*m)]) + + return net + +# Monitoring helper functions +def getsignalwatts(node, port=None): + "Return monitored signal, ase noise, and nli noise power in watts" + monitor = node.monitor + return {s.index: {'pwrW': monitor.get_power(s), + 'aseW': monitor.get_ase_noise(s), + 'nliW': monitor.get_nli_noise(s)} + for s in monitor.get_optical_signals(port)} + +def wtodbm(W): + "Return watts as dBm" + return abs_to_db(W*1000.0) if W != 0 else float('-inf') + +def printdbm(sigwatts, fn=wtodbm, units='dBm'): + "Print signal watts as dBm" + for ch, entries in sigwatts.items(): + pdbm = fn(entries['pwrW']) + adbm = fn(entries['aseW']) + ndbm = fn(entries['nliW']) + print(f'ch{ch}: pwr {pdbm:.2e}{units} ' + f'ase {adbm:.2e}{units} nli {ndbm:.2e}{units}') + +def printmw(sigwatts): + "Print signal watts as milliwatts" + printdbm(sigwatts, fn=lambda W: W*1e3, units='mW') + +# Configuration + +def configroadms(net): + "Configure ROADMs (as mux or demux)" + r1, r2, r3 = net.roadms + for ch in range(1, txcount+1): + r1.install_switch_rule(ADD+ch, LINEOUT, ch) + r2.install_switch_rule(LINEIN, DROP+ch, ch) + r3.install_switch_rule(LINEIN, DROP+ch, ch) + +def configterms(net): + "Configure terminals" + t1, t2, t3 = net.line_terminals + # Set TX and RX channels + for t in range(1, txcount+1): + ch = t # channel number == transceiver number + tx1 = t1.id_to_transceivers[t] + rx2 = t2.id_to_transceivers[t] + rx3 = t3.id_to_transceivers[t] + t1.assoc_tx_to_channel(tx1, ch, out_port=TX+ch) + t2.assoc_rx_to_channel(rx2, ch, in_port=RX+ch) + t3.assoc_rx_to_channel(rx3, ch, in_port=RX+ch) + # Start transmission + for term in t1, t2, t3: + term.turn_on() + +def ratio(a, b): + "Return percent ratio of a to b" + total = a+b + apct = int(100.0 * a/total + .5) + bpct = int(100.0 * b/total + .5) + return apct, bpct + +def test(): + "Verify that a 99/1 split works as expected" + # Create and configure network + net = createnetwork() + configroadms(net) + configterms(net) + # Check power + t1, t2, t3 = net.line_terminals + print(f"*** {t2} power") + t2watts = getsignalwatts(t2) + printmw(t2watts) + print(f"*** {t3} power") + t3watts = getsignalwatts(t3) + printmw(t3watts) + print(f"*** Checking split ratios") + errors = 0 + for ch in range(1,4): + for f in 'pwr', 'ase': + w = f'{f}W' + a, b = ratio(t2watts[ch][w], t3watts[ch][w]) + print(f"ch{ch} {f} split is {a}/{b}") + if (a,b) != (99,1): + print(f'Expected 99/1 split but got {a}/{b}') + errors += 1 + print(f"*** Completed with {errors} errors.") + if errors: + sp1 = net.name_to_node['sp1'] + sp1.debug() + return errors + +if __name__ == '__main__': + result = test() + exit(result)