|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# -*- coding: utf-8 -*- |
| 3 | + |
| 4 | +# SPDX-License-Identifier: BSD-3-Clause |
| 5 | +# test_raman_simulation |
| 6 | +# Copyright (C) 2025 Telecom Infra Project and GNPy contributors |
| 7 | +# see AUTHORS.rst for a list of contributors |
| 8 | + |
| 9 | +""" |
| 10 | +Tests for the Raman simulation module, covering: |
| 11 | +- Attenuation profile calculation (no SRS) |
| 12 | +- Stimulated Raman Scattering with perturbative solver at various orders |
| 13 | +- Numerical vs perturbative solver comparison |
| 14 | +- Spectral tilt due to SRS |
| 15 | +- StimulatedRamanScattering data class |
| 16 | +- Simple single-link topology sanity checks |
| 17 | +""" |
| 18 | + |
| 19 | +from pathlib import Path |
| 20 | +from copy import deepcopy |
| 21 | + |
| 22 | +import pytest |
| 23 | +from numpy import array, ones, exp, outer, sqrt, allclose |
| 24 | +from numpy.testing import assert_allclose, assert_array_less |
| 25 | + |
| 26 | +from gnpy.core.info import create_input_spectral_information |
| 27 | +from gnpy.core.elements import Fiber, RamanFiber |
| 28 | +from gnpy.core.parameters import SimParams |
| 29 | +from gnpy.core.science_utils import RamanSolver, StimulatedRamanScattering |
| 30 | +from gnpy.tools.json_io import load_json |
| 31 | + |
| 32 | +TEST_DIR = Path(__file__).parent |
| 33 | + |
| 34 | + |
| 35 | +def _create_spectral_info(n_channels=10): |
| 36 | + """Helper: create spectral information with a few channels.""" |
| 37 | + f_min = 191.3e12 |
| 38 | + spacing = 50e9 |
| 39 | + f_max = f_min + (n_channels - 1) * spacing |
| 40 | + return create_input_spectral_information( |
| 41 | + f_min=f_min, f_max=f_max, roll_off=0.15, |
| 42 | + baud_rate=32e9, spacing=spacing, tx_osnr=40.0, tx_power=1e-3) |
| 43 | + |
| 44 | + |
| 45 | +@pytest.fixture |
| 46 | +def fiber(): |
| 47 | + """Standard SSMF fiber for testing.""" |
| 48 | + fiber_config = load_json(TEST_DIR / 'data' / 'test_science_utils_fiber_config.json') |
| 49 | + f = Fiber(**fiber_config) |
| 50 | + f.ref_pch_in_dbm = 0.0 |
| 51 | + return f |
| 52 | + |
| 53 | + |
| 54 | +@pytest.fixture |
| 55 | +def raman_fiber(): |
| 56 | + """RamanFiber with Raman pumps for testing.""" |
| 57 | + fiber_config = load_json(TEST_DIR / 'data' / 'test_science_utils_fiber_config.json') |
| 58 | + f = RamanFiber(**fiber_config) |
| 59 | + f.ref_pch_in_dbm = 0.0 |
| 60 | + return f |
| 61 | + |
| 62 | + |
| 63 | +@pytest.mark.usefixtures('set_sim_params') |
| 64 | +class TestAttenuationProfile: |
| 65 | + """Tests for RamanSolver.calculate_attenuation_profile (no SRS).""" |
| 66 | + |
| 67 | + def test_power_decreases_along_fiber(self, fiber): |
| 68 | + """Power should decrease monotonically along the fiber without SRS.""" |
| 69 | + spectral_info = _create_spectral_info(5) |
| 70 | + srs = RamanSolver.calculate_attenuation_profile(spectral_info, fiber) |
| 71 | + |
| 72 | + # Power at end should be less than at start for all channels |
| 73 | + for ch in range(srs.power_profile.shape[0]): |
| 74 | + assert srs.power_profile[ch, -1] < srs.power_profile[ch, 0] |
| 75 | + |
| 76 | + def test_loss_profile_starts_at_unity(self, fiber): |
| 77 | + """Loss profile should start at 1.0 (no loss at z=0).""" |
| 78 | + spectral_info = _create_spectral_info(5) |
| 79 | + srs = RamanSolver.calculate_attenuation_profile(spectral_info, fiber) |
| 80 | + |
| 81 | + assert_allclose(srs.loss_profile[:, 0], ones(srs.loss_profile.shape[0]), rtol=1e-10) |
| 82 | + |
| 83 | + def test_loss_profile_consistent_with_alpha(self, fiber): |
| 84 | + """Loss at fiber end should be consistent with exp(-alpha * L).""" |
| 85 | + spectral_info = _create_spectral_info(5) |
| 86 | + srs = RamanSolver.calculate_attenuation_profile(spectral_info, fiber) |
| 87 | + |
| 88 | + alpha = fiber.alpha(spectral_info.frequency) |
| 89 | + expected_loss = exp(-alpha * fiber.params.length) |
| 90 | + assert_allclose(srs.loss_profile[:, -1], expected_loss, rtol=1e-3) |
| 91 | + |
| 92 | + def test_rho_is_sqrt_of_loss(self, fiber): |
| 93 | + """The rho field should be sqrt(loss_profile).""" |
| 94 | + spectral_info = _create_spectral_info(5) |
| 95 | + srs = RamanSolver.calculate_attenuation_profile(spectral_info, fiber) |
| 96 | + |
| 97 | + assert_allclose(srs.rho, sqrt(srs.loss_profile), rtol=1e-10) |
| 98 | + |
| 99 | + def test_z_array_spans_fiber(self, fiber): |
| 100 | + """z array should start at 0 and end at fiber length.""" |
| 101 | + spectral_info = _create_spectral_info(5) |
| 102 | + srs = RamanSolver.calculate_attenuation_profile(spectral_info, fiber) |
| 103 | + |
| 104 | + assert srs.z[0] == 0 |
| 105 | + assert srs.z[-1] == fiber.params.length |
| 106 | + |
| 107 | + |
| 108 | +@pytest.mark.usefixtures('set_sim_params') |
| 109 | +class TestStimulatedRamanScattering: |
| 110 | + """Tests for SRS with perturbative solver at various orders.""" |
| 111 | + |
| 112 | + def test_srs_order_1(self, fiber): |
| 113 | + """SRS order 1 should produce a spectral tilt (lower freq gains power from higher freq).""" |
| 114 | + spectral_info = _create_spectral_info(10) |
| 115 | + SimParams.set_params({'raman_params': {'flag': True, 'order': 1}}) |
| 116 | + srs = RamanSolver.calculate_stimulated_raman_scattering(spectral_info, fiber) |
| 117 | + |
| 118 | + # With SRS, lower frequencies should have relatively more power than without SRS |
| 119 | + loss_first = srs.loss_profile[0, -1] |
| 120 | + loss_last = srs.loss_profile[-1, -1] |
| 121 | + # Lower frequency channel should have less loss (more gain from Raman) |
| 122 | + assert loss_first > loss_last, "SRS should tilt spectrum: lower freq should have less loss" |
| 123 | + |
| 124 | + def test_srs_order_2_refines_order_1(self, fiber): |
| 125 | + """Higher order perturbative solution should differ slightly from order 1.""" |
| 126 | + spectral_info = _create_spectral_info(10) |
| 127 | + |
| 128 | + SimParams.set_params({'raman_params': {'flag': True, 'order': 1}}) |
| 129 | + srs_o1 = RamanSolver.calculate_stimulated_raman_scattering(deepcopy(spectral_info), fiber) |
| 130 | + |
| 131 | + SimParams.set_params({'raman_params': {'flag': True, 'order': 2}}) |
| 132 | + srs_o2 = RamanSolver.calculate_stimulated_raman_scattering(deepcopy(spectral_info), fiber) |
| 133 | + |
| 134 | + # Should be close but not identical (difference may be very small for low channel count) |
| 135 | + assert_allclose(srs_o1.power_profile[:, -1], srs_o2.power_profile[:, -1], rtol=0.1) |
| 136 | + |
| 137 | + def test_srs_orders_converge(self, fiber): |
| 138 | + """Higher perturbative orders should converge: order 3 closer to order 4 than order 1 to order 2.""" |
| 139 | + spectral_info = _create_spectral_info(10) |
| 140 | + |
| 141 | + results = {} |
| 142 | + for order in [1, 2, 3, 4]: |
| 143 | + SimParams.set_params({'raman_params': {'flag': True, 'order': order}}) |
| 144 | + srs = RamanSolver.calculate_stimulated_raman_scattering(deepcopy(spectral_info), fiber) |
| 145 | + results[order] = srs.power_profile[:, -1].copy() |
| 146 | + |
| 147 | + # Difference between consecutive orders should decrease |
| 148 | + diff_12 = abs(results[1] - results[2]).max() |
| 149 | + diff_23 = abs(results[2] - results[3]).max() |
| 150 | + diff_34 = abs(results[3] - results[4]).max() |
| 151 | + assert diff_23 < diff_12, "Perturbative orders should converge" |
| 152 | + assert diff_34 < diff_23, "Perturbative orders should converge" |
| 153 | + |
| 154 | + def test_numerical_vs_perturbative(self, fiber): |
| 155 | + """Numerical and perturbative methods should produce similar results.""" |
| 156 | + spectral_info = _create_spectral_info(10) |
| 157 | + |
| 158 | + SimParams.set_params({'raman_params': { |
| 159 | + 'flag': True, 'order': 2, 'method': 'perturbative', |
| 160 | + 'solver_spatial_resolution': 10}}) |
| 161 | + srs_pert = RamanSolver.calculate_stimulated_raman_scattering(deepcopy(spectral_info), fiber) |
| 162 | + |
| 163 | + SimParams.set_params({'raman_params': { |
| 164 | + 'flag': True, 'method': 'numerical', |
| 165 | + 'solver_spatial_resolution': 10}}) |
| 166 | + srs_num = RamanSolver.calculate_stimulated_raman_scattering(deepcopy(spectral_info), fiber) |
| 167 | + |
| 168 | + # Should agree within 5% |
| 169 | + assert_allclose(srs_pert.power_profile[:, -1], srs_num.power_profile[:, -1], rtol=0.05) |
| 170 | + |
| 171 | + |
| 172 | +@pytest.mark.usefixtures('set_sim_params') |
| 173 | +class TestRamanFiberPropagation: |
| 174 | + """End-to-end tests for fiber propagation with Raman effects.""" |
| 175 | + |
| 176 | + def test_propagation_output_power_reasonable(self, fiber): |
| 177 | + """Output power after propagation should be reasonable (not zero, not amplified).""" |
| 178 | + spectral_info = _create_spectral_info(10) |
| 179 | + SimParams.set_params({'raman_params': {'flag': True, 'order': 2}}) |
| 180 | + fiber.ref_pch_in_dbm = 0.0 |
| 181 | + spectral_info_out = fiber(spectral_info) |
| 182 | + |
| 183 | + # Output signal should be positive |
| 184 | + assert (spectral_info_out.signal > 0).all() |
| 185 | + # Output should be less than input (fiber attenuates) |
| 186 | + assert spectral_info_out.ptot_dbm < 0 # input was 0 dBm per channel |
| 187 | + |
| 188 | + def test_raman_tilt_in_propagation(self): |
| 189 | + """Propagation with SRS should show spectral tilt vs without SRS.""" |
| 190 | + spectral_info_no_raman = _create_spectral_info(10) |
| 191 | + spectral_info_raman = deepcopy(spectral_info_no_raman) |
| 192 | + |
| 193 | + fiber_config = load_json(TEST_DIR / 'data' / 'test_science_utils_fiber_config.json') |
| 194 | + |
| 195 | + # Without Raman |
| 196 | + SimParams.set_params({'raman_params': {'flag': False}}) |
| 197 | + fiber_no_raman = Fiber(**fiber_config) |
| 198 | + fiber_no_raman.ref_pch_in_dbm = 0.0 |
| 199 | + out_no_raman = fiber_no_raman(spectral_info_no_raman) |
| 200 | + |
| 201 | + # With Raman |
| 202 | + SimParams.set_params({'raman_params': {'flag': True, 'order': 2}}) |
| 203 | + fiber_raman = Fiber(**fiber_config) |
| 204 | + fiber_raman.ref_pch_in_dbm = 0.0 |
| 205 | + out_raman = fiber_raman(spectral_info_raman) |
| 206 | + |
| 207 | + # Raman should cause spectral tilt: power difference between first and last channel |
| 208 | + # should be different with vs without SRS |
| 209 | + tilt_no_raman = out_no_raman.signal[0] / out_no_raman.signal[-1] |
| 210 | + tilt_raman = out_raman.signal[0] / out_raman.signal[-1] |
| 211 | + assert tilt_raman > tilt_no_raman, "SRS should increase spectral tilt" |
| 212 | + |
| 213 | + def test_nli_generated_during_propagation(self, fiber): |
| 214 | + """Propagation should generate NLI noise.""" |
| 215 | + spectral_info = _create_spectral_info(10) |
| 216 | + SimParams.set_params({'raman_params': {'flag': True, 'order': 2}}) |
| 217 | + spectral_info_out = fiber(spectral_info) |
| 218 | + |
| 219 | + # NLI should be nonzero |
| 220 | + assert (spectral_info_out.nli > 0).all() |
| 221 | + |
| 222 | + |
| 223 | +@pytest.mark.usefixtures('set_sim_params') |
| 224 | +class TestStimulatedRamanScatteringDataClass: |
| 225 | + """Tests for the StimulatedRamanScattering data class.""" |
| 226 | + |
| 227 | + def test_construction(self): |
| 228 | + """Basic construction test.""" |
| 229 | + power = array([[1.0, 0.5], [1.0, 0.6]]) |
| 230 | + loss = array([[1.0, 0.5], [1.0, 0.6]]) |
| 231 | + freq = array([191.3e12, 196.1e12]) |
| 232 | + z = array([0, 80000]) |
| 233 | + srs = StimulatedRamanScattering(power, loss, freq, z) |
| 234 | + |
| 235 | + assert_allclose(srs.power_profile, power) |
| 236 | + assert_allclose(srs.loss_profile, loss) |
| 237 | + assert_allclose(srs.frequency, freq) |
| 238 | + assert_allclose(srs.z, z) |
| 239 | + assert_allclose(srs.rho, sqrt(loss)) |
0 commit comments