chromatic dispersion applied

models/optical_channel.py

@@ -0,0 +1,118 @@
+import matplotlib.pyplot as plt
+
+import defs
+import numpy as np
+import math
+from scipy.fft import fft, ifft
+
+
+class OpticalChannel(defs.Channel):
+    def __init__(self, noise_level, dispersion, symbol_rate, sample_rate, length, show_graphs=False, **kwargs):
+        """
+        :param noise_level: Noise level in dB
+        :param dispersion: dispersion coefficient is ps^2/km
+        :param symbol_rate: Symbol rate of modulated signal in Hz
+        :param sample_rate: Sample rate of time-domain model (time steps in simulation) in Hz
+        :param length: fibre length in km
+        :param show_graphs: if graphs should be displayed or not
+
+        Optical Channel class constructor
+        """
+        super().__init__(**kwargs)
+        self.noise = 10 ** (noise_level / 10)
+
+        self.dispersion = dispersion
+        self.symbol_rate = symbol_rate
+        self.symbol_period = 1 / self.symbol_rate
+        self.sample_rate = sample_rate
+        self.sample_period = 1 / self.sample_rate
+        self.length = length
+        self.show_graphs = show_graphs
+
+    def __get_time_domain(self, symbol_vals):
+        samples_per_symbol = int(self.sample_rate / self.symbol_rate)
+        samples = int(symbol_vals.shape[0] * samples_per_symbol)
+
+        symbol_vals_a = np.repeat(symbol_vals, repeats=samples_per_symbol, axis=0)
+        t = np.linspace(start=0, stop=samples * self.sample_period, num=samples)
+        val_t = symbol_vals_a[:, 0] * np.cos(2 * math.pi * symbol_vals_a[:, 2] * t + symbol_vals_a[:, 1])
+
+        return t, val_t
+
+    def __time_to_frequency(self, values):
+        val_f = fft(values)
+        f = np.linspace(0.0, 1 / (2 * self.sample_period), (values.size // 2))
+        f_neg = -1 * np.flip(f)
+        f = np.concatenate((f, f_neg), axis=0)
+        return f, val_f
+
+    def __frequency_to_time(self, values):
+        val_t = ifft(values)
+        t = np.linspace(start=0, stop=values.size * self.sample_period, num=values.size)
+        return t, val_t
+
+    def __apply_dispersion(self, values):
+        # Obtain fft
+        f, val_f = self.__time_to_frequency(values)
+
+        if self.show_graphs:
+            plt.plot(f, val_f)
+            plt.title('frequency domain (pre-distortion)')
+            plt.show()
+
+        np.savetxt("foo.csv", f, delimiter=",")
+        np.savetxt("barr.csv", np.real(val_f), delimiter=",")
+        np.savetxt("bari.csv", np.imag(val_f), delimiter=",")
+
+        # Apply distortion
+        dist_val_f = val_f * np.exp(0.5j * self.dispersion * self.length * np.power(2 * math.pi * f, 2))
+
+        if self.show_graphs:
+            plt.plot(f, dist_val_f)
+            plt.title('frequency domain (post-distortion)')
+            plt.show()
+
+        # Inverse fft
+        t, val_t = self.__frequency_to_time(dist_val_f)
+
+        return t, val_t
+
+    def forward(self, values):
+        # Converting APF representation to time-series
+        t, val_t = self.__get_time_domain(values)
+
+        if self.show_graphs:
+            plt.plot(t, val_t)
+            plt.title('time domain (raw)')
+            plt.show()
+
+        # Adding AWGN
+        val_t += np.random.normal(0, 1, val_t.shape) * self.noise
+
+        if self.show_graphs:
+            plt.plot(t, val_t)
+            plt.title('time domain (AWGN)')
+            plt.show()
+
+        # Applying chromatic dispersion
+        t, val_t = self.__apply_dispersion(val_t)
+
+        if self.show_graphs:
+            plt.plot(t, val_t)
+            plt.title('time domain (post-distortion)')
+            plt.show()
+
+        return t, val_t
+
+
+if __name__ == '__main__':
+    # Simple OOK modulation
+    num_of_symbols = 10
+    symbol_vals = np.zeros((num_of_symbols, 3))
+
+    symbol_vals[:, 0] = np.random.randint(2, size=symbol_vals.shape[0])
+    symbol_vals[:, 2] = 10e6
+
+    channel = OpticalChannel(noise_level=-20, dispersion=-21.7, symbol_rate=100e3,
+                             sample_rate=500e6, length=100, show_graphs=True)
+    time, v = channel.forward(symbol_vals)