jono/sdrplay-fm-radio
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

Added ability to control either the SDRPlay or nooelec radio

Browse Source
This commit is contained in:
Jono Targett 2023-05-15 15:36:50 +09:30
parent a4e4a11d86
commit 802f3c0051
3 changed files with 195 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
microservice.py
View File

@ -147,6 +147,27 @@ def end_stream(radio):
except Exception as e:
return str(e), 400
@app.route('/radio/<radio>/info')
def radio_info(radio):
"""Get information about a radio.
---
parameters:
- name: radio
description: Radio device driver name, or serial number.
in: path
type: string
required: true
responses:
200:
description: JSON
400:
description: The specified radio is not connected.
"""
try:
return jsonify(radios[radio].get_info())
except Exception as e:
return str(e), 400
if __name__ == '__main__':
import subprocess

60
radio.py
View File

@ -6,6 +6,7 @@ import sys
import subprocess
import struct
from soapyhelpers import *
from samplerates import *
class Radio:
@ -23,16 +24,22 @@ class Radio:
if self.device is None:
raise RuntimeError("Failed to connect to radio device")
self.capabilities = self._get_capabilities()
def configure(self, frequency):
if self.is_streaming():
raise RuntimeError("Cannot configure radio while a stream is active")
frequency = int(prefixed.Float(frequency))
sample_rate = 384000
bandwidth = 200000
sample_rates = preferred_sample_rates(self.capabilities['rx']['sample-rates'])
if len(sample_rates) == 0:
raise RuntimeError("No suitable sample rates are available")
self.sample_rate, self.output_rate = sample_rates[0]
self.device.setFrequency(soapy.SOAPY_SDR_RX, 0, frequency)
self.device.setSampleRate(soapy.SOAPY_SDR_RX, 0, sample_rate)
self.device.setSampleRate(soapy.SOAPY_SDR_RX, 0, self.sample_rate)
self.device.setBandwidth(soapy.SOAPY_SDR_RX, 0, bandwidth)
# Set automatic gain
@ -45,8 +52,39 @@ class Radio:
'gain-mode': 'auto' if self.device.getGainMode(soapy.SOAPY_SDR_RX, 0) else 'manual',
}
def get_info(self):
return {
'name': self.name,
'device': self.device_info,
'capabilities': self.capabilities,
'stream-path': self._stream_path(),
'streaming': self.is_streaming(),
}
def _get_capabilities(self):
def get_direction_capabilities(direction):
return {
'antennas': self.device.listAntennas(direction, 0),
'gains': self.device.listGains(direction, 0),
'frequencies': self.device.listFrequencies(direction, 0),
'sample-rates': self.device.listSampleRates(direction, 0),
'bandwidths': self.device.listBandwidths(direction, 0),
'sensors': self.device.listSensors(direction, 0),
'formats': self.device.getStreamFormats(direction, 0),
}
return {
'rx': get_direction_capabilities(soapy.SOAPY_SDR_RX),
'tx': get_direction_capabilities(soapy.SOAPY_SDR_TX),
'clock-sources': self.device.listClockSources(),
'time-sources': self.device.listTimeSources(),
'register-interfaces': self.device.listRegisterInterfaces(),
'gpios': self.device.listGPIOBanks(),
'uarts': self.device.listUARTs(),
}
def is_streaming(self):
return (self.thread and self.thread.is_alive())
return True if (self.thread and self.thread.is_alive()) else False
def start_stream(self):
if self.is_streaming():
@ -82,8 +120,10 @@ class Radio:
continue
elif result.ret < 0:
error = SoapyError(result.ret)
if error is not SoapyError.Timeout:
print("Stream read failed, aborting stream:", error, file=sys.stderr)
break
continue
else:
read_size = int(result.ret * 2)
self.demod.stdin.write(
@ -96,8 +136,8 @@ class Radio:
def _init_stream(self):
self.playback = subprocess.Popen(
[
'/usr/bin/ffmpeg', '-f', 's16le', '-ar', '32000', '-ac', '2', '-i', '-',
'-f', 'rtsp', f"rtsp://localhost:{Radio.PORT}/radio/{self.name}"
'/usr/bin/ffmpeg', '-f', 's16le', '-ar', str(self.output_rate), '-ac', '2', '-i', '-',
'-f', 'rtsp', f"rtsp://localhost:{self._stream_path()}"
],
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
@ -105,7 +145,7 @@ class Radio:
)
self.demod = subprocess.Popen(
['/usr/bin/python3', 'fm_demod.py', '-f', 'CS16', '-s', '384k', '-d', '32k'],
['/usr/bin/python3', 'fm_demod.py', '-f', 'CS16', '-s', str(self.sample_rate), '-d', str(self.output_rate)],
stdin=subprocess.PIPE,
stdout=self.playback.stdin,
stderr=subprocess.DEVNULL
@ -113,7 +153,10 @@ class Radio:
self.buffer = np.array([0] * Radio.SAMPLES * 2, FORMATS[Radio.FORMAT].numpy)
self.stream = self.device.setupStream(soapy.SOAPY_SDR_RX, FORMATS[Radio.FORMAT].soapy)
self.device.activateStream(self.stream)
result = self.device.activateStream(self.stream)
if result != 0:
raise RuntimeError(f"Error activating stream: {result}")
def _cleanup_stream(self):
self.run = False
@ -133,6 +176,9 @@ class Radio:
self.playback.wait()
self.playback = None
def _stream_path(self):
return f"{Radio.PORT}/radio/{self.name}"
"""
Quick and dirty test of the Radio class.

119
samplerates.py Normal file
View File

@ -0,0 +1,119 @@
from pprint import pprint
"""
Not all output rates are going to be supported by the system sound card.
Choosing from a limited subset of preferred output rates is how we guarantee
that the system will be able to play the output audio.
Sample rates stolen from:
https://en.wikipedia.org/wiki/Sampling_(signal_processing)
"""
supported_ouput_rates = [
8000, # Telephone, P25 audio (sufficient for speech, some consonants unintelligble)
16000, # Modern telephone/VoIP, good quality speech
32000, # FM radio
44100, # CD audio quality
48000,
50000, # Uncommon but supported
88200,
96000, # DVD/Blu-ray audio
192000, # Too much.
]
def score(pair, target_output=32000, target_ratio=10):
"""
Heuristic for scoring input & output sample rates. The criteria are:
- closest to 44.1 kHz is better;
- closer to a ratio of 10 is better,
- additionally penalising ratios lower than 10.
Lower scores are better.
This should result in selected sample rates that give good audio quality
without wasting CPU cycles on a needlessly high input rate.
"""
# Give the worst score possible for impossible pairs.
if pair[0] % pair[1] != 0:
return float("inf")
ratio = pair[0] // pair[1]
return abs(pair[1] - target_output)/2500 \
+ max(0, target_output - pair[1])/2500 \
+ abs(ratio - target_ratio)**0.8 \
+ max(0, target_ratio - ratio)**2
def flatten(l):
return [item for sublist in l for item in sublist]
def flatten_dict(d):
return [(key,value) for key,rates in d.items() for value in rates]
def get_pairs(input_rate):
return [
(input_rate, rate)
for rate in supported_ouput_rates
if (input_rate % rate == 0)
]
def supported_sample_rates(supported_input_rates):
return {
in_rate: [out_rate for out_rate in supported_ouput_rates if in_rate % out_rate == 0]
for in_rate in supported_input_rates
}
def preferred_sample_rates(supported_input_rates):
return sorted(flatten_dict(supported_sample_rates(supported_input_rates)), key=score)
if __name__ == '__main__':
nesdr_sample_rates = [
250000,
1024000,
1536000,
1792000,
1920000,
2048000,
2160000,
2560000,
2880000,
3200000
]
sdrplay_sample_rates = [
62500,
96000,
125000,
192000,
250000,
384000,
500000,
768000,
1000000,
2000000,
2048000,
3000000,
4000000,
5000000,
6000000,
7000000,
8000000,
9000000,
10000000
]
print('nesdr')
pprint(preferred_sample_rates(nesdr_sample_rates))
#for rate in nesdr_sample_rates:
# print(rate, ' \t', get_pairs(rate))
#supported = flatten([get_pairs(rate) for rate in nesdr_sample_rates])
#supported.sort(key=score)
#pprint(supported)
print('sdrplay')
pprint(preferred_sample_rates(sdrplay_sample_rates))
#for rate in sdrplay_sample_rates:
# print(rate, ' \t', get_pairs(rate))
#supported = flatten([get_pairs(rate) for rate in sdrplay_sample_rates])
#supported.sort(key=score)
#pprint(supported)