##// END OF EJS Templates
radarsys_met
RSS
Experiment detail view update
Experiment detail view update
amorales - - Load All Authors

File last commit:

r360:46316742ce9f
r365:24b24259841b
Show More
models.py
379 lines | 10.9 KiB | text/x-python | PythonLexer
/ apps / usrp_tx / models.py
History | Annotation | Raw |Copy content |Copy permalink
import ast
import json
import requests
import base64
from struct import pack
from django.db import models
from django.urls import reverse
from django.core.validators import MinValueValidator, MaxValueValidator
from apps.main.models import Configuration
BOARD_VALUE = (
('1', 'A:AB'),
('2', 'AB')
)
ANT_VALUE = (
('1', 'RX'),
('2', 'TX'),
('3', 'TX/RX')
)
TYPE_VALUE = (
('0', 'None'),
('1', 'User define'),
('2', 'Barker 2'),
('3', 'Barker 3'),
('4', 'Barker 4'),
('5', 'Barker 5'),
('7', 'Barker 7'),
('11', 'Barker 11'),
('13', 'Barker 13')
)
class USRPTXConfiguration(Configuration):
ip_address = models.GenericIPAddressField(
verbose_name = 'IP address',
protocol='IPv4',
default='0.0.0.0')
daughterboard = models.CharField(
verbose_name='Daughterboard',
max_length=3,
choices=BOARD_VALUE,
null=False,
blank=False
)
antenna = models.CharField(
verbose_name='Antenna',
max_length=3,
choices=ANT_VALUE,
null=False,
blank=False
)
frequency = models.FloatField(
verbose_name='Frequency',
blank=False,
null=False,
help_text='Introduce the value in MHz (10^6)'
)
samplerate = models.FloatField(
verbose_name='Sample rate',
blank=False,
null=False,
help_text='Introduce the value in MHz (10^6)'
)
ipp = models.FloatField(
verbose_name='IPP',
blank=False,
null=False,
help_text='Introduce the value in km'
)
enable_1 = models.BooleanField(
verbose_name='Pulse 1',
default=True,
blank=False,
null=False
)
pulse_1 = models.FloatField(
verbose_name='Width (Pulse 1)',
blank=False,
null=False,
help_text='Introduce the value in %'
)
delay_1 = models.IntegerField(
verbose_name='Delay (Pulse 1)',
default=0,
blank=True,
null=False,
help_text='Introduce the value in µs'
)
type_1 = models.CharField(
verbose_name='Type (Pulse 1)',
default = 0,
max_length=3,
choices=TYPE_VALUE,
blank=False,
null=False
)
code_1 = models.IntegerField(
verbose_name="Code (Pulse 1)",
blank=True,
null=True,
help_text="Introduce the binary code"
)
repetitions_1 = models.IntegerField(
verbose_name='N° of repetitions (Pulse 1)',
default=1,
blank=True,
null=True
)
enable_2 = models.BooleanField(
verbose_name='Pulse 2',
blank=False,
null=False
)
pulse_2 = models.FloatField(
verbose_name='Width (Pulse 2)',
default=1,
blank=True,
null=True,
help_text='Introduce the value in %'
)
delay_2 = models.IntegerField(
verbose_name='Delay (Pulse 2)',
default=0,
blank=True,
null=True,
help_text='Introduce the value in µs'
)
type_2 = models.CharField(
verbose_name='Type (Pulse 2)',
default = 0,
max_length=3,
choices=TYPE_VALUE,
blank=True,
null=False
)
code_2 = models.IntegerField(
verbose_name="Code (Pulse 2)",
blank=True,
null=True,
help_text="Introduce the binary code"
)
repetitions_2 = models.IntegerField(
verbose_name='N° of repetitions (Pulse 2)',
default=1,
blank=True,
null=True
)
class Meta:
db_table = 'usrp_tx_configurations'
def __str__(self):
return str(self.label)
def get_absolute_url_plot(self):
return reverse('url_plot_usrp_tx_pulses', args=[str(self.id)])
def request(self, cmd, method='get', **kwargs):
req = getattr(requests, method)(self.device.url(cmd), **kwargs)
payload = req.json()
return payload
def status_device(self):
try:
#self.device.status = 0
#payload = self.request('status')
payload = requests.get(self.device.url())
print(payload)
if payload:
self.device.status = 1
elif payload['status']=='disable':
self.device.status = 2
else:
self.device.status = 1
self.device.save()
self.message = 'USRP Tx status: {}'.format(payload['status'])
return False
except Exception as e:
if 'No route to host' not in str(e):
self.device.status = 4
self.device.save()
self.message = 'USRP Tx status: {}'.format(str(e))
return False
self.device.save()
return True
def reset_device(self):
try:
payload = self.request('reset', 'post')
if payload['reset']=='ok':
self.message = 'USRP Tx restarted OK'
self.device.status = 2
self.device.save()
else:
self.message = 'USRP Tx restart fail'
self.device.status = 4
self.device.save()
except Exception as e:
self.message = 'USRP Tx reset: {}'.format(str(e))
return False
return True
def stop_device(self):
try:
command = self.device.url() + "stoptx"
r = requests.get(command)
if r:
self.device.status = 4
self.device.save()
self.message = 'USRP stopped'
else:
self.device.status = 4
self.device.save()
return False
except Exception as e:
if 'No route to host' not in str(e):
self.device.status = 4
else:
self.device.status = 0
#self.message = 'USRP Tx stop: {}'.format(str(e))
self.message = "USRP Tx can't start, please check network/device connection or IP address/port configuration"
self.device.save()
return False
return True
def start_device(self):
print("Entró al start")
try:
usrp = USRPTXConfiguration.objects.get(pk=self)
usrp_daughterboard = usrp.get_daughterboard_display()
usrp_antenna = usrp.get_antenna_display()
print(usrp)
payload = {'ip_address': usrp.ip_address, 'daughterboard': usrp_daughterboard, 'antenna': usrp_antenna,
'frequency': usrp.frequency, 'sample_rate': usrp.samplerate, 'ipp': usrp.ipp, 'enable_1': usrp.enable_1,
'pulse_1': usrp.pulse_1, 'delay_1': usrp.delay_1, 'type_1': usrp.type_1, 'code_1': usrp.code_1,
'repetitions_1': usrp.repetitions_1, 'enable_2': usrp.enable_2, 'pulse_2': usrp.pulse_2, 'delay_2': usrp.delay_2,
'type_2': usrp.type_2, 'code_2': usrp.code_2, 'repetitions_2': usrp.repetitions_2}
print(payload)
r = requests.post(self.device.url("usrptx"), json=payload)
print(r.text)
#payload = self.request('usrp', 'post', data=json.dumps(data))
#print(payload)
if r:
self.device.status = 3
self.device.save()
self.message = 'USRP Tx configured and started'
else:
return False
except Exception as e:
if 'No route to host' not in str(e):
self.device.status = 4
else:
self.device.status = 0
#self.message = 'USRP Tx start: {}'.format(str(e))
self.message = "USRP Tx can't start, please check network/device connection or IP address/port configuration"
self.device.save()
return False
return True
#def write_device(self, raw=False):
if not raw:
clock = RCClock.objects.get(rc_configuration=self)
print(clock)
if clock.mode:
data = {'default': clock.frequency}
else:
data = {'manual': [clock.multiplier, clock.divisor, clock.reference]}
payload = self.request('setfreq', 'post', data=json.dumps(data))
print(payload)
if payload['command'] != 'ok':
self.message = 'USRP Tx write: {}'.format(payload['command'])
else:
self.message = payload['programming']
if payload['programming'] == 'fail':
self.message = 'USRP Tx write: error programming CGS chip'
values = []
for pulse, delay in zip(self.get_pulses(), self.get_delays()):
while delay>65536:
values.append((pulse, 65535))
delay -= 65536
values.append((pulse, delay-1))
data = bytearray()
#reset
data.extend((128, 0))
#disable
data.extend((129, 0))
#SW switch
if self.control_sw:
data.extend((130, 2))
else:
data.extend((130, 0))
#divider
data.extend((131, self.clock_divider-1))
#enable writing
data.extend((139, 62))
last = 0
for tup in values:
vals = pack('<HH', last^tup[0], tup[1])
last = tup[0]
data.extend((133, vals[1], 132, vals[0], 133, vals[3], 132, vals[2]))
#enable
data.extend((129, 1))
if raw:
return b64encode(data)
try:
payload = self.request('stop', 'post')
payload = self.request('reset', 'post')
#payload = self.request('divider', 'post', data={'divider': self.clock_divider-1})
#payload = self.request('write', 'post', data=b64encode(bytearray((139, 62))), timeout=20)
n = len(data)
x = 0
#while x < n:
payload = self.request('write', 'post', data=b64encode(data))
# x += 1024
if payload['write']=='ok':
self.device.status = 3
self.device.save()
self.message = 'USRP Tx configured and started'
else:
self.device.status = 1
self.device.save()
self.message = 'USRP Tx write: {}'.format(payload['write'])
return False
#payload = self.request('start', 'post')
except Exception as e:
if 'No route to host' not in str(e):
self.device.status = 4
else:
self.device.status = 0
self.message = 'USRP Tx write: {}'.format(str(e))
self.device.save()
return False
return True
def get_absolute_url_import(self):
return reverse('url_import_usrp_tx_conf', args=[str(self.id)])