(self, x)
| 19 | target_freq: float = 0 |
| 20 | |
| 21 | def run(self, x): |
| 22 | # Freq shift if desired |
| 23 | if self.target_freq != 0: |
| 24 | x = x * np.exp(-2j * np.pi * self.target_freq * np.arange(len(x))/self.sample_rate) |
| 25 | |
| 26 | # Low pass filter to isolate FM signal |
| 27 | h = signal.firwin(101, cutoff=150e3, fs=self.sample_rate).astype(np.complex64) |
| 28 | x = np.convolve(x, h, "valid") |
| 29 | |
| 30 | x = signal.resample_poly(x, 10, int(self.sample_rate/500e3*10) ) # 500 kHz is the target |
| 31 | |
| 32 | x = np.diff(np.unwrap(np.angle(x))) # Demodulation |
| 33 | |
| 34 | # De-emphasis filter, H(s) = 1/(RC*s + 1), implemented as IIR via bilinear transform |
| 35 | bz, az = signal.bilinear(1, [75e-6, 1], fs=self.sample_rate) |
| 36 | x = signal.lfilter(bz, az, x) |
| 37 | |
| 38 | # decimate by 10 to get mono audio close to 48 kHz |
| 39 | x = x[::10] |
| 40 | |
| 41 | # normalize volume so its between -1 and +1 |
| 42 | x /= np.max(np.abs(x)) |
| 43 | |
| 44 | # some machines want int16s |
| 45 | x *= 32767 |
| 46 | x = x.astype(np.int16) |
| 47 | |
| 48 | # Also save to file, for testing |
| 49 | if False: |
| 50 | write('test.wav', 48000, x) |
| 51 | |
| 52 | # Create wav file out of real samples |
| 53 | byte_io = io.BytesIO(bytes()) |
| 54 | write(byte_io, 48000, x) |
| 55 | |
| 56 | samples_obj = { |
| 57 | "samples": base64.b64encode(byte_io.read()), |
| 58 | "data_type": "audio/wav", |
| 59 | } |
| 60 | return {"data_output": [samples_obj], "annotations": []} |
| 61 | |
| 62 | |
| 63 | if __name__ == "__main__": |
no outgoing calls
no test coverage detected