mirror of
https://github.com/Telecominfraproject/oopt-gnpy.git
synced 2025-11-01 02:28:05 +00:00
ad64595d75
- test oms building:
o there should not be roadm or transceivers between end points except endpoints
o the element oms is must correspond to the correct oms id
o the element uid in the oms must match the id list
- test the alignment function
this function enables to have different grid end points in different parts
of the network. The grid used in the network must cover all these grids, so
some bit stuffing is needed on the oms that have different sizes.
The test checks that
o min and max attribute are correctly updated
o min and max n or freq values are consistent and consistent with bitmap
o alignment is correct
- test that the assignment of n and m values is correct
o check that assign_spectrum has returned an error code if the requested
assignment is not possible and that the bitmap has not been set to 0
o check that the bitmap sum works correctly when assignment is feasible
and that all range of spectrum has been set to zero. eg:
[1 1 0 0 0 0 1 1 1 1 1 1] and [1 1 1 1 1 0 0 0 0 1 1] must be:
[1 1 0 0 0 0 0 0 0 0 1 1]
- Check that spectrum assignment of 13,7 is correct in Hz
This example has been extracted from ITU-T G694.1
expected value in Hz for 13,7 is 193137500000000.0,193225000000000.0 in Hz
see fig I.3 of this document https://www.itu.int/rec/T-REC-G.694.1-201202-I/fr
- test assignment limits
o verify that inconsistent values raise error: ie with defined fmin fmax
n and m have limited values.
combine valid and non valid data for n and m
o verify that Bitmap created with a 0/1 list is consistent with fmin,
fmax, grid and guard band
- Test with a path configuration
o loop on assignment on a given path. assignment should be OK until n = 96
at that value the assignment is no more feasible (exceed grid) and the selection
function should return None value for all center_n, startn and stopn
o select an arbitrary request and try to assign 1 slot more than the whole
spectrum or exactly the whole spectrum and check that function correctly
return None or not None values
- Test assignment with reversed path
o add data and requests fixtures to reduce test time
o test that if spectrum is assigned on one direction it is also
assigned on reversed direction (bitmaps must be identical)
Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
Co-authored-by: Jan Kundrát <jan.kundrat@telecominfraproject.com>
Change-Id: I96dd15452cc2e59086d4022ec4b026be845f4355
309 lines
13 KiB
Python
309 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# TelecomInfraProject/gnpy/examples
|
|
# Module name: test_spectrum_assignment.py
|
|
# Version:
|
|
# License: BSD 3-Clause Licence
|
|
# Copyright (c) 2018, Telecom Infra Project
|
|
|
|
"""
|
|
@author: esther.lerouzic
|
|
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from copy import deepcopy
|
|
import json
|
|
from math import ceil
|
|
import pytest
|
|
from gnpy.core.equipment import load_equipment, automatic_nch
|
|
from gnpy.core.network import load_network, build_network
|
|
from gnpy.core.utils import lin2db
|
|
from gnpy.core.elements import Roadm, Transceiver
|
|
from gnpy.core.spectrum_assignment import (build_oms_list, align_grids, nvalue_to_frequency,
|
|
bitmap_sum, Bitmap, spectrum_selection, pth_assign_spectrum)
|
|
from gnpy.core.exceptions import SpectrumError
|
|
from gnpy.core.request import compute_path_dsjctn, find_reversed_path
|
|
from examples.path_requests_run import requests_from_json, disjunctions_from_json, correct_disjn
|
|
|
|
TEST_DIR = Path(__file__).parent
|
|
DATA_DIR = TEST_DIR / 'data'
|
|
EQPT_FILENAME = DATA_DIR / 'eqpt_config.json'
|
|
NETWORK_FILENAME = DATA_DIR / 'testTopology_auto_design_expected.json'
|
|
SERVICE_FILENAME = DATA_DIR / 'testTopology_services_expected.json'
|
|
|
|
grid = 0.00625e12
|
|
slot = 0.0125e12
|
|
guardband = 0.15e12
|
|
cband_freq_min = 191.3e12
|
|
cband_freq_max = 196.1e12
|
|
|
|
@pytest.fixture()
|
|
def equipment():
|
|
equipment = load_equipment(EQPT_FILENAME)
|
|
return equipment
|
|
|
|
@pytest.fixture()
|
|
def setup(equipment):
|
|
""" 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
|
|
p_total_db = p_db + lin2db(automatic_nch(spectrum.f_min, spectrum.f_max, spectrum.spacing))
|
|
build_network(network, equipment, p_db, p_total_db)
|
|
oms_list = build_oms_list(network, equipment)
|
|
return network, oms_list
|
|
|
|
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
|
|
"""
|
|
network, oms_list = setup
|
|
for oms in oms_list:
|
|
assert isinstance(oms.el_list[0], Roadm) and isinstance(oms.el_list[-1], Roadm)
|
|
for i, elem in enumerate(oms.el_list[1:-1]):
|
|
assert not isinstance(elem, Roadm)
|
|
assert elem in network.nodes()
|
|
assert elem.oms.oms_id == oms.oms_id
|
|
assert elem.uid == oms.el_id_list[i+1]
|
|
|
|
@pytest.mark.parametrize('nval', [0, 10, -255])
|
|
@pytest.mark.parametrize('mval', [1, 8])
|
|
def test_acceptable_values(nval, mval, setup):
|
|
"""Reasonable parameters should not throw"""
|
|
network, oms_list = setup
|
|
random_oms = oms_list[5]
|
|
random_oms.assign_spectrum(nval, mval)
|
|
|
|
@pytest.mark.parametrize('nval,mval', (
|
|
(0, 600),
|
|
(-300, 1),
|
|
(500, 1),
|
|
(0, -2),
|
|
(0, 4.5),
|
|
(0.5, 8),
|
|
))
|
|
def test_wrong_values(nval, mval, setup):
|
|
"""Out of range or invalid values"""
|
|
network, oms_list = setup
|
|
random_oms = oms_list[5]
|
|
with pytest.raises(SpectrumError):
|
|
random_oms.assign_spectrum(nval, mval)
|
|
|
|
@pytest.mark.parametrize('nmin', [-288, -260, -300])
|
|
@pytest.mark.parametrize('nmax', [480, 320, 450])
|
|
def test_aligned(nmin, nmax, setup):
|
|
"""Checks that the OMS grid is correctly aligned
|
|
|
|
Note that bitmap index uses guardband on both ends so that if nmin, nmax = -200, +200,
|
|
min/max navalue in bitmap are -224, +223, which makes 223 -(-224) +1 frequencies.
|
|
"""
|
|
network, oms_list = setup
|
|
nguard = guardband / grid
|
|
center = 193.1e12
|
|
freq_min = center + nmin * grid
|
|
freq_max = center + nmax * grid
|
|
random_oms = oms_list[10]
|
|
|
|
# We're always starting with full C-band
|
|
assert pytest.approx(nvalue_to_frequency(random_oms.spectrum_bitmap.freq_index_min) * 1e-12, abs=1e-12) == 191.3
|
|
assert pytest.approx(nvalue_to_frequency(random_oms.spectrum_bitmap.freq_index_max) * 1e-12, abs=1e-12) == 196.1
|
|
ind_max = len(random_oms.spectrum_bitmap.bitmap) - 1
|
|
|
|
# "inner" frequencies, without the guard baand
|
|
inner_n_min = random_oms.spectrum_bitmap.getn(0) + nguard
|
|
inner_n_max = random_oms.spectrum_bitmap.getn(ind_max) - nguard + 1
|
|
assert inner_n_min == random_oms.spectrum_bitmap.freq_index_min
|
|
assert inner_n_max == random_oms.spectrum_bitmap.freq_index_max
|
|
assert inner_n_min == -288
|
|
assert inner_n_max == 480
|
|
|
|
# checks that changes are applied on bitmap and freq attributes of Bitmap object
|
|
random_oms.update_spectrum(freq_min, freq_max, grid=grid, guardband=guardband)
|
|
assert nmin == random_oms.spectrum_bitmap.freq_index_min
|
|
assert nmax == random_oms.spectrum_bitmap.freq_index_max
|
|
|
|
print('Adjusted spectrum: {:f} - {:f}'.format(
|
|
nvalue_to_frequency(random_oms.spectrum_bitmap.freq_index_min) * 1e-12,
|
|
nvalue_to_frequency(random_oms.spectrum_bitmap.freq_index_max) * 1e-12
|
|
))
|
|
|
|
ind_max = len(random_oms.spectrum_bitmap.bitmap) - 1
|
|
inner_n_min = random_oms.spectrum_bitmap.getn(0) + nguard
|
|
inner_n_max = random_oms.spectrum_bitmap.getn(ind_max) - nguard + 1
|
|
assert inner_n_min == random_oms.spectrum_bitmap.freq_index_min
|
|
assert inner_n_max == random_oms.spectrum_bitmap.freq_index_max
|
|
|
|
oms_list = align_grids(oms_list)
|
|
ind_max = len(random_oms.spectrum_bitmap.bitmap) - 1
|
|
nvalmin = random_oms.spectrum_bitmap.getn(0)
|
|
nvalmax = random_oms.spectrum_bitmap.getn(ind_max)
|
|
assert nvalmin <= nmin and nvalmax >= nmax
|
|
|
|
@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
|
|
"""
|
|
network, oms_list = setup
|
|
guardband = grid
|
|
mval = 4 # slot in 12.5GHz
|
|
freq_max = cband_freq_min + 24 * grid
|
|
# arbitrary test on oms #10 and #11
|
|
# first reduce the grid to 24 center frequencies to ease reading when test fails
|
|
oms1 = oms_list[10]
|
|
oms1.update_spectrum(cband_freq_min, freq_max, grid=grid, guardband=guardband)
|
|
oms2 = oms_list[11]
|
|
oms2.update_spectrum(cband_freq_min, freq_max, grid=grid, guardband=guardband)
|
|
print('initial spectrum')
|
|
print(nvalue_to_frequency(oms1.spectrum_bitmap.freq_index_min) * 1e-12,
|
|
nvalue_to_frequency(oms1.spectrum_bitmap.freq_index_max) * 1e-12)
|
|
# checks initial values consistancy
|
|
ind_max = len(oms1.spectrum_bitmap.bitmap) - 1
|
|
print('with guardband', oms1.spectrum_bitmap.getn(0),
|
|
oms1.spectrum_bitmap.getn(ind_max))
|
|
print('without guardband', oms1.spectrum_bitmap.freq_index_min,
|
|
oms1.spectrum_bitmap.freq_index_max)
|
|
# if requested slots exceed grid spectrum should not be assigned and assignment
|
|
# should return False
|
|
if ((nval1 - mval) < oms1.spectrum_bitmap.getn(0) or
|
|
(nval1 + mval-1) > oms1.spectrum_bitmap.getn(ind_max)):
|
|
with pytest.raises(SpectrumError):
|
|
oms1.assign_spectrum(nval1, mval)
|
|
for elem in oms1.spectrum_bitmap.bitmap:
|
|
assert elem == 1
|
|
else:
|
|
oms2.assign_spectrum(nval2, mval)
|
|
print(oms2.spectrum_bitmap.bitmap)
|
|
test2 = bitmap_sum(oms1.spectrum_bitmap.bitmap, oms2.spectrum_bitmap.bitmap)
|
|
print(test2)
|
|
range1 = range(oms1.spectrum_bitmap.geti(nval1) - mval,
|
|
oms1.spectrum_bitmap.geti(nval1) + mval -1)
|
|
range2 = range(oms2.spectrum_bitmap.geti(nval2) - mval,
|
|
oms2.spectrum_bitmap.geti(nval2) + mval -1)
|
|
for elem in range1:
|
|
print(f'value should be zero at index {elem}')
|
|
assert test2[elem] == 0
|
|
|
|
for elem in range2:
|
|
print(f'value should be zero at index {elem}')
|
|
assert test2[elem] == 0
|
|
|
|
def test_bitmap_assignment(setup):
|
|
""" test that a bitmap can be assigned
|
|
"""
|
|
network, oms_list = setup
|
|
random_oms = oms_list[2]
|
|
random_oms.assign_spectrum(13, 7)
|
|
|
|
btmp = deepcopy(random_oms.spectrum_bitmap.bitmap)
|
|
# try a first assignment that must pass
|
|
spectrum_btmp = Bitmap(cband_freq_min, cband_freq_max, grid=0.00625e12, guardband=0.15e12, bitmap=btmp)
|
|
|
|
# try a wrong assignment that should not pass
|
|
btmp = btmp[1:-1]
|
|
with pytest.raises(SpectrumError):
|
|
spectrum_btmp = Bitmap(cband_freq_min, cband_freq_max, grid=0.00625e12, guardband=0.15e12, bitmap=btmp)
|
|
|
|
@pytest.fixture()
|
|
def services(equipment):
|
|
""" 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
|
|
|
|
@pytest.fixture()
|
|
def requests(equipment, services):
|
|
""" 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
|
|
"""
|
|
network, oms_list = setup
|
|
req = [deepcopy(requests[1])]
|
|
paths = compute_path_dsjctn(network, equipment, req, [])
|
|
|
|
print(req)
|
|
for nval in range(100):
|
|
req = [deepcopy(requests[1])]
|
|
(center_n, startn, stopn), path_oms = spectrum_selection(paths[0], oms_list, 4)
|
|
pth_assign_spectrum(paths, req, oms_list, [find_reversed_path(paths[0])])
|
|
print(f'testing on following oms {path_oms}')
|
|
# check that only 96 channels are feasible
|
|
if nval >= 96:
|
|
print(center_n, startn, stopn)
|
|
print('only 96 channels of 4 slots pass in this grid')
|
|
assert center_n is None and startn is None and stopn is None
|
|
else:
|
|
print(center_n, startn, stopn)
|
|
print('at least 96 channels of 4 slots should pass in this grid')
|
|
assert center_n is not None and startn is not None and stopn is not None
|
|
|
|
req = [requests[2]]
|
|
paths = compute_path_dsjctn(network, equipment, req, [])
|
|
(center_n, startn, stopn), path_oms = spectrum_selection(paths[0], oms_list, 4, 478)
|
|
print(oms_list[0].spectrum_bitmap.freq_index_max)
|
|
print(oms_list[0])
|
|
print(center_n, startn, stopn)
|
|
print('spectrum selection error: should be None')
|
|
assert center_n is None and startn is None and stopn is None
|
|
(center_n, startn, stopn), path_oms = spectrum_selection(paths[0], oms_list, 4, 477)
|
|
print(center_n, startn, stopn)
|
|
print('spectrum selection error should not be None')
|
|
assert center_n is not None and startn is not None and stopn is not None
|
|
|
|
def test_reversed_direction(equipment, setup, requests, services):
|
|
""" 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 = correct_disjn(dsjn)
|
|
paths = compute_path_dsjctn(network, equipment, requests, dsjn)
|
|
rev_pths = []
|
|
for pth in paths:
|
|
if pth:
|
|
rev_pths.append(find_reversed_path(pth))
|
|
else:
|
|
rev_pths.append([])
|
|
# build the list of spectrum slots that will be used for each request. For this purpose
|
|
# play the selection part of path_assign_spectrum
|
|
spectrum_list = []
|
|
for i, pth in enumerate(paths):
|
|
if pth:
|
|
number_wl = ceil(requests[i].path_bandwidth / requests[i].bit_rate)
|
|
requested_m = ceil(requests[i].spacing / slot) * number_wl
|
|
(center_n, startn, stopn), path_oms = spectrum_selection(pth, oms_list, requested_m,
|
|
requested_n=None)
|
|
spectrum_list.append([center_n, startn, stopn])
|
|
else:
|
|
spectrum_list.append([])
|
|
pth_assign_spectrum(paths, requests, oms_list, rev_pths)
|
|
# pth-assign concatenates path and reversed path
|
|
for i, pth in enumerate(paths):
|
|
# verifies that each element (not trx and not roadm) in the path has same
|
|
# spectrum occupation
|
|
if pth:
|
|
this_path = [elem for elem in pth if not isinstance(elem, Roadm) and\
|
|
not isinstance(elem, Transceiver)]
|
|
print(f'path {[el.uid for el in this_path]}')
|
|
this_revpath = [elem for elem in rev_pths[i] if not isinstance(elem, Roadm) and\
|
|
not isinstance(elem, Transceiver)]
|
|
print(f'rev_path {[el.uid for el in this_revpath]}')
|
|
print('')
|
|
for j, elem in enumerate(this_revpath):
|
|
imin = elem.oms.spectrum_bitmap.geti(spectrum_list[i][1])
|
|
imax = elem.oms.spectrum_bitmap.geti(spectrum_list[i][2])
|
|
print(f'rev_elem {elem.uid}')
|
|
print(f' elem {this_path[len(this_path)-j-1].uid}')
|
|
print(f'\trev_spectrum: {elem.oms.spectrum_bitmap.bitmap[imin:imax]}')
|
|
print(f'\t spectrum: ' +\
|
|
f'{this_path[len(this_path)-j-1].oms.spectrum_bitmap.bitmap[imin:imax]}')
|
|
assert elem.oms.spectrum_bitmap.bitmap[imin:imax] == \
|
|
this_path[len(this_path)-j-1].oms.spectrum_bitmap.bitmap[imin:imax]
|