import h5py
from matplotlib.ticker import EngFormatter, MultipleLocator
from torchsig.datasets.datasets import TorchSigIterableDataset
from torchsig.utils.data_loading import WorkerSeedingDataLoader
from torchsig.signals.builders.fm import FMSignalGenerator
from torchsig.utils.writer import DatasetCreator, default_collate_fn
from torchsig.transforms.transforms import Spectrogram
from torchsig.transforms.impairments import Impairments
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle
import numpy as np
dataset_metadata = {
"num_iq_samples_dataset": 108_032,
"fft_size": 512,
"num_signals_min": 1,
"num_signals_max": 1,
"sample_rate": 20e6,
"noise_power_db": 0,
"snr_db_min": 50,
"snr_db_max": 50,
"signal_duration_min": 0.00378112,
"signal_duration_max": 0.00378112,
"bandwidth_min": 500e3,
"bandwidth_max": 1e6,
"signal_duration_in_samples_min": 75622,
"signal_duration_in_samples_max": 75622,
"frequency_min": -10e6,
"frequency_max": 10e6,
"signal_center_freq_min": -9e6,
"signal_center_freq_max": 9e6,
"class_list": "fm",
"class_distribution": "1",
"fft_stride": 512,
"cochannel_overlap_probability": 0,
}
samp_period = 1 / dataset_metadata["sample_rate"]
rf_min = 90e6
fc = 100e6
rf_max = 110e6
ex_duration = 0.0054016
def get_dataset():
impairments = Impairments(level=0) # noiseless channel
iter_dataset = TorchSigIterableDataset(
signal_generators=[FMSignalGenerator(**dataset_metadata)], # NOTE: added in next step
metadata=dataset_metadata,
transforms=[impairments.dataset_transforms, Spectrogram(fft_size=512)],
component_transforms=[impairments.signal_transforms],
target_labels=['class_name', 'start_in_samples', 'duration_in_samples', 'lower_freq', 'upper_freq', 'snr_db'],
seed=14,
)
dataloader = WorkerSeedingDataLoader(
iter_dataset,
batch_size=8,
num_workers=10,
collate_fn=default_collate_fn,
)
dataset_creator = DatasetCreator(
dataset_length=200,
dataloader=dataloader,
root='dataset',
overwrite=True,
multithreading=True,
)
dataset_creator.create()
print(f"Dataset created at {dataset_creator.root}")
def plot_spec_bboxes():
with h5py.File("dataset/data.h5", "r") as f_raw:
example_idxs_raw = sorted(f_raw['data'].keys(), key=lambda value: int(value))
# Plot and save the spectrogram with physical axes (time, frequency)
for i in range(0, len(f_raw['data'].keys()), 2):
spectrogram = f_raw['data'][str(i)][()]
spectrogram = np.flip(spectrogram, axis=0) # flip spectrogram vertically to match the RF axis orientation (lowest matrix rows should correspond to the highest RF frequencies)
emitter = f_raw["metadata"][str(i+1)]
plt.figure(figsize=(10, 4))
plt.imshow(
spectrogram,
aspect='auto',
origin='lower',
extent=[0, ex_duration, rf_min, rf_max],
)
# add metadata
ax = plt.gca()
mod_name = emitter['class_name'][()].decode("utf-8").lower()
snr_db = emitter['snr_db'][()]
t_start_samp = emitter['start_in_samples'][()]
t_duration_samp = emitter['duration_in_samples'][()]
t_start_s = t_start_samp * samp_period
t_duration_s = t_duration_samp * samp_period
bb_start_freq_hz = emitter['_lower_frequency'][()]
bb_end_freq_hz = emitter['_upper_frequency'][()]
bw_hz = emitter['bandwidth'][()]
center_freq_hz = emitter['center_freq'][()]
# HACK: this solves a bug in TorchSig where the sign of the baseband frequencies is flipped for some examples for unknown reasons
if (
center_freq_hz < 0 and (bb_start_freq_hz > 0 or bb_end_freq_hz > 0)
or
center_freq_hz > 0 and (bb_start_freq_hz < 0 or bb_end_freq_hz < 0)
):
bb_start_freq_hz, bb_end_freq_hz = -1*bb_end_freq_hz, -1*bb_start_freq_hz
# Draw bounding box (true)
rf_start_hz = fc + bb_start_freq_hz
rect = Rectangle(
(t_start_s, rf_start_hz), t_duration_s, bw_hz,
linewidth=2, edgecolor='red', facecolor='none'
)
ax.add_patch(rect)
# format axes
plt.colorbar(label='Intensity [dB]')
ax.xaxis.set_major_formatter(EngFormatter(unit='s', sep=' '))
ax.yaxis.set_major_locator(MultipleLocator(10e6))
ax.yaxis.set_major_formatter(EngFormatter(unit='Hz', sep=' '))
plt.title('Spectrogram Example')
plt.xlabel('Time')
plt.ylabel('Frequency')
# Save the figure
plt.savefig(f"spectrogram_{i}.png", bbox_inches='tight')
def main():
get_dataset()
plot_spec_bboxes()
if __name__ == "__main__":
main()
This a generated spectrogram.
We have neither information about the x/y axes nor the support vector of them. Is x axis in time? frequency? What are their values? The dataset metadata on the axis are not self-contained, we need the source code to recover the support vector of the spectrogram. It should be included in the generated HDF5 file.
Is there an existing feature already?
Description
It follows an MWE that generates a spectrogram dataset from TorchSig pipeline
This a generated spectrogram.
We have neither information about the x/y axes nor the support vector of them. Is x axis in time? frequency? What are their values? The dataset metadata on the axis are not self-contained, we need the source code to recover the support vector of the spectrogram. It should be included in the generated HDF5 file.