scottk/oopt-gnpy
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/oopt-gnpy.git synced 2025-10-29 17:22:42 +00:00
Code Issues Packages Projects Releases 10 Wiki Activity

docs: docstring formatting

Browse Source 
Let's use the pythonic indenting, quoting and structure in general as
specified in PEP 0257.

Change-Id: Icd0b4fbd94dabd9a163ae3f6887b236e76c486ab
This commit is contained in:
Jan Kundrát
2023-04-18 01:32:47 +02:00
parent 2a07eec966
commit 76e9146043
25 changed files with 228 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
gnpy/__init__.py
View File

@@ -1,8 +1,8 @@
'''
"""
GNPy is an open-source, community-developed library for building route planning and optimization tools in real-world mesh optical networks. It is based on the Gaussian Noise Model.
Signal propagation is implemented in :py:mod:`.core`.
Path finding and spectrum assignment is in :py:mod:`.topology`.
Various tools and auxiliary code, including the JSON I/O handling, is in
:py:mod:`.tools`.
'''
"""

4
gnpy/core/__init__.py
View File

@@ -1,4 +1,4 @@
'''
"""
Simulation of signal propagation in the DWDM network
Optical signals, as defined via :class:`.info.SpectralInformation`, enter
@@ -6,4 +6,4 @@ Optical signals, as defined via :class:`.info.SpectralInformation`, enter
through the :py:mod:`.network`.
The simulation is controlled via :py:mod:`.parameters` and implemented mainly
via :py:mod:`.science_utils`.
'''
"""

4
gnpy/core/ansi_escapes.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
"""
gnpy.core.ansi_escapes
======================
A random subset of ANSI terminal escape codes for colored messages
'''
"""
red = '\x1b[1;31;40m'
blue = '\x1b[1;34;40m'

6
gnpy/core/elements.py
View File

@@ -42,11 +42,11 @@ class Location(namedtuple('Location', 'latitude longitude city region')):
class _Node:
'''Convenience class for providing common functionality of all network elements
"""Convenience class for providing common functionality of all network elements
This class is just an internal implementation detail; do **not** assume that all network elements
inherit from :class:`_Node`.
'''
"""
def __init__(self, uid, name=None, params=None, metadata=None, operational=None, type_variety=None):
if name is None:
name = uid
@@ -92,7 +92,7 @@ class Transceiver(_Node):
self.propagated_labels = [""]
def _calc_cd(self, spectral_info):
""" Updates the Transceiver property with the CD of the received channels. CD in ps/nm.
"""Updates the Transceiver property with the CD of the received channels. CD in ps/nm.
"""
self.chromatic_dispersion = spectral_info.chromatic_dispersion * 1e3

4
gnpy/core/equipment.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
"""
gnpy.core.equipment
===================
This module contains functionality for specifying equipment.
'''
"""
from gnpy.core.utils import automatic_nch, db2lin
from gnpy.core.exceptions import EquipmentConfigError

27
gnpy/core/info.py
View File

@@ -29,21 +29,23 @@ class Power(namedtuple('Power', 'signal nli ase')):
class Channel(namedtuple('Channel',
'channel_number frequency baud_rate slot_width roll_off power chromatic_dispersion pmd pdl')):
""" Class containing the parameters of a WDM signal.
:param channel_number: channel number in the WDM grid
:param frequency: central frequency of the signal (Hz)
:param baud_rate: the symbol rate of the signal (Baud)
:param slot_width: the slot width (Hz)
:param roll_off: the roll off of the signal. It is a pure number between 0 and 1
:param power (gnpy.core.info.Power): power of signal, ASE noise and NLI (W)
:param chromatic_dispersion: chromatic dispersion (s/m)
:param pmd: polarization mode dispersion (s)
:param pdl: polarization dependent loss (dB)
"""Class containing the parameters of a WDM signal.
:param channel_number: channel number in the WDM grid
:param frequency: central frequency of the signal (Hz)
:param baud_rate: the symbol rate of the signal (Baud)
:param slot_width: the slot width (Hz)
:param roll_off: the roll off of the signal. It is a pure number between 0 and 1
:param power (gnpy.core.info.Power): power of signal, ASE noise and NLI (W)
:param chromatic_dispersion: chromatic dispersion (s/m)
:param pmd: polarization mode dispersion (s)
:param pdl: polarization dependent loss (dB)
"""
class Pref(namedtuple('Pref', 'p_span0, p_spani, ref_carrier')):
"""noiseless reference power in dBm:
p_span0: inital target carrier power for a reference channel defined by user
p_spani: carrier power after element i for a reference channel defined by user
ref_carrier records the baud rate of the reference channel
@@ -51,7 +53,8 @@ class Pref(namedtuple('Pref', 'p_span0, p_spani, ref_carrier')):
class SpectralInformation(object):
""" Class containing the parameters of the entire WDM comb.
"""Class containing the parameters of the entire WDM comb.
delta_pdb_per_channel: (per frequency) per channel delta power in dbm for the actual mix of channels"""
def __init__(self, frequency: array, baud_rate: array, slot_width: array, signal: array, nli: array, ase: array,
@@ -305,7 +308,7 @@ def create_arbitrary_spectral_information(frequency: Union[ndarray, Iterable, fl
def create_input_spectral_information(f_min, f_max, roll_off, baud_rate, power, spacing, tx_osnr, ref_carrier=None):
""" Creates a fixed slot width spectral information with flat power.
"""Creates a fixed slot width spectral information with flat power.
all arguments are scalar values"""
number_of_channels = automatic_nch(f_min, f_max, spacing)
frequency = [(f_min + spacing * i) for i in range(1, number_of_channels + 1)]

7
gnpy/core/network.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
"""
gnpy.core.network
=================
Working with networks which consist of network elements
'''
"""
from operator import attrgetter
from gnpy.core import ansi_escapes, elements
@@ -235,8 +235,7 @@ def set_amplifier_voa(amp, power_target, power_mode):
def set_egress_amplifier(network, this_node, equipment, pref_ch_db, pref_total_db):
""" this node can be a transceiver or a ROADM (same function called in both cases)
"""
"""this node can be a transceiver or a ROADM (same function called in both cases)"""
power_mode = equipment['Span']['default'].power_mode
ref_carrier = ReferenceCarrier(baud_rate=equipment['SI']['default'].baud_rate,
slot_width=equipment['SI']['default'].spacing)

6
gnpy/core/parameters.py
View File

@@ -35,7 +35,8 @@ class PumpParams(Parameters):
class RamanParams(Parameters):
def __init__(self, flag=False, result_spatial_resolution=10e3, solver_spatial_resolution=50):
""" Simulation parameters used within the Raman Solver
"""Simulation parameters used within the Raman Solver
:params flag: boolean for enabling/disable the evaluation of the Raman power profile in frequency and position
:params result_spatial_resolution: spatial resolution of the evaluated Raman power profile
:params solver_spatial_resolution: spatial step for the iterative solution of the first order ode
@@ -48,7 +49,8 @@ class RamanParams(Parameters):
class NLIParams(Parameters):
def __init__(self, method='gn_model_analytic', dispersion_tolerance=1, phase_shift_tolerance=0.1,
computed_channels=None):
""" Simulation parameters used within the Nli Solver
"""Simulation parameters used within the Nli Solver
:params method: formula for NLI calculation
:params dispersion_tolerance: tuning parameter for ggn model solution
:params phase_shift_tolerance: tuning parameter for ggn model solution

18
gnpy/core/science_utils.py
View File

@@ -27,7 +27,7 @@ logger = getLogger(__name__)
sim_params = SimParams()
def raised_cosine_comb(f, *carriers):
""" Returns an array storing the PSD of a WDM comb of raised cosine shaped
"""Returns an array storing the PSD of a WDM comb of raised cosine shaped
channels at the input frequencies defined in array f
:param f: numpy array of frequencies in Hz
@@ -271,11 +271,11 @@ class RamanSolver:
class NliSolver:
""" This class implements the NLI models.
Model and method can be specified in `sim_params.nli_params.method`.
List of implemented methods:
'gn_model_analytic': eq. 120 from arXiv:1209.0394
'ggn_spectrally_separated': eq. 21 from arXiv: 1710.02225 spectrally separated
"""This class implements the NLI models.
Model and method can be specified in `sim_params.nli_params.method`.
List of implemented methods:
'gn_model_analytic': eq. 120 from arXiv:1209.0394
'ggn_spectrally_separated': eq. 21 from arXiv: 1710.02225 spectrally separated
"""
@staticmethod
@@ -287,7 +287,7 @@ class NliSolver:
@staticmethod
def compute_nli(spectral_info: SpectralInformation, srs: StimulatedRamanScattering, fiber):
""" Compute NLI power generated by the WDM comb `*carriers` on the channel under test `carrier`
"""Compute NLI power generated by the WDM comb `*carriers` on the channel under test `carrier`
at the end of the fiber span.
"""
logger.debug('Start computing fiber NLI noise')
@@ -310,7 +310,7 @@ class NliSolver:
# Methods for computing GN-model
@staticmethod
def _gn_analytic(spectral_info: SpectralInformation, alpha, beta2, gamma, length):
""" Computes the nonlinear interference power evaluated at the fiber input.
"""Computes the nonlinear interference power evaluated at the fiber input.
The method uses eq. 120 from arXiv:1209.0394
"""
spm_weight = (16.0 / 27.0) * gamma ** 2
@@ -350,7 +350,7 @@ class NliSolver:
@staticmethod
def _ggn_spectrally_separated(spectral_info: SpectralInformation, srs: StimulatedRamanScattering,
alpha, beta2, beta3, f_ref_beta, gamma):
""" Computes the nonlinear interference power evaluated at the fiber input.
"""Computes the nonlinear interference power evaluated at the fiber input.
The method uses eq. 21 from arXiv: 1710.02225
"""
dispersion_tolerance = sim_params.nli_params.dispersion_tolerance

18
gnpy/core/utils.py
View File

@@ -213,7 +213,7 @@ freq2wavelength = constants.nu2lambda
def freq2wavelength(value):
""" Converts frequency units to wavelength units.
"""Converts frequency units to wavelength units.
>>> round(freq2wavelength(191.35e12) * 1e9, 3)
1566.723
@@ -247,8 +247,7 @@ def per_label_average(values, labels):
def pretty_summary_print(summary):
"""Build a prettty string that shows the summary dict values per label with 2 digits
"""
"""Build a prettty string that shows the summary dict values per label with 2 digits"""
if len(summary) == 1:
return f'{list(summary.values())[0]:.2f}'
text = ', '.join([f'{label}: {value:.2f}' for label, value in summary.items()])
@@ -256,7 +255,7 @@ def pretty_summary_print(summary):
def deltawl2deltaf(delta_wl, wavelength):
""" deltawl2deltaf(delta_wl, wavelength):
"""deltawl2deltaf(delta_wl, wavelength):
delta_wl is BW in wavelength units
wavelength is the center wl
units for delta_wl and wavelength must be same
@@ -274,9 +273,9 @@ def deltawl2deltaf(delta_wl, wavelength):
def deltaf2deltawl(delta_f, frequency):
""" deltawl2deltaf(delta_f, frequency):
converts delta frequency to delta wavelength
units for delta_wl and wavelength must be same
"""convert delta frequency to delta wavelength
Units for delta_wl and wavelength must be same.
:param delta_f: delta frequency in same units as frequency
:param frequency: frequency BW is relevant for
@@ -291,8 +290,7 @@ def deltaf2deltawl(delta_f, frequency):
def rrc(ffs, baud_rate, alpha):
""" rrc(ffs, baud_rate, alpha): computes the root-raised cosine filter
function.
"""compute the root-raised cosine filter function
:param ffs: A numpy array of frequencies
:param baud_rate: The Baud Rate of the System
@@ -318,7 +316,7 @@ def rrc(ffs, baud_rate, alpha):
def merge_amplifier_restrictions(dict1, dict2):
"""Updates contents of dicts recursively
"""Update contents of dicts recursively
>>> d1 = {'params': {'restrictions': {'preamp_variety_list': [], 'booster_variety_list': []}}}
>>> d2 = {'params': {'target_pch_out_db': -20}}

4
gnpy/tools/__init__.py
View File

@@ -1,5 +1,5 @@
'''
"""
Processing of data via :py:mod:`.json_io`.
Utilities for Excel conversion in :py:mod:`.convert` and :py:mod:`.service_sheet`.
Example code in :py:mod:`.cli_examples` and :py:mod:`.plots`.
'''
"""

6
gnpy/tools/cli_examples.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
"""
gnpy.tools.cli_examples
=======================
Common code for CLI examples
'''
"""
import argparse
import logging
@@ -48,7 +48,7 @@ def show_example_data_dir():
def load_common_data(equipment_filename, topology_filename, simulation_filename, save_raw_network_filename):
'''Load common configuration from JSON files'''
"""Load common configuration from JSON files"""
try:
equipment = load_equipment(equipment_filename)

19
gnpy/tools/json_io.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
"""
gnpy.tools.json_io
==================
Loading and saving data from JSON files in GNPy's internal data format
'''
"""
from networkx import DiGraph
from logging import getLogger
@@ -368,9 +368,7 @@ def _update_dual_stage(equipment):
def _roadm_restrictions_sanity_check(equipment):
""" verifies that booster and preamp restrictions specified in roadm equipment are listed
in the edfa.
"""
"""verifies that booster and preamp restrictions specified in roadm equipment are listed in the edfa."""
restrictions = equipment['Roadm']['default'].restrictions['booster_variety_list'] + \
equipment['Roadm']['default'].restrictions['preamp_variety_list']
for amp_name in restrictions:
@@ -440,11 +438,11 @@ def load_network(filename, equipment):
def save_network(network: DiGraph, filename: str):
'''Dump the network into a JSON file
"""Dump the network into a JSON file
:param network: network to work on
:param filename: file to write to
'''
"""
save_json(network_to_json(network), filename)
@@ -538,8 +536,7 @@ def save_json(obj, filename):
def load_requests(filename, eqpt, bidir, network, network_filename):
""" loads the requests from a json or an excel file into a data string
"""
"""loads the requests from a json or an excel file into a data string"""
if filename.suffix.lower() in ('.xls', '.xlsx'):
_logger.info('Automatically converting requests from XLS to JSON')
try:
@@ -648,8 +645,8 @@ def _check_one_request(params, f_max_from_si):
def disjunctions_from_json(json_data):
""" reads the disjunction requests from the json dict and create the list
of requested disjunctions for this set of requests
"""reads the disjunction requests from the json dict and create the list
of requested disjunctions for this set of requests
"""
disjunctions_list = []
if 'synchronization' in json_data:

4
gnpy/tools/plots.py
View File

@@ -1,12 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
"""
gnpy.tools.plots
================
Graphs and plots usable from a CLI application
'''
"""
from matplotlib.pyplot import show, axis, figure, title, text
from networkx import draw_networkx

4
gnpy/topology/__init__.py
View File

@@ -1,3 +1,3 @@
'''
"""
Tracking :py:mod:`.request` for spectrum and their :py:mod:`.spectrum_assignment`.
'''
"""

77
gnpy/topology/request.py
View File

@@ -41,8 +41,7 @@ DisjunctionParams = namedtuple('DisjunctionParams', 'disjunction_id relaxable li
class PathRequest:
""" the class that contains all attributes related to a request
"""
"""the class that contains all attributes related to a request"""
def __init__(self, *args, **params):
params = RequestParams(**params)
@@ -104,8 +103,7 @@ class PathRequest:
class Disjunction:
""" the class that contains all attributes related to disjunction constraints
"""
"""the class that contains all attributes related to disjunction constraints"""
def __init__(self, *args, **params):
params = DisjunctionParams(**params)
@@ -150,8 +148,7 @@ class ResultElement:
@property
def detailed_path_json(self):
""" a function that builds path object for normal and blocking cases
"""
"""a function that builds path object for normal and blocking cases"""
index = 0
pro_list = []
for element in self.computed_path:
@@ -207,11 +204,9 @@ class ResultElement:
@property
def path_properties(self):
""" a function that returns the path properties (metrics, crossed elements) into a dict
"""
"""a function that returns the path properties (metrics, crossed elements) into a dict"""
def path_metric(pth, req):
""" creates the metrics dictionary
"""
"""creates the metrics dictionary"""
return [
{
'metric-type': 'SNR-bandwidth',
@@ -253,8 +248,7 @@ class ResultElement:
@property
def pathresult(self):
""" create the result dictionnary (response for a request)
"""
"""create the result dictionnary (response for a request)"""
try:
if self.path_request.blocking_reason in BLOCKING_NOPATH:
response = {
@@ -350,8 +344,7 @@ def ref_carrier(equipment):
def propagate(path, req, equipment):
""" propagates signals in each element according to initial spectrum set by user
"""
"""propagates signals in each element according to initial spectrum set by user"""
if req.initial_spectrum is not None:
si = carriers_to_spectral_information(initial_spectrum=req.initial_spectrum,
power=req.power, ref_carrier=ref_carrier(equipment))
@@ -442,8 +435,7 @@ def propagate_and_optimize_mode(path, req, equipment):
def jsontopath_metric(path_metric):
""" a functions that reads resulting metric from json string
"""
"""a functions that reads resulting metric from json string"""
output_snr = next(e['accumulative-value']
for e in path_metric if e['metric-type'] == 'SNR-0.1nm')
output_snrbandwidth = next(e['accumulative-value']
@@ -461,9 +453,7 @@ def jsontopath_metric(path_metric):
def jsontoparams(my_p, tsp, mode, equipment):
""" a function that derives optical params from transponder type and mode
supports the no mode case
"""
"""a function that derives optical params from transponder type and mode supports the no mode case"""
temp = []
for elem in my_p['path-properties']['path-route-objects']:
if 'num-unnum-hop' in elem['path-route-object']:
@@ -498,10 +488,10 @@ def jsontoparams(my_p, tsp, mode, equipment):
def jsontocsv(json_data, equipment, fileout):
""" reads json path result file in accordance with:
Yang model for requesting Path Computation
draft-ietf-teas-yang-path-computation-01.txt.
and write results in an CSV file
"""reads json path result file in accordance with:
Yang model for requesting Path Computation
draft-ietf-teas-yang-path-computation-01.txt.
and write results in an CSV file
"""
mywriter = writer(fileout)
mywriter.writerow(('response-id', 'source', 'destination', 'path_bandwidth', 'Pass?',
@@ -887,8 +877,7 @@ def compute_path_dsjctn(network, equipment, pathreqlist, disjunctions_list):
def isdisjoint(pth1, pth2):
""" returns 0 if disjoint
"""
"""returns 0 if disjoint"""
edge1 = list(pairwise(pth1))
edge2 = list(pairwise(pth2))
for edge in edge1:
@@ -898,9 +887,9 @@ def isdisjoint(pth1, pth2):
def find_reversed_path(pth):
""" select of intermediate roadms and find the path between them
note that this function may not give an exact result in case of multiple
links between two adjacent nodes.
"""select of intermediate roadms and find the path between them
note that this function may not give an exact result in case of multiple
links between two adjacent nodes.
"""
# TODO add some indication on elements to indicate from which other they
# are the reversed direction. This is partly done with oms indication
@@ -933,9 +922,7 @@ def find_reversed_path(pth):
def ispart(ptha, pthb):
""" the functions takes two paths a and b and retrns True
if all a elements are part of b and in the same order
"""
"""the functions takes two paths a and b and retrns True if all a elements are part of b and in the same order"""
j = 0
for elem in ptha:
if elem in pthb:
@@ -949,8 +936,7 @@ def ispart(ptha, pthb):
def remove_candidate(candidates, allpaths, rqst, pth):
""" filter duplicate candidates
"""
"""filter duplicate candidates"""
# print(f'coucou {rqst.request_id}')
for key, candidate in candidates.items():
temp = candidate.copy()
@@ -965,8 +951,7 @@ def remove_candidate(candidates, allpaths, rqst, pth):
def compare_reqs(req1, req2, disjlist):
""" compare two requests: returns True or False
"""
"""compare two requests: returns True or False"""
dis1 = [d for d in disjlist if req1.request_id in d.disjunctions_req]
dis2 = [d for d in disjlist if req2.request_id in d.disjunctions_req]
same_disj = False
@@ -1008,8 +993,8 @@ def compare_reqs(req1, req2, disjlist):
def requests_aggregation(pathreqlist, disjlist):
""" this function aggregates requests so that if several requests
exist between same source and destination and with same transponder type
"""this function aggregates requests so that if several requests
exist between same source and destination and with same transponder type
"""
# todo maybe add conditions on mode ??, spacing ...
# currently if undefined takes the default values
@@ -1037,9 +1022,10 @@ def requests_aggregation(pathreqlist, disjlist):
def correct_json_route_list(network, pathreqlist):
""" all names in list should be exact name in the network, and there is no ambiguity
This function only checks that list is correct, warns user if the name is incorrect and
suppresses the constraint it it is loose or raises an error if it is strict
"""all names in list should be exact name in the network, and there is no ambiguity
This function only checks that list is correct, warns user if the name is incorrect and
suppresses the constraint it it is loose or raises an error if it is strict
"""
all_uid = [n.uid for n in network.nodes()]
transponders = [n.uid for n in network.nodes() if isinstance(n, Transceiver)]
@@ -1088,8 +1074,7 @@ def correct_json_route_list(network, pathreqlist):
def deduplicate_disjunctions(disjn):
""" clean disjunctions to remove possible repetition
"""
"""clean disjunctions to remove possible repetition"""
local_disjn = disjn.copy()
for elem in local_disjn:
for dis_elem in local_disjn:
@@ -1100,8 +1085,9 @@ def deduplicate_disjunctions(disjn):
def compute_path_with_disjunction(network, equipment, pathreqlist, pathlist):
""" use a list but a dictionnary might be helpful to find path based on request_id
TODO change all these req, dsjct, res lists into dict !
"""use a list but a dictionnary might be helpful to find path based on request_id
TODO change all these req, dsjct, res lists into dict !
"""
path_res_list = []
reversed_path_res_list = []
@@ -1217,7 +1203,8 @@ def compute_path_with_disjunction(network, equipment, pathreqlist, pathlist):
def compute_spectrum_slot_vs_bandwidth(bandwidth, spacing, bit_rate, slot_width=0.0125e12):
""" Compute the number of required wavelengths and the M value (number of consumed slots)
"""Compute the number of required wavelengths and the M value (number of consumed slots)
Each wavelength consumes one `spacing`, and the result is rounded up to consume a natural number of slots.
>>> compute_spectrum_slot_vs_bandwidth(400e9, 50e9, 200e9)

94
gnpy/topology/spectrum_assignment.py
View File

@@ -23,8 +23,7 @@ LOGGER = getLogger(__name__)
class Bitmap:
""" records the spectrum occupation
"""
"""records the spectrum occupation"""
def __init__(self, f_min, f_max, grid, guardband=0.15e12, bitmap=None):
# n is the min index including guardband. Guardband is require to be sure
@@ -45,26 +44,22 @@ class Bitmap:
raise SpectrumError(f'bitmap is not consistant with f_min{f_min} - n: {n_min} and f_max{f_max}- n :{n_max}')
def getn(self, i):
""" converts the n (itu grid) into a local index
"""
"""converts the n (itu grid) into a local index"""
return self.freq_index[i]
def geti(self, nvalue):
""" converts the local index into n (itu grid)
"""
"""converts the local index into n (itu grid)"""
return self.freq_index.index(nvalue)
def insert_left(self, newbitmap):
""" insert bitmap on the left to align oms bitmaps if their start frequencies are different
"""
"""insert bitmap on the left to align oms bitmaps if their start frequencies are different"""
self.bitmap = newbitmap + self.bitmap
temp = list(range(self.n_min - len(newbitmap), self.n_min))
self.freq_index = temp + self.freq_index
self.n_min = self.freq_index[0]
def insert_right(self, newbitmap):
""" insert bitmap on the right to align oms bitmaps if their stop frequencies are different
"""
"""insert bitmap on the right to align oms bitmaps if their stop frequencies are different"""
self.bitmap = self.bitmap + newbitmap
self.freq_index = self.freq_index + list(range(self.n_max, self.n_max + len(newbitmap)))
self.n_max = self.freq_index[-1]
@@ -75,8 +70,8 @@ OMSParams = namedtuple('OMSParams', 'oms_id el_id_list el_list')
class OMS:
""" OMS class is the logical container that represent a link between two adjacent ROADMs and
records the crossed elements and the occupied spectrum
"""OMS class is the logical container that represent a link between two adjacent ROADMs and
records the crossed elements and the occupied spectrum
"""
def __init__(self, *args, **params):
@@ -98,15 +93,13 @@ class OMS:
f'{self.el_id_list[0]} - {self.el_id_list[-1]}', '\n'])
def add_element(self, elem):
""" records oms elements
"""
"""records oms elements"""
self.el_id_list.append(elem.uid)
self.el_list.append(elem)
def update_spectrum(self, f_min, f_max, guardband=0.15e12, existing_spectrum=None,
grid=0.00625e12):
""" frequencies expressed in Hz
"""
"""frequencies expressed in Hz"""
if existing_spectrum is None:
# add some 150 GHz margin to enable a center channel on f_min
# use ITU-T G694.1
@@ -126,8 +119,7 @@ class OMS:
# print(len(self.spectrum_bitmap.bitmap))
def assign_spectrum(self, nvalue, mvalue):
""" change oms spectrum to mark spectrum assigned
"""
"""change oms spectrum to mark spectrum assigned"""
if not isinstance(nvalue, int):
raise SpectrumError(f'N must be a signed integer, got {nvalue}')
if not isinstance(mvalue, int):
@@ -146,16 +138,16 @@ class OMS:
self.spectrum_bitmap.bitmap[self.spectrum_bitmap.geti(startn):self.spectrum_bitmap.geti(stopn) + 1] = [0] * (stopn - startn + 1)
def add_service(self, service_id, nb_wl):
""" record service and mark spectrum as occupied
"""
"""record service and mark spectrum as occupied"""
self.service_list.append(service_id)
self.nb_channels += nb_wl
def frequency_to_n(freq, grid=0.00625e12):
""" converts frequency into the n value (ITU grid)
reference to Recommendation G.694.1 (02/12), Figure I.3
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
"""converts frequency into the n value (ITU grid)
reference to Recommendation G.694.1 (02/12), Figure I.3
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
>>> frequency_to_n(193.1375e12)
6
@@ -167,9 +159,10 @@ def frequency_to_n(freq, grid=0.00625e12):
def nvalue_to_frequency(nvalue, grid=0.00625e12):
""" converts n value into a frequency
reference to Recommendation G.694.1 (02/12), Table 1
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
"""converts n value into a frequency
reference to Recommendation G.694.1 (02/12), Table 1
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
>>> nvalue_to_frequency(6)
193137500000000.0
@@ -181,17 +174,17 @@ def nvalue_to_frequency(nvalue, grid=0.00625e12):
def mvalue_to_slots(nvalue, mvalue):
""" convert center n an m into start and stop n
"""
"""convert center n an m into start and stop n"""
startn = nvalue - mvalue
stopn = nvalue + mvalue - 1
return startn, stopn
def slots_to_m(startn, stopn):
""" converts the start and stop n values to the center n and m value
reference to Recommendation G.694.1 (02/12), Figure I.3
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
"""converts the start and stop n values to the center n and m value
reference to Recommendation G.694.1 (02/12), Figure I.3
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
>>> nval, mval = slots_to_m(6, 20)
>>> nval
@@ -206,10 +199,11 @@ def slots_to_m(startn, stopn):
def m_to_freq(nvalue, mvalue, grid=0.00625e12):
""" converts m into frequency range
spectrum(13,7) is (193137500000000.0, 193225000000000.0)
reference to Recommendation G.694.1 (02/12), Figure I.3
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
"""converts m into frequency range
spectrum(13,7) is (193137500000000.0, 193225000000000.0)
reference to Recommendation G.694.1 (02/12), Figure I.3
https://www.itu.int/rec/T-REC-G.694.1-201202-I/en
>>> fstart, fstop = m_to_freq(13, 7)
>>> fstart
@@ -225,9 +219,7 @@ def m_to_freq(nvalue, mvalue, grid=0.00625e12):
def align_grids(oms_list):
""" used to apply same grid to all oms : same starting n, stop n and slot size
out of grid slots are set to 0
"""
"""Used to apply same grid to all oms : same starting n, stop n and slot size. Out of grid slots are set to 0."""
n_min = min([o.spectrum_bitmap.n_min for o in oms_list])
n_max = max([o.spectrum_bitmap.n_max for o in oms_list])
for this_o in oms_list:
@@ -239,12 +231,13 @@ def align_grids(oms_list):
def build_oms_list(network, equipment):
""" initialization of OMS list in the network
an oms is build reading all intermediate nodes between two adjacent ROADMs
each element within the list is being added an oms and oms_id to record the
oms it belongs to.
the function supports different spectrum width and supposes that the whole network
works with the min range among OMSs
"""initialization of OMS list in the network
an oms is build reading all intermediate nodes between two adjacent ROADMs
each element within the list is being added an oms and oms_id to record the
oms it belongs to.
the function supports different spectrum width and supposes that the whole network
works with the min range among OMSs
"""
oms_id = 0
oms_list = []
@@ -296,8 +289,9 @@ def build_oms_list(network, equipment):
def reversed_oms(oms_list):
""" identifies reversed OMS
only applicable for non parallel OMS
"""identifies reversed OMS
only applicable for non parallel OMS
"""
for oms in oms_list:
has_reversed = False
@@ -369,8 +363,7 @@ def spectrum_selection(pth, oms_list, requested_m, requested_n=None):
def select_candidate(candidates, policy):
""" selects a candidate among all available spectrum
"""
"""selects a candidate among all available spectrum"""
if policy == 'first_fit':
if candidates:
return candidates[0]
@@ -381,8 +374,9 @@ def select_candidate(candidates, policy):
def pth_assign_spectrum(pths, rqs, oms_list, rpths):
""" basic first fit assignment
if reversed path are provided, means that occupation is bidir
"""basic first fit assignment
if reversed path are provided, means that occupation is bidir
"""
for pth, rq, rpth in zip(pths, rqs, rpths):
# computes the number of channels required

2
tests/test_amplifier.py
View File

@@ -115,7 +115,7 @@ def test_si(si, nch_and_spacing):
@pytest.mark.parametrize("gain", [17, 19, 21, 23])
def test_compare_nf_models(gain, setup_edfa_variable_gain, si):
""" compare the 2 amplifier models (polynomial and estimated from nf_min and max)
"""compare the 2 amplifier models (polynomial and estimated from nf_min and max)
=> nf_model vs nf_poly_fit for intermediate gain values:
between gain_min and gain_flatmax some discrepancy is expected but target < 0.5dB
=> unitary test for Edfa._calc_nf (and Edfa.interpol_params)"""

52
tests/test_disjunction.py
View File

@@ -4,12 +4,12 @@
# License: BSD 3-Clause Licence
# Copyright (c) 2018, Telecom Infra Project
'''
"""
@author: esther.lerouzic
checks that computed paths are disjoint as specified in the json service file
that computed paths do not loop
that include node constraints are correctly taken into account
'''
"""
from pathlib import Path
import pytest
@@ -31,8 +31,7 @@ EQPT_LIBRARY_NAME = Path(__file__).parent.parent / 'tests/data/eqpt_config.json'
@pytest.fixture()
def serv(test_setup):
''' common setup for service list
'''
"""common setup for service list"""
network, equipment = test_setup
data = load_requests(SERVICE_FILE_NAME, equipment, bidir=False, network=network, network_filename=NETWORK_FILE_NAME)
rqs = requests_from_json(data, equipment)
@@ -43,8 +42,7 @@ def serv(test_setup):
@pytest.fixture()
def test_setup():
''' common setup for tests: builds network, equipment and oms only once
'''
"""common setup for tests: builds network, equipment and oms only once"""
equipment = load_equipment(EQPT_LIBRARY_NAME)
network = load_network(NETWORK_FILE_NAME, equipment)
# Build the network once using the default power defined in SI in eqpt config
@@ -61,9 +59,10 @@ def test_setup():
def test_disjunction(serv):
''' service_file contains sevaral combination of disjunction constraint. The test checks
that computed paths with disjunction constraint are effectively disjoint
'''
"""service_file contains sevaral combination of disjunction constraint
The test checks that computed paths with disjunction constraint are effectively disjoint.
"""
network, equipment, rqs, dsjn = serv
pths = compute_path_dsjctn(network, equipment, rqs, dsjn)
print(dsjn)
@@ -86,8 +85,7 @@ def test_disjunction(serv):
def test_does_not_loop_back(serv):
''' check that computed paths do not loop back ie each element appears only once
'''
"""check that computed paths do not loop back ie each element appears only once"""
network, equipment, rqs, dsjn = serv
pths = compute_path_dsjctn(network, equipment, rqs, dsjn)
test = True
@@ -108,8 +106,7 @@ def test_does_not_loop_back(serv):
def create_rq(equipment, srce, dest, bdir, node_list, loose_list, rqid='test_request'):
''' create the usual request list according to parameters
'''
"""create the usual request list according to parameters"""
requests_list = []
params = {
'request_id': rqid,
@@ -151,19 +148,20 @@ def create_rq(equipment, srce, dest, bdir, node_list, loose_list, rqid='test_req
['trx a', 'trx h', 'pass', 'found_path', ['trx h'], ['STRICT']],
['trx a', 'trx h', 'pass', 'found_path', ['roadm a'], ['STRICT']]])
def test_include_constraints(test_setup, srce, dest, result, pth, node_list, loose_list):
''' check that all combinations of constraints are correctly handled:
- STRICT/LOOSE
- correct names/incorrect names -> pass/fail
- possible include/impossible include
if incorrect name -> fail
else:
constraint |one or more STRICT | all LOOSE
----------------------------------------------------------------------------------
>1 path from s to d | can be applied | found_path | found_path
| cannot be applied | no_path | found_path
----------------------------------------------------------------------------------
0 | | computation stops
'''
"""check that all combinations of constraints are correctly handled:
- STRICT/LOOSE
- correct names/incorrect names -> pass/fail
- possible include/impossible include
if incorrect name -> fail
else:
constraint |one or more STRICT | all LOOSE
----------------------------------------------------------------------------------
>1 path from s to d | can be applied | found_path | found_path
| cannot be applied | no_path | found_path
----------------------------------------------------------------------------------
0 | | computation stops
"""
network, equipment = test_setup
dsjn = []
bdir = False
@@ -201,7 +199,7 @@ def test_include_constraints(test_setup, srce, dest, result, pth, node_list, loo
['roadm c', 'roadm f'],
['roadm a', 'roadm b', 'roadm f', 'roadm h']]]])
def test_create_disjunction(test_setup, dis1, dis2, node_list1, loose_list1, result, expected_paths):
""" verifies that the expected result is obtained for a set of particular constraints:
"""verifies that the expected result is obtained for a set of particular constraints:
in particular, verifies that:
- multiple disjunction constraints are correcly handled
- in case a loose constraint can not be met, the first alternate candidate is selected

19
tests/test_equalization.py
View File

@@ -299,8 +299,7 @@ def test_2low_input_power(target_out, delta_pdb_per_channel, correction):
def net_setup(equipment):
""" common setup for tests: builds network, equipment and oms only once
"""
"""common setup for tests: builds network, equipment and oms only once"""
network = load_network(NETWORK_FILENAME, equipment)
spectrum = equipment['SI']['default']
p_db = spectrum.power_dbm
@@ -310,8 +309,7 @@ def net_setup(equipment):
def create_voyager_req(equipment, source, dest, bidir, nodes_list, loose_list, mode, spacing, power_dbm):
""" create the usual request list according to parameters
"""
"""create the usual request list according to parameters"""
params = {'request_id': 'test_request',
'source': source,
'bidir': bidir,
@@ -336,8 +334,7 @@ def create_voyager_req(equipment, source, dest, bidir, nodes_list, loose_list, m
@pytest.mark.parametrize('power_dbm', [0, 1, -2, None])
@pytest.mark.parametrize('mode, slot_width', (['mode 1', 50e9], ['mode 2', 75e9]))
def test_initial_spectrum(mode, slot_width, power_dbm):
""" checks that propagation using the user defined spectrum identical to SI, gives same result as SI
"""
"""checks that propagation using the user defined spectrum identical to SI, gives same result as SI"""
# first propagate without any req.initial_spectrum attribute
equipment = load_equipment(EQPT_FILENAME)
req = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
@@ -373,7 +370,7 @@ def test_initial_spectrum(mode, slot_width, power_dbm):
def test_initial_spectrum_not_identical():
""" checks that user defined spectrum overrides spectrum defined in SI
"""checks that user defined spectrum overrides spectrum defined in SI
"""
# first propagate without any req.initial_spectrum attribute
equipment = load_equipment(EQPT_FILENAME)
@@ -408,7 +405,7 @@ def test_initial_spectrum_not_identical():
('target_psd_out_mWperGHz', power_dbm_to_psd_mw_ghz(-20, 32e9))])
@pytest.mark.parametrize('power_dbm', [0, 2, -0.5])
def test_target_psd_or_psw(power_dbm, equalization, target_value):
""" checks that if target_out_mWperSlotWidth or target_psd_out_mWperGHz is defined, it is used as equalization
"""checks that if target_out_mWperSlotWidth or target_psd_out_mWperGHz is defined, it is used as equalization
and it gives same result if computed target is the same
"""
equipment = load_equipment(EQPT_FILENAME)
@@ -438,8 +435,7 @@ def test_target_psd_or_psw(power_dbm, equalization, target_value):
def ref_network():
""" Create a network instance with a instance of propagated path
"""
"""Create a network instance with a instance of propagated path"""
equipment = load_equipment(EQPT_FILENAME)
network = net_setup(equipment)
req0 = create_voyager_req(equipment, 'trx Brest_KLA', 'trx Vannes_KBE', False, ['trx Vannes_KBE'], ['STRICT'],
@@ -451,7 +447,8 @@ def ref_network():
@pytest.mark.parametrize('deltap', [0, +1.2, -0.5])
def test_target_psd_out_mwperghz_deltap(deltap):
""" checks that if target_psd_out_mWperGHz is defined, delta_p of amps is correctly updated
"""checks that if target_psd_out_mWperGHz is defined, delta_p of amps is correctly updated
Power over 1.2dBm saturate amp with this test: TODO add a test on this saturation
"""
equipment = load_equipment(EQPT_FILENAME)

4
tests/test_invocation.py
View File

@@ -30,7 +30,7 @@ SRC_ROOT = Path(__file__).parent.parent
def test_example_invocation(capfd, output, handler, args):
'''Make sure that our examples produce useful output'''
"""Make sure that our examples produce useful output"""
os.chdir(SRC_ROOT)
expected = open(SRC_ROOT / 'tests' / 'invocation' / output, mode='r', encoding='utf-8').read()
handler(args)
@@ -41,7 +41,7 @@ def test_example_invocation(capfd, output, handler, args):
@pytest.mark.parametrize('program', ('gnpy-transmission-example', 'gnpy-path-request'))
def test_run_wrapper(program):
'''Ensure that our wrappers really, really work'''
"""Ensure that our wrappers really, really work"""
proc = subprocess.run((program, '--help'), stdout=subprocess.PIPE, stderr=subprocess.PIPE,
check=True, universal_newlines=True)
assert proc.stderr == ''

58
tests/test_parser.py
View File

@@ -3,16 +3,17 @@
# @Author: Esther Le Rouzic
# @Date: 2018-06-15
""" Adding tests to check the parser non regression
convention of naming of test files:
- ..._expected.json for the reference output
tests:
- generation of topology json
- reading of Eqpt sheet w and W/ power mode
- consistency of autodesign
- generation of service list based on service sheet
- writing of results in csv
- writing of results in json (same keys)
"""Adding tests to check the parser non regression
convention of naming of test files:
- ..._expected.json for the reference output
tests:
- generation of topology json
- reading of Eqpt sheet w and W/ power mode
- consistency of autodesign
- generation of service list based on service sheet
- writing of results in csv
- writing of results in json (same keys)
"""
from pathlib import Path
@@ -46,8 +47,7 @@ equipment = load_equipment(eqpt_filename)
}.items())
def test_excel_json_generation(tmpdir, xls_input, expected_json_output):
""" tests generation of topology json
"""
"""tests generation of topology json"""
xls_copy = Path(tmpdir) / xls_input.name
shutil.copyfile(xls_input, xls_copy)
convert_file(xls_copy)
@@ -68,9 +68,7 @@ def test_excel_json_generation(tmpdir, xls_input, expected_json_output):
DATA_DIR / 'testTopology_auto_design_expected.json',
}.items())
def test_auto_design_generation_fromxlsgainmode(tmpdir, xls_input, expected_json_output):
""" tests generation of topology json
test that the build network gives correct results in gain mode
"""
"""tests generation of topology json and that the build network gives correct results in gain mode"""
equipment = load_equipment(eqpt_filename)
network = load_network(xls_input, equipment)
# in order to test the Eqpt sheet and load gain target,
@@ -100,8 +98,7 @@ def test_auto_design_generation_fromxlsgainmode(tmpdir, xls_input, expected_json
True
}.items())
def test_auto_design_generation_fromjson(tmpdir, json_input, power_mode):
"""test that autodesign creates same file as an input file already autodesigned
"""
"""test that autodesign creates same file as an input file already autodesigned"""
equipment = load_equipment(eqpt_filename)
network = load_network(json_input, equipment)
# in order to test the Eqpt sheet and load gain target,
@@ -127,8 +124,7 @@ def test_auto_design_generation_fromjson(tmpdir, json_input, power_mode):
DATA_DIR / 'testService.xls': DATA_DIR / 'testService_services_expected.json'
}.items())
def test_excel_service_json_generation(xls_input, expected_json_output):
""" test services creation
"""
"""test services creation"""
equipment = load_equipment(eqpt_filename)
network = load_network(DATA_DIR / 'testTopology.xls', equipment)
# Build the network once using the default power defined in SI in eqpt config
@@ -148,9 +144,7 @@ def test_excel_service_json_generation(xls_input, expected_json_output):
(DATA_DIR / 'testTopology_response.json', )
)
def test_csv_response_generation(tmpdir, json_input):
""" tests if generated csv is consistant with expected generation
same columns (order not important)
"""
"""tests if generated csv is consistant with expected generation same columns (order not important)"""
json_data = load_json(json_input)
equipment = load_equipment(eqpt_filename)
csv_filename = Path(tmpdir / json_input.name).with_suffix('.csv')
@@ -215,8 +209,7 @@ def test_csv_response_generation(tmpdir, json_input):
DATA_DIR / 'testTopology.xls': DATA_DIR / 'testTopology_response.json',
}.items())
def test_json_response_generation(xls_input, expected_response_file):
""" tests if json response is correctly generated for all combinations of requests
"""
"""tests if json response is correctly generated for all combinations of requests"""
equipment = load_equipment(eqpt_filename)
network = load_network(xls_input, equipment)
@@ -323,8 +316,7 @@ def test_json_response_generation(xls_input, expected_response_file):
('trx Brest_KLA', 'trx Rennes_STA', 'Brest_KLA | trx Lannion_CAS', 'STRICT', 'Fail')
])
def test_excel_ila_constraints(source, destination, route_list, hoptype, expected_correction):
""" add different kind of constraints to test all correct_route cases
"""
"""add different kind of constraints to test all correct_route cases"""
service_xls_input = DATA_DIR / 'testTopology.xls'
network_json_input = DATA_DIR / 'testTopology_auto_design_expected.json'
equipment = load_equipment(eqpt_filename)
@@ -376,8 +368,7 @@ def test_excel_ila_constraints(source, destination, route_list, hoptype, expecte
def setup_per_degree(case):
""" common setup for degree: returns the dict network for different cases
"""
"""common setup for degree: returns the dict network for different cases"""
json_network = load_json(DATA_DIR / 'testTopology_expected.json')
json_network_auto = load_json(DATA_DIR / 'testTopology_auto_design_expected.json')
if case == 'no':
@@ -401,8 +392,7 @@ def setup_per_degree(case):
@pytest.mark.parametrize('case', ['no', 'all', 'Lannion_CAS and all', 'Lannion_CAS and one'])
def test_target_pch_out_db_global(case):
""" check that per degree attributes are correctly created with global values if none are given
"""
"""check that per degree attributes are correctly created with global values if none are given"""
json_network = setup_per_degree(case)
per_degree = {}
for elem in json_network['elements']:
@@ -442,14 +432,12 @@ def test_target_pch_out_db_global(case):
def all_rows(sh, start=0):
""" reads excel sheet row per row
"""
"""reads excel sheet row per row"""
return (sh.row(x) for x in range(start, sh.nrows))
class Amp:
""" Node element contains uid, list of connected nodes and eqpt type
"""
"""Node element contains uid, list of connected nodes and eqpt type"""
def __init__(self, uid, to_node, eqpt=None, west=None):
self.uid = uid
@@ -459,7 +447,7 @@ class Amp:
def test_eqpt_creation(tmpdir):
""" tests that convert correctly creates equipment according to equipment sheet
"""tests that convert correctly creates equipment according to equipment sheet
including all cominations in testTopologyconvert.xls: if a line exists the amplifier
should be created even if no values are provided.
"""

16
tests/test_roadm_restrictions.py
View File

@@ -31,10 +31,10 @@ NETWORK_FILE_NAME = TEST_DIR / 'data/testTopology_expected.json'
# mark node_uid amps as fused for testing purpose
@pytest.mark.parametrize("node_uid", ['east edfa in Lannion_CAS to Stbrieuc'])
def test_no_amp_feature(node_uid):
''' Check that booster is not placed on a roadm if fused is specified
test_parser covers partly this behaviour. This test should guaranty that the
feature is preserved even if convert is changed
'''
"""Check that booster is not placed on a roadm if fused is specified
test_parser covers partly this behaviour. This test should guaranty that the
feature is preserved even if convert is changed
"""
equipment = load_equipment(EQPT_LIBRARY_NAME)
json_network = load_json(NETWORK_FILE_NAME)
@@ -145,9 +145,9 @@ def equipment():
'booster_variety_list':[]
}])
def test_restrictions(restrictions, equipment):
''' test that restriction is correctly applied if provided in eqpt_config and if no Edfa type
"""test that restriction is correctly applied if provided in eqpt_config and if no Edfa type
were provided in the network json
'''
"""
# add restrictions
equipment['Roadm']['default'].restrictions = restrictions
# build network
@@ -212,11 +212,11 @@ def test_restrictions(restrictions, equipment):
@pytest.mark.parametrize('power_dbm', [0, +1, -2])
@pytest.mark.parametrize('prev_node_type, effective_pch_out_db', [('edfa', -20.0), ('fused', -22.0)])
def test_roadm_target_power(prev_node_type, effective_pch_out_db, power_dbm):
''' Check that egress power of roadm is equal to target power if input power is greater
"""Check that egress power of roadm is equal to target power if input power is greater
than target power else, that it is equal to input power. Use a simple two hops A-B-C topology
for the test where the prev_node in ROADM B is either an amplifier or a fused, so that the target
power can not be met in this last case.
'''
"""
equipment = load_equipment(EQPT_LIBRARY_NAME)
json_network = load_json(TEST_DIR / 'data/twohops_roadm_power_test.json')
prev_node = next(n for n in json_network['elements'] if n['uid'] == 'west edfa in node B to ila2')

8
tests/test_science_utils.py
View File

@@ -23,7 +23,7 @@ TEST_DIR = Path(__file__).parent
def test_fiber():
""" Test the accuracy of propagating the Fiber."""
"""Test the accuracy of propagating the Fiber."""
fiber = Fiber(**load_json(TEST_DIR / 'data' / 'test_science_utils_fiber_config.json'))
# fix grid spectral information generation
@@ -65,7 +65,7 @@ def test_fiber():
@pytest.mark.usefixtures('set_sim_params')
def test_raman_fiber():
""" Test the accuracy of propagating the RamanFiber."""
"""Test the accuracy of propagating the RamanFiber."""
# spectral information generation
spectral_info_input = create_input_spectral_information(f_min=191.3e12, f_max=196.1e12, roll_off=0.15,
baud_rate=32e9, power=1e-3, spacing=50e9, tx_osnr=40.0,
@@ -92,7 +92,7 @@ def test_raman_fiber():
(0.5, 81, "Lumped loss positions must be between 0 and the fiber length (80.0 km), boundaries excluded.")))
@pytest.mark.usefixtures('set_sim_params')
def test_fiber_lumped_losses(loss, position, errmsg, set_sim_params):
""" Lumped losses length sanity checking."""
"""Lumped losses length sanity checking."""
SimParams.set_params(load_json(TEST_DIR / 'data' / 'sim_params.json'))
fiber_dict = load_json(TEST_DIR / 'data' / 'test_lumped_losses_raman_fiber_config.json')
fiber_dict['params']['lumped_losses'] = [{'position': position, 'loss': loss}]
@@ -103,7 +103,7 @@ def test_fiber_lumped_losses(loss, position, errmsg, set_sim_params):
@pytest.mark.usefixtures('set_sim_params')
def test_fiber_lumped_losses_srs(set_sim_params):
""" Test the accuracy of Fiber with lumped losses propagation."""
"""Test the accuracy of Fiber with lumped losses propagation."""
# spectral information generation
spectral_info_input = create_input_spectral_information(f_min=191.3e12, f_max=196.1e12, roll_off=0.15,
baud_rate=32e9, power=1e-3, spacing=50e9, tx_osnr=40.0,

40
tests/test_spectrum_assignment.py
View File

@@ -45,8 +45,7 @@ def equipment():
@pytest.fixture()
def setup(equipment):
""" common setup for tests: builds network, equipment and oms only once
"""
"""common setup for tests: builds network, equipment and oms only once"""
network = load_network(NETWORK_FILENAME, equipment)
spectrum = equipment['SI']['default']
p_db = spectrum.power_dbm
@@ -57,9 +56,9 @@ def setup(equipment):
def test_oms(setup):
""" tests that the OMS is between two ROADMs, that there is no ROADM or transceivers in the OMS
except end points, checks that the id of OMS is present in the element and that the element
OMS id is consistant
"""tests that the OMS is between two ROADMs, that there is no ROADM or transceivers in the OMS
except end points, checks that the id of OMS is present in the element and that the element
OMS id is consistant
"""
network, oms_list = setup
for oms in oms_list:
@@ -150,8 +149,7 @@ def test_aligned(nmin, nmax, setup):
@pytest.mark.parametrize('nval1', [0, 15, 24])
@pytest.mark.parametrize('nval2', [8, 12])
def test_assign_and_sum(nval1, nval2, setup):
""" checks that bitmap sum gives correct result
"""
"""checks that bitmap sum gives correct result"""
network, oms_list = setup
guardband = grid
mval = 4 # slot in 12.5GHz
@@ -198,8 +196,7 @@ def test_assign_and_sum(nval1, nval2, setup):
def test_bitmap_assignment(setup):
""" test that a bitmap can be assigned
"""
"""test that a bitmap can be assigned"""
network, oms_list = setup
random_oms = oms_list[2]
random_oms.assign_spectrum(13, 7)
@@ -216,8 +213,7 @@ def test_bitmap_assignment(setup):
@pytest.fixture()
def services(equipment):
""" common setup for service list: builds service only once
"""
"""common setup for service list: builds service only once"""
with open(SERVICE_FILENAME, encoding='utf-8') as my_f:
services = json.loads(my_f.read())
return services
@@ -225,15 +221,13 @@ def services(equipment):
@pytest.fixture()
def requests(equipment, services):
""" common setup for requests, builds requests list only once
"""
"""common setup for requests, builds requests list only once"""
requests = requests_from_json(services, equipment)
return requests
def test_spectrum_assignment_on_path(equipment, setup, requests):
""" test assignment functions on path and network
"""
"""test assignment functions on path and network"""
network, oms_list = setup
req = [deepcopy(requests[1])]
paths = compute_path_dsjctn(network, equipment, req, [])
@@ -270,8 +264,7 @@ def test_spectrum_assignment_on_path(equipment, setup, requests):
@pytest.fixture()
def request_set():
""" creates default request dict
"""
"""creates default request dict"""
return {
'request_id': '0',
'source': 'trx a',
@@ -299,8 +292,7 @@ def request_set():
def test_freq_slot_exist(setup, equipment, request_set):
""" test that assignment works even if effective_freq_slot is not populated
"""
"""test that assignment works even if effective_freq_slot is not populated"""
network, oms_list = setup
params = request_set
params['effective_freq_slot'] = None
@@ -312,8 +304,7 @@ def test_freq_slot_exist(setup, equipment, request_set):
def test_inconsistant_freq_slot(setup, equipment, request_set):
""" test that an inconsistant M correctly raises an error
"""
"""test that an inconsistant M correctly raises an error"""
network, oms_list = setup
params = request_set
# minimum required nb of slots is 32 (800Gbit/100Gbit/s channels each occupying 50GHz ie 4 slots)
@@ -346,8 +337,7 @@ def test_inconsistant_freq_slot(setup, equipment, request_set):
(-60, 20, None, None, 'NOT_ENOUGH_RESERVED_SPECTRUM')
])
def test_n_m_requests(setup, equipment, n, m, final_n, final_m, blocking_reason, request_set):
""" test that various N and M values for a request end up with the correct path assgnment
"""
"""test that various N and M values for a request end up with the correct path assgnment"""
network, oms_list = setup
# add an occupation on one of the span of the expected path OMS list on both directions
# as defined by its offsets within the OMS list: [17, 20, 13, 22] and reversed path [19, 16, 21, 26]
@@ -372,9 +362,7 @@ def test_n_m_requests(setup, equipment, n, m, final_n, final_m, blocking_reason,
def test_reversed_direction(equipment, setup, requests, services):
""" checks that if spectrum is selected on one direction it is also selected on reversed
direction
"""
"""checks that if spectrum is selected on one direction it is also selected on reversed direction"""
network, oms_list = setup
dsjn = disjunctions_from_json(services)
dsjn = deduplicate_disjunctions(dsjn)