VE7UWU/sdrangel
1
0
Fork 0
mirror of https://github.com/f4exb/sdrangel.git synced 2025-10-29 12:00:25 -04:00
Code Issues Projects Releases Wiki Activity
sdrangel/scriptsapi/ptt_active.py

324 lines
15 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
''' Active PTT
Handles the switchover between two arbitrary device sets
- Both device sets should have the reverse API feature set with the address and port of this server
- Once in place and you have started one of the devices you should only stop one or the other never start
There are two reasons for this:
- This module reacts on an action already taken so if you start Tx then the Rx is not stopped immediately
and damage to the Rx could occur. If you start with a stop action you cannot get in this situation.
- For half duplex devices (only the HackRF) it will lock Tx or Rx. You can always recover the situation
by stopping the running side.
- There is no assumption on the Rx or Tx nature you may as well switchover 2 Rx or 2 Tx
- Both devices have not to belong to the same physical device necessarily. You could mix a RTL-SDR Rx and a
HackRF Tx for example
- It can pilot a LimeRFE device via USB through SDRangel API (by giving TTY device with --limerfe-dev parameter)
and with the assumption that the first device set index given in the --link parameter is the Rx and the second
is the Tx
'''
import requests
import time
import argparse
from flask import Flask
from flask import request, jsonify
SDRANGEL_API_PORT = 8091
START_COUNT = 0
STOP_COUNT = 0
OTHER_DICT = {}
RUNNING = set()
DELAY = 1
FREQ_SYNC = False
FREQ_HAS_CHANGED = False
LIMERFE_DEVICE = None
LIMERFE_RX = False
LIMERFE_TX = False
app = Flask(__name__)
# ======================================================================
def start_device(device_index, sdrangel_ip, sdrangel_port):
""" Start the specified device """
# ----------------------------------------------------------------------
base_url = f'http://{sdrangel_ip}:{sdrangel_port}/sdrangel'
dev_run_url = base_url + f'/deviceset/{device_index}/device/run'
r = requests.get(url=dev_run_url)
if r.status_code // 100 == 2:
rj = r.json()
state = rj.get("state", None)
if state is not None:
if state == "idle":
r = requests.post(url=dev_run_url)
if r.status_code == 200:
print(f'start_device: Device {device_index} started')
else:
print(f'start_device: Error starting device {device_index}')
if LIMERFE_DEVICE:
limerfe_switch(device_index, True, sdrangel_ip, sdrangel_port)
else:
print(f'start_device: Device {device_index} not in idle state')
else:
print(f'start_device: Cannot get device {device_index} running state')
else:
print(f'start_device: Error {r.status_code} getting device {device_index} running state')
# ======================================================================
def stop_device(device_index, sdrangel_ip, sdrangel_port):
""" Stop the specified device """
# ----------------------------------------------------------------------
base_url = f'http://{sdrangel_ip}:{sdrangel_port}/sdrangel'
dev_run_url = base_url + f'/deviceset/{device_index}/device/run'
r = requests.get(url=dev_run_url)
if r.status_code // 100 == 2:
rj = r.json()
state = rj.get("state", None)
if state is not None:
if state == "running":
r = requests.delete(url=dev_run_url)
if r.status_code == 200:
print(f'stop_device: Device {device_index} stopped')
else:
print(f'stop_device: Error stopping device {device_index}')
if LIMERFE_DEVICE:
limerfe_switch(device_index, False, sdrangel_ip, sdrangel_port)
else:
print(f'stop_device: Device {device_index} not in running state')
else:
print(f'stop_device: Cannot get device {device_index} running state')
else:
print(f'stop_device: Error {r.status_code} getting device {device_index} running state')
# ======================================================================
def set_focus(device_index, sdrangel_ip, sdrangel_port):
""" Set focus on the specified device """
# ----------------------------------------------------------------------
base_url = f'http://{sdrangel_ip}:{sdrangel_port}/sdrangel'
dev_focus_url = base_url + f'/deviceset/{device_index}/focus'
r = requests.patch(url=dev_focus_url)
if r.status_code // 100 == 2:
print(f'set_focus: Focus set on device set {device_index}')
elif r.status_code == 400:
print(f'set_focus: Focus on device set is not supported in a server instance')
else:
print(f'set_focus: Error {r.status_code} setting focus on device set {device_index}')
# ======================================================================
def get_sdrangel_ip(request):
""" Extract originator address from request """
# ----------------------------------------------------------------------
if request.environ.get('HTTP_X_FORWARDED_FOR') is None:
return request.environ['REMOTE_ADDR']
else:
return request.environ['HTTP_X_FORWARDED_FOR']
# ======================================================================
def get_center_frequency(content):
""" Look for center frequency recursively """
# ----------------------------------------------------------------------
for k in content:
if isinstance(content[k], dict):
return get_center_frequency(content[k])
elif k == "centerFrequency":
return content[k]
# ======================================================================
def change_center_frequency(content, new_frequency):
""" Change center frequency searching recursively """
# ----------------------------------------------------------------------
for k in content:
if isinstance(content[k], dict):
change_center_frequency(content[k], new_frequency)
elif k == "centerFrequency":
content[k] = new_frequency
# ======================================================================
def set_center_frequency(new_frequency, device_index, sdrangel_ip, sdrangel_port):
""" Set a new center frequency for given device """
# ----------------------------------------------------------------------
base_url = f'http://{sdrangel_ip}:{sdrangel_port}/sdrangel'
r = requests.get(url=base_url + f'/deviceset/{device_index}/device/settings')
if r.status_code // 100 == 2:
rj = r.json()
frequency = get_center_frequency(rj)
if new_frequency != frequency:
change_center_frequency(rj, new_frequency)
r = requests.patch(url=base_url + f'/deviceset/{device_index}/device/settings', json=rj)
if r.status_code / 100 == 2:
print(f'set_center_frequency: changed center frequency of device {device_index} to {new_frequency}')
global FREQ_HAS_CHANGED
FREQ_HAS_CHANGED = True
return jsonify(rj)
else:
print(f'set_center_frequency: failed to change center frequency of device {device_index} with error {r.status_code}')
else:
print(f'set_center_frequency: frequency of device {device_index} is unchanged')
else:
print(f'set_center_frequency: error {r.status_code} getting settings for device {device_index}')
return ""
# ======================================================================
def limerfe_switch(originator_index, start, sdrangel_ip, sdrangel_port):
""" Start or stop the LimeRFE device connected with first linked originator on Rx and second on Tx.
the start parameter is True to start and False to stop
"""
# ----------------------------------------------------------------------
endpoint_url = f'http://{sdrangel_ip}:{sdrangel_port}/sdrangel/limerfe/run'
try:
if OTHER_DICT.keys().index(originator_index) == 0: # Rx
global LIMERFE_RX
LIMERFE_RX = start
else:
global LIMERFE_TX
LIMERFE_TX = start
except ValueError:
print(f'Invalid device index {originator_index}')
return
payload = {
'devicePath': LIMERFE_DEVICE,
'rxOn': 1 if LIMERFE_RX else 0,
'txOn': 1 if LIMERFE_TX else 0,
}
r = requests.put(url=endpoint_url, json=payload)
if r.status_code / 100 == 2:
print(f'LimeRFE at {LIMERFE_DEVICE} switched with Rx: {LIMERFE_RX} Tx: {LIMERFE_TX}')
else:
print(f'failed to switch LimeRFE at {LIMERFE_DEVICE} with Rx: {LIMERFE_RX} Tx: {LIMERFE_TX}')
# ======================================================================
@app.route('/sdrangel')
def hello_sdrangel():
""" Just to test if it works """
# ----------------------------------------------------------------------
return 'Hello, SDRangel!'
# ======================================================================
@app.route('/sdrangel/deviceset/<int:deviceset_index>/device/run', methods=['GET', 'POST', 'DELETE'])
def device_run(deviceset_index):
''' Reply with the expected reply of a working device '''
# ----------------------------------------------------------------------
originator_index = None
direction = None
content = request.get_json(silent=True)
if content:
originator_index = content.get('originatorIndex')
direction = content.get('direction')
if originator_index is None or direction is None:
print('device_run: SDRangel reverse API v4.5.2 or higher required. No or invalid originator information')
return ""
sdrangel_ip = get_sdrangel_ip(request)
other_device_index = OTHER_DICT.get(originator_index)
if other_device_index is None:
print('device_run: Device {originator_index} is not part of the linked pair. Aborting request.')
return ""
global RUNNING
print(f'device_run: Device: {originator_index} Other device: {other_device_index} Running: {RUNNING}')
if request.method == 'POST':
print(f'device_run: Device {originator_index} (direction={direction}) has started at {sdrangel_ip}:{SDRANGEL_API_PORT}')
if originator_index not in RUNNING:
RUNNING.add(originator_index)
if other_device_index in RUNNING:
time.sleep(DELAY)
stop_device(other_device_index, sdrangel_ip, SDRANGEL_API_PORT)
RUNNING.remove(other_device_index)
else:
print(f'device_run: Device {other_device_index} is stopped already')
reply = { "state": "idle" }
return jsonify(reply)
elif request.method == 'DELETE':
print(f'device_run: Device {originator_index} (direction={direction}) has stopped at {sdrangel_ip}:{SDRANGEL_API_PORT}')
if originator_index in RUNNING:
RUNNING.remove(originator_index)
if other_device_index not in RUNNING:
time.sleep(DELAY)
start_device(other_device_index, sdrangel_ip, SDRANGEL_API_PORT)
set_focus(other_device_index, sdrangel_ip, SDRANGEL_API_PORT)
RUNNING.add(other_device_index)
else:
print(f'device_run: Device {other_device_index} is running already')
reply = { "state": "running" }
return jsonify(reply)
elif request.method == 'GET':
return f'RUN device {deviceset_index}'
# ======================================================================
@app.route('/sdrangel/deviceset/<int:deviceset_index>/device/settings', methods=['GET', 'PATCH', 'PUT'])
def device_settings(deviceset_index):
''' Reply with the expected reply of a working device '''
# ----------------------------------------------------------------------
originator_index = None
content = request.get_json(silent=True)
if content:
originator_index = content.get('originatorIndex')
if originator_index is None:
print('device_settings: SDRangel reverse API v4.5.2 or higher required. No or invalid originator information')
return ""
sdrangel_ip = get_sdrangel_ip(request)
other_device_index = OTHER_DICT.get(originator_index)
if other_device_index is None:
print('device_settings: Device {originator_index} is not part of the linked pair. Aborting request.')
return ""
new_frequency = get_center_frequency(content)
if new_frequency and FREQ_SYNC:
global FREQ_HAS_CHANGED
if FREQ_HAS_CHANGED:
FREQ_HAS_CHANGED = False
print('device_settings: frequency was just changed. Ignoring this change.')
else:
set_center_frequency(new_frequency, other_device_index, sdrangel_ip, SDRANGEL_API_PORT)
print(f'device_settings: Device {originator_index} changed frequency to {new_frequency}')
return ""
# ======================================================================
def getInputOptions():
""" This is the argument line parser """
# ----------------------------------------------------------------------
parser = argparse.ArgumentParser(description="Manages PTT from an SDRangel instance automatically")
parser.add_argument("-A", "--address", dest="addr", help="listening address", metavar="IP", type=str)
parser.add_argument("-P", "--port", dest="port", help="listening port", metavar="PORT", type=int)
parser.add_argument("-p", "--port-sdr", dest="sdrangel_port", help="SDRangel REST API port", metavar="PORT", type=int)
parser.add_argument("-l", "--link", dest="linked_devices", help="pair of indexes of devices to link", metavar="LIST", type=int, nargs=2)
parser.add_argument("-d", "--delay", dest="delay", help="switchover delay in seconds", metavar="SECONDS", type=int)
parser.add_argument("-f", "--freq-sync", dest="freq_sync", help="synchronize linked devices frequencies", action="store_true")
parser.add_argument("-L", "--limerfe-dev", dest="limerfe_dev", help="LimeRFE USB serial device (optional)", metavar="DEVICE", type=str)
options = parser.parse_args()
if options.addr == None:
options.addr = "0.0.0.0"
if options.port == None:
options.port = 8000
if options.sdrangel_port == None:
options.sdrangel_port = 8091
if options.linked_devices == None:
options.linked_devices = [0, 1]
if options.delay == None:
options.delay = 1
if options.freq_sync == None:
options.freq_sync = False
other_dict = {
options.linked_devices[0]: options.linked_devices[1],
options.linked_devices[1]: options.linked_devices[0]
}
return options.addr, options.port, options.sdrangel_port, options.delay, options.freq_sync, other_dict, options.limerfe_dev
# ======================================================================
def main():
""" This is the main routine """
# ----------------------------------------------------------------------
global SDRANGEL_API_PORT
global OTHER_DICT
global DELAY
global FREQ_SYNC
global LIMERFE_DEVICE
addr, port, SDRANGEL_API_PORT, DELAY, FREQ_SYNC, OTHER_DICT, LIMERFE_DEVICE = getInputOptions()
print(f'main: starting: SDRangel port: {SDRANGEL_API_PORT} links: {OTHER_DICT} freq sync: {FREQ_SYNC} LimeRFE: {LIMERFE_DEVICE}')
app.run(debug=True, host=addr, port=port)
# ======================================================================
if __name__ == "__main__":
""" When called from command line... """
# ----------------------------------------------------------------------
main()
Reference in New Issue View Git Blame Copy Permalink