109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
import os
|
|
import subprocess
|
|
from functools import reduce
|
|
from operator import and_
|
|
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class UnknownFeatureException(Exception):
|
|
pass
|
|
|
|
class FeatureDetector(object):
|
|
features = {
|
|
"core": [ "csdr", "nmux", "nc" ],
|
|
"rtl_sdr": [ "rtl_sdr" ],
|
|
"sdrplay": [ "rx_tools" ],
|
|
"hackrf": [ "hackrf_transfer" ],
|
|
"airspy": [ "airspy_rx" ],
|
|
"digital_voice_digiham": [ "digiham", "sox" ],
|
|
"digital_voice_dsd": [ "dsd", "sox" ]
|
|
}
|
|
|
|
def feature_availability(self):
|
|
return {name: self.is_available(name) for name in FeatureDetector.features}
|
|
|
|
def is_available(self, feature):
|
|
return self.has_requirements(self.get_requirements(feature))
|
|
|
|
def get_requirements(self, feature):
|
|
try:
|
|
return FeatureDetector.features[feature]
|
|
except KeyError:
|
|
raise UnknownFeatureException("Feature \"{0}\" is not known.".format(feature))
|
|
|
|
def has_requirements(self, requirements):
|
|
passed = True
|
|
for requirement in requirements:
|
|
methodname = "has_" + requirement
|
|
if hasattr(self, methodname) and callable(getattr(self, methodname)):
|
|
passed = passed and getattr(self, methodname)()
|
|
else:
|
|
logger.error("detection of requirement {0} not implement. please fix in code!".format(requirement))
|
|
return passed
|
|
|
|
def command_is_runnable(self, command):
|
|
return os.system("{0} 2>/dev/null >/dev/null".format(command)) != 32512
|
|
|
|
def has_csdr(self):
|
|
return self.command_is_runnable("csdr")
|
|
|
|
def has_nmux(self):
|
|
return self.command_is_runnable("nmux --help")
|
|
|
|
def has_nc(self):
|
|
return self.command_is_runnable('nc --help')
|
|
|
|
def has_rtl_sdr(self):
|
|
return self.command_is_runnable("rtl_sdr --help")
|
|
|
|
def has_rx_tools(self):
|
|
return self.command_is_runnable("rx_sdr --help")
|
|
|
|
"""
|
|
To use a HackRF, compile the HackRF host tools from its "stdout" branch:
|
|
git clone https://github.com/mossmann/hackrf/
|
|
cd hackrf
|
|
git fetch
|
|
git checkout origin/stdout
|
|
cd host
|
|
mkdir build
|
|
cd build
|
|
cmake .. -DINSTALL_UDEV_RULES=ON
|
|
make
|
|
sudo make install
|
|
"""
|
|
def has_hackrf_transfer(self):
|
|
# TODO i don't have a hackrf, so somebody doublecheck this.
|
|
# TODO also check if it has the stdout feature
|
|
return self.command_is_runnable("hackrf_transfer --help")
|
|
|
|
def command_exists(self, command):
|
|
return os.system("which {0}".format(command)) == 0
|
|
|
|
def has_digiham(self):
|
|
# the digiham tools expect to be fed via stdin, they will block until their stdin is closed.
|
|
def check_with_stdin(command):
|
|
try:
|
|
process = subprocess.Popen(command, stdin=subprocess.PIPE)
|
|
process.communicate("")
|
|
return process.wait() == 0
|
|
except FileNotFoundError:
|
|
return False
|
|
return reduce(and_,
|
|
map(
|
|
check_with_stdin,
|
|
["rrc_filter", "ysf_decoder", "dmr_decoder", "mbe_synthesizer", "gfsk_demodulator"]
|
|
),
|
|
True)
|
|
|
|
def has_dsd(self):
|
|
return self.command_is_runnable("dsd")
|
|
|
|
def has_sox(self):
|
|
return self.command_is_runnable("sox")
|
|
|
|
def has_airspy_rx(self):
|
|
return self.command_is_runnable("airspy_rx --help 2> /dev/null")
|