VE7UWU/sdrangel
1
0
Fork 0
mirror of https://github.com/f4exb/sdrangel.git synced 2024-12-22 01:20:56 -05:00
Code Issues Projects Releases Wiki Activity

scriptsapi: removed obsolete ptt_active.py and created ptt_feature.py based on PTT amd LimeRFE features

Browse Source
This commit is contained in:
f4exb 2022-08-17 11:42:26 +02:00
parent 00c98fe455
commit 52e20382c0
3 changed files with 120 additions and 344 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
scriptsapi/Readme.md
View File

@ -32,31 +32,19 @@ Normal sequence of operations:
- In SDRangel connect the Frequency Tracker plugin by clicking on the grey square at the left of the top bar of the Frequency Tracker GUI. It opens the channel settings dialog. Check the 'Reverse API' box. Next to this box is the address and port at which the channel will be connected. If you use the defaults for `freqtracking.py` you may leave it as it is else you have to adjust it to the address and port of `freqtracking.py` (options `-A` and `-P`).
- In the same manner connect the channel you want to be controlled by `freqtracking.py`. You may connect any number of channels like this. When a channel is removed `freqtracking.py` will automatically remove it from its list at the first attempt to synchronize that will fail.
<h2>ptt_active.py</h2>
<h2>ptt_feature.py</h2>
PTT (Push To Talk) actively listening system. For a pair of given device set indexes it actively listens to start and stop commands on the corresponding devices to switch over to the other
Control a PTT feature and optionally a LimeRFE feature in coordination.
Options are:
- `-h` or `--help` show help message and exit
- `-A` or `--address` listening IP address. Default `0.0.0.0` (all interfaces)
- `-P` or `--port` listening port. Default `8000`
- `-p` or `--port-sdr` SDRangel instance REST API listening port. Default `8091`
- `-l` or `--link` Pair of indexes of the device sets to link. Default `0 1`
- `-d` or `--delay` Switch over delay in seconds. Default `1`
- `-f` or `--freq-sync` Synchronize devices center frequencies
- `-a` or `--address` address and port of SDRangel instance API. Default: 127.0.0.1:8091
- `-p` or `--ptt-index` PTT feature index. Mandatory
- `-l` or `--limerfe-index` LimeRFE feature index. Optional if present synchronizes the PTT and LimeRFE features
- `-t` or `--transmit` Switch to transmission if not present switches to reception
Normal sequence of operations:
In this example we have a Rx device on index 0 and a Tx device on index 1. All settings are assumed to be the default settings.
- Start `ptt_active.py` in a terminal
- On the Rx device right click on the start/stop button and activate reverse API at address `127.0.0.1` port `8000` (default)
- On the Tx device right click on the start/stop button and activate reverse API at address `127.0.0.1` port `8000` (default)
- Start the Rx or Tx device
- Stop the running device (Rx or Tx) this will switch over automatically to the other
Important: you should initiate switch over by stopping the active device and not by starting the other.
In order for it to work properly you should start your SDRangel instance in reception mode with Rx and PTT feature running and set to control the desired Rx and Tx devices.
<h2>qo100_datv.py</h2>
@ -80,7 +68,6 @@ Dumps an instance setup to a sequence of commands in a JSON file that can be use
- Deviceset main spectrum settings
- Deviceset channels instantiations
- Deviceset channels settings
- Featureset instantiations
- Featureset features instantiations
- Featureset features settings
@ -102,7 +89,6 @@ Options are:
- `-a` or `--address` address and port of SDRangel instance. Default is `127.0.0.1:8091`
- `-j` or `--json-file` JSON file containing description of API commands
- `-i` or `--init` Initialize instance before running script. This is useful when running a sequence of initialization commands
- `-1` or `--ignore-first-posts`. Ignore first deviceset or featureset post in sequence. This is useful when running a sequence of initialization commands for a GUI instance
Each command in the JSON file is a JSON document with the following keys:

323
scriptsapi/ptt_active.py
View File

@ -1,323 +0,0 @@
#!/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()

113
scriptsapi/ptt_feature.py Executable file
View File

@ -0,0 +1,113 @@
#!/usr/bin/env python3
"""
PTT feature control
Switches a PTT Feature on/off with optional link to a LimeRFE feature
"""
import sys
import traceback
import requests
import time
from optparse import OptionParser
base_url = "http://127.0.0.1:8091/sdrangel"
# ======================================================================
def getInputOptions():
parser = OptionParser(usage="usage: %%prog [-t]\n")
parser.add_option("-a", "--address", dest="address", help="Address and port. Default: 127.0.0.1:8091", metavar="ADDRESS", type="string")
parser.add_option("-p", "--ptt-index", dest="ptt_index", help="Index of PTT feture", metavar="INTEGER", type="int")
parser.add_option("-l", "--limerfe-index", dest="limerfe_index", help="Index of LimeRFE feture", metavar="INTEGER", type="int")
parser.add_option("-t", "--transmit", dest="transmit", help="PTT on else off", action="store_true")
(options, args) = parser.parse_args()
if (options.address == None):
options.address = "127.0.0.1:8091"
return options
# ======================================================================
def turn_ptt(options):
if options.limerfe_index is not None:
limerfe_url = f"{base_url}/featureset/feature/{options.limerfe_index}/actions"
if options.transmit:
limerfe_payload_rx = {
"featureType": "LimeRFE",
"LimeRFEActions": {
"setRx": 0
}
}
r = requests.post(url=limerfe_url, json=limerfe_payload_rx) # switch off Rx
print(f"LimeRFE switch off Rx {r.status_code}")
else:
limerfe_payload_tx = {
"featureType": "LimeRFE",
"LimeRFEActions": {
"setTx": 0
}
}
r = requests.post(url=limerfe_url, json=limerfe_payload_tx) # switch off Tx
print(f"LimeRFE switch off Tx {r.status_code}")
time.sleep(0.2)
ptt_payload = {
"featureType": "SimplePTT",
"SimplePTTActions": {
"ptt": 1 if options.transmit else 0
}
}
ptt_url = f"{base_url}/featureset/feature/{options.ptt_index}/actions"
r = requests.post(url=ptt_url, json=ptt_payload)
print(f"SimplePTT post action {r.status_code}")
if options.limerfe_index is not None:
time.sleep(0.2)
limerfe_url = f"{base_url}/featureset/feature/{options.limerfe_index}/actions"
if options.transmit:
limerfe_payload_tx = {
"featureType": "LimeRFE",
"LimeRFEActions": {
"setTx": 1
}
}
r = requests.post(url=limerfe_url, json=limerfe_payload_tx) # switch on Tx
print(f"LimeRFE switch on Tx {r.status_code}")
else:
limerfe_payload_rx = {
"featureType": "LimeRFE",
"LimeRFEActions": {
"setRx": 1
}
}
for i in range(5):
r = requests.post(url=limerfe_url, json=limerfe_payload_rx) # switch on Rx
print(f"LimeRFE switch on Rx {r.status_code}")
if r.status_code // 100 == 2:
break
time.sleep(0.5)
# ======================================================================
def main():
try:
options = getInputOptions()
global base_url
base_url = "http://%s/sdrangel" % options.address
if options.ptt_index is None:
print("You must at least give the PTT feature index (-p option)")
sys.exit(1)
turn_ptt(options)
except Exception as ex:
tb = traceback.format_exc()
print(tb, file=sys.stderr)
# ======================================================================
if __name__ == "__main__":
main()