Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Create utils/DSP for DSP-only stuff. #116

Merged
merged 1 commit into from
May 25, 2023
Merged

Create utils/DSP for DSP-only stuff. #116

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 16 additions & 115 deletions torchsig/datasets/synthetic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,39 +1,17 @@
import torch
import pickle
import itertools
import numpy as np
from copy import deepcopy
from scipy import signal as sp
from collections import OrderedDict
from torch.utils.data import ConcatDataset
from typing import Tuple, Any, List, Union, Optional
from torchsig.utils.dataset import SignalDataset
from torchsig.utils.types import SignalData, SignalDescription
from torchsig.utils.dsp import low_pass, convolve, rrc_taps, gaussian_taps
from torchsig.transforms.functional import IntParameter, FloatParameter
from torchsig.datasets import estimate_filter_length


def torchsig_convolve(
signal: np.ndarray, taps: np.ndarray, gpu: bool = False
) -> np.ndarray:
return sp.convolve(signal, taps, "same")
# This will run into issues is signal is smaller than taps
# torch_signal = torch.from_numpy(signal.astype(np.complex128)).reshape(1, -1)
# torch_taps = torch.flip(
# torch.from_numpy(taps.astype(np.complex128)).reshape(1, 1, -1), dims=(2,)
# )
# if gpu:
# result = torch.nn.functional.conv1d(
# torch_signal.cuda(), torch_taps.cuda(), padding=torch_signal.shape[0] - 1
# )
# return result.cpu().numpy()[0]

# result = torch.nn.functional.conv1d(
# torch_signal, torch_taps, padding=torch_signal.shape[0] - 1
# )
# return result.numpy()[0]


def remove_corners(const):
spacing = 2.0 / (np.sqrt(len(const)) - 1)
cutoff = spacing * (np.sqrt(len(const)) / 6 - 0.5)
Expand Down Expand Up @@ -375,49 +353,18 @@ def _generate_samples(self, item: Tuple) -> np.ndarray:
pulse_shape_filter_span = int(
(pulse_shape_filter_length - 1) / 2
) # convert filter length into the span
self.pulse_shape_filter = self._rrc_taps(
pulse_shape_filter_span, signal_description.excess_bandwidth
)
filtered = torchsig_convolve(
zero_padded, self.pulse_shape_filter, gpu=self.use_gpu
self.pulse_shape_filter = rrc_taps(
self.iq_samples_per_symbol,
pulse_shape_filter_span,
signal_description.excess_bandwidth,
)
filtered = convolve(zero_padded, self.pulse_shape_filter)

if not self.random_data:
np.random.set_state(orig_state) # return numpy back to its previous state

return filtered[-self.num_iq_samples :]

def _rrc_taps(self, size_in_symbols: int, alpha: float = 0.35) -> np.ndarray:
# this could be made into a transform
M = size_in_symbols
Ns = float(self.iq_samples_per_symbol)
n = np.arange(-M * Ns, M * Ns + 1)
taps = np.zeros(int(2 * M * Ns + 1))
for i in range(int(2 * M * Ns + 1)):
# handle the discontinuity at t=+-Ns/(4*alpha)
if n[i] * 4 * alpha == Ns or n[i] * 4 * alpha == -Ns:
taps[i] = (
1
/ 2.0
* (
(1 + alpha) * np.sin((1 + alpha) * np.pi / (4.0 * alpha))
- (1 - alpha) * np.cos((1 - alpha) * np.pi / (4.0 * alpha))
+ (4 * alpha)
/ np.pi
* np.sin((1 - alpha) * np.pi / (4.0 * alpha))
)
)
else:
taps[i] = 4 * alpha / (np.pi * (1 - 16 * alpha**2 * (n[i] / Ns) ** 2))
taps[i] = taps[i] * (
np.cos((1 + alpha) * np.pi * n[i] / Ns)
+ np.sinc((1 - alpha) * n[i] / Ns)
* (1 - alpha)
* np.pi
/ (4.0 * alpha)
)
return taps


class OFDMDataset(SyntheticDataset):
"""OFDM Dataset
Expand Down Expand Up @@ -503,18 +450,8 @@ def __init__(
self.use_gpu = use_gpu
self.index = []
if "lpf" in sidelobe_suppression_methods:
# Precompute LPF
cutoff = 0.3
transition_bandwidth = (0.5 - cutoff) / 4
num_taps = estimate_filter_length(transition_bandwidth)
self.taps = sp.firwin(
num_taps,
cutoff,
width=transition_bandwidth,
window=sp.get_window("blackman", num_taps),
scale=True,
fs=1,
)
self.taps = low_pass(cutoff=cutoff, transition_bandwidth=(0.5 - cutoff) / 4)

# Precompute all possible random symbols for speed at sample generation
self.random_symbols = []
Expand Down Expand Up @@ -585,7 +522,7 @@ def _generate_samples(self, item: Tuple) -> np.ndarray:
orig_state = np.random.get_state()
if not self.random_data:
np.random.seed(index)

if mod_type == "random":
symbols_idxs = np.random.randint(0, 1024, size=self.num_iq_samples)
const_idxes = np.random.choice(
Expand Down Expand Up @@ -716,24 +653,15 @@ def _generate_samples(self, item: Tuple) -> np.ndarray:
elif sidelobe_suppression_method == "lpf":
flattened = cyclic_prefixed.T.flatten()
# Apply pre-computed LPF
output = torchsig_convolve(flattened, self.taps, gpu=self.use_gpu)[:-50]
output = convolve(flattened, self.taps)[:-50]

elif sidelobe_suppression_method == "rand_lpf":
flattened = cyclic_prefixed.T.flatten()
# Generate randomized LPF
cutoff = np.random.uniform(0.25, 0.475)
transition_bandwidth = (0.5 - cutoff) / 4
num_taps = estimate_filter_length(transition_bandwidth)
taps = sp.firwin(
num_taps,
cutoff,
width=transition_bandwidth,
window=sp.get_window("blackman", num_taps),
scale=True,
fs=1,
)
taps = low_pass(cutoff=cutoff, transition_bandwidth=(0.5 - cutoff) / 4)
# Apply random LPF
output = torchsig_convolve(flattened, taps, gpu=self.use_gpu)[:-num_taps]
output = convolve(flattened, taps)[: -len(taps)]
else:
# Apply appropriate windowing technique
window_len = cyclic_prefix_len
Expand Down Expand Up @@ -939,9 +867,9 @@ def _generate_samples(self, item: Tuple) -> np.ndarray:

if "g" in const_name:
# GMSK, GFSK
taps = self._gaussian_taps(samples_per_symbol_recalculated, bandwidth)
taps = gaussian_taps(samples_per_symbol_recalculated, bandwidth)
signal_description.excess_bandwidth = bandwidth
filtered = torchsig_convolve(symbols_repeat, taps, gpu=self.use_gpu)
filtered = convolve(symbols_repeat, taps)
else:
# FSK, MSK
filtered = symbols_repeat
Expand All @@ -954,44 +882,17 @@ def _generate_samples(self, item: Tuple) -> np.ndarray:
modulated = np.exp(phase)

if self.random_pulse_shaping:
# Apply a randomized LPF simulating a noisy detector/burst extractor, then downsample to ~fs/2 bw
# accept the cutoff-frequency of the filter as external
# parameter, randomized as part of outer framework
cutoff_frequency = bandwidth
# calculate transition bandwidth. a larger cutoff frequency requires
# a smaller transition bandwidth, and a smaller cutoff frequency
# allows for a larger transition bandwidth
transition_bandwidth = (1.0 / 2 - (cutoff_frequency)) / 4
# estimate number of taps needed to implement filter
num_taps = estimate_filter_length(transition_bandwidth)

# design the filter
taps = sp.firwin(
num_taps,
cutoff_frequency,
width=transition_bandwidth,
window=sp.get_window("blackman", num_taps),
scale=True,
fs=1,
taps = low_pass(
cutoff=bandwidth / 2, transition_bandwidth=(0.5 - bandwidth / 2) / 4
)
# apply the filter
modulated = torchsig_convolve(modulated, taps, gpu=self.use_gpu)
modulated = convolve(modulated, taps)

if not self.random_data:
np.random.set_state(orig_state) # return numpy back to its previous state

return modulated[-self.num_iq_samples :]

def _gaussian_taps(self, samples_per_symbol, BT: float = 0.35) -> np.ndarray:
# pre-modulation Bb*T product which sets the bandwidth of the Gaussian lowpass filter
M = 4 # duration in symbols
n = np.arange(-M * samples_per_symbol, M * samples_per_symbol + 1)
p = np.exp(
-2 * np.pi**2 * BT**2 / np.log(2) * (n / float(samples_per_symbol)) ** 2
)
p = p / np.sum(p)
return p

def _mod_index(self, const_name):
# returns the modulation index based on the modulation
if "gfsk" in const_name:
Expand Down
47 changes: 10 additions & 37 deletions torchsig/datasets/wideband.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from typing import Tuple, Any, List, Optional, Callable, Union
from torchsig.utils.dataset import SignalDataset
from torchsig.utils.types import SignalData, SignalDescription
from torchsig.utils.dsp import low_pass
from torchsig.datasets.synthetic import OFDMDataset, ConstellationDataset, FSKDataset
from torchsig.transforms import (
SignalTransform,
Expand All @@ -19,7 +20,7 @@
AddNoise,
RandomPhaseShift,
RayleighFadingChannel,
Normalize,
Normalize,
RandomResample,
RandomTimeShift,
RandomFrequencyShift,
Expand All @@ -34,7 +35,7 @@
to_distribution,
uniform_continuous_distribution,
uniform_discrete_distribution,
FloatParameter,
FloatParameter,
NumericParameter,
)
from torchsig.datasets import estimate_filter_length
Expand Down Expand Up @@ -88,14 +89,9 @@ def generate_iq(self):
center = lower + bandwidth / 2

# Filter noise
num_taps = estimate_filter_length((0.5 - 0.02 * bandwidth) / 4)
sinusoid = np.exp(2j * np.pi * center * np.linspace(0, num_taps - 1, num_taps))
taps = signal.firwin(
num_taps,
bandwidth,
width=bandwidth * 0.02,
window=signal.get_window("blackman", num_taps),
scale=True,
taps = low_pass(cutoff=bandwidth / 2, transition_bandwidth=(0.5 - bandwidth / 2) / 4)
sinusoid = np.exp(
2j * np.pi * center * np.linspace(0, len(taps) - 1, len(taps))
)
taps = taps * sinusoid
iq_samples = signal.fftconvolve(iq_samples, taps, mode="same")
Expand Down Expand Up @@ -362,15 +358,8 @@ def generate_iq(self):
iq_samples = iq_samples[
-int(self.num_iq_samples * self.duration * oversample) :
]

num_taps = estimate_filter_length((0.5 - 0.02 / oversample) / 4)

taps = signal.firwin(
num_taps,
1 / oversample,
width=1 / oversample * 0.02,
window=signal.get_window("blackman", num_taps),
scale=True,
taps = low_pass(
cutoff=1 / oversample / 2, transition_bandwidth=(0.5 - 1 / oversample / 2) / 4
)
iq_samples = np.convolve(iq_samples, taps, mode="same")

Expand Down Expand Up @@ -466,15 +455,7 @@ def generate_iq(self):
)

# Filter around center
num_taps = estimate_filter_length((0.5 - 0.02 * 0.5) / 4)

taps = signal.firwin(
num_taps,
0.5,
width=0.5 * 0.02,
window=signal.get_window("blackman", num_taps),
scale=True,
)
taps = low_pass(cutoff=1 / 4, transition_bandwidth=(0.5 - 1 / 4) / 4)
iq_samples = signal.fftconvolve(iq_samples, taps, mode="same")

# Decimate back down to correct sample rate
Expand Down Expand Up @@ -573,15 +554,7 @@ def generate_iq(self):
)

# Filter around center
num_taps = estimate_filter_length((0.5 - 0.5 * 0.02) / 4)

taps = signal.firwin(
num_taps,
0.5,
width=0.5 * 0.02,
window=signal.get_window("blackman", num_taps),
scale=True,
)
taps = low_pass(cutoff=1 / 4, transition_bandwidth=(0.5 - 1 / 4) / 4)
iq_samples = signal.fftconvolve(iq_samples, taps, mode="same")

# Decimate back down to correct sample rate
Expand Down
Loading