(container_client, basename)
| 26 | |
| 27 | |
| 28 | def spectro_maker(container_client, basename): |
| 29 | metaname = basename + ".sigmf-meta" |
| 30 | dataname = basename + ".sigmf-data" |
| 31 | |
| 32 | # Download data file |
| 33 | try: |
| 34 | data_bytes = container_client.download_blob( |
| 35 | dataname, offset=offset, length=num_bytes |
| 36 | ).readall() |
| 37 | except Exception: |
| 38 | print(basename + ".sigmf-data not found") |
| 39 | return |
| 40 | |
| 41 | # Download meta file |
| 42 | try: |
| 43 | metainfo = container_client.download_blob(metaname).readall().decode("utf-8") |
| 44 | except Exception: |
| 45 | print(basename + ".sigmf-meta not found") |
| 46 | return |
| 47 | |
| 48 | metainfo = json.loads(metainfo) |
| 49 | |
| 50 | dtype = metainfo["global"]["core:datatype"] |
| 51 | if dtype == "ci16_le" or dtype == "ci16": |
| 52 | samples = np.frombuffer(data_bytes, dtype=np.int16) |
| 53 | samples = samples[::2] + 1j * samples[1::2] |
| 54 | elif dtype == "cf32_le": |
| 55 | samples = np.frombuffer(data_bytes, dtype=np.complex64) |
| 56 | elif dtype == "ci8" or dtype == "i8": |
| 57 | samples = np.frombuffer(data_bytes, dtype=np.int8) |
| 58 | samples = samples[::2] + 1j * samples[1::2] |
| 59 | else: |
| 60 | print("Datatype " + dtype + " not implemented") |
| 61 | return |
| 62 | |
| 63 | # Generate spectrogram |
| 64 | num_rows = int(np.floor(len(samples) / fftSize)) |
| 65 | spectrogram = np.zeros((num_rows, fftSize)) |
| 66 | for i in range(num_rows): |
| 67 | spectrogram[i, :] = 10 * np.log10( |
| 68 | np.abs( |
| 69 | np.fft.fftshift(np.fft.fft(samples[i * fftSize: (i + 1) * fftSize])) |
| 70 | ) |
| 71 | ** 2 |
| 72 | ) |
| 73 | |
| 74 | fig = plt.figure(frameon=False) |
| 75 | ax = plt.Axes(fig, [0.0, 0.0, 1.0, 1.0]) |
| 76 | ax.set_axis_off() |
| 77 | fig.add_axes(ax) |
| 78 | ax.imshow(spectrogram, cmap="viridis", aspect="auto", vmin=30+np.min(np.min(spectrogram))) # used to be jet |
| 79 | |
| 80 | img_buf = io.BytesIO() |
| 81 | plt.savefig(img_buf, bbox_inches="tight", pad_inches=0) |
| 82 | img_buf.seek(0) |
| 83 | im = Image.open(img_buf) |
| 84 | img_byte_arr = io.BytesIO() |
| 85 | im.convert("RGB").save(img_byte_arr, format="jpeg") |
no test coverage detected