##// END OF EJS Templates
radarsys_met
RSS
new structure and views.py (pedestal) update
new structure and views.py (pedestal) update
eynilupu - - Load All Authors

File last commit:

r366:bdf15dfd5ebe
r366:bdf15dfd5ebe
Show More
models.py
310 lines | 9.2 KiB | text/x-python | PythonLexer
/ volumes / sirm / apps / usrp_rx / 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'),
('3', 'A:A A:B'),
)
ANT_VALUE = (
('1', 'RX'),
('2', 'TX')
)
CLK_VALUE = (
('1', 'internal'),
('2', 'external')
)
TIME_VALUE = (
('1', 'internal'),
('2', 'external')
)
class USRPRXConfiguration(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
)
samplerate = models.FloatField(
verbose_name='Sample Rate',
blank=False,
null=False,
help_text='Introduce the value in MHz (10^6)'
)
frequency = models.FloatField(
verbose_name='Frequency',
blank=False,
null=False,
help_text='Introduce the value in MHz (10^6)'
)
datadir = models.CharField(
verbose_name="Data Directory",
max_length=100,
default='',
blank=False,
null=False,
help_text='Fill with a directory. Example: /media/soporte/DATA'
)
clocksource = models.CharField(
verbose_name='Clock Source',
max_length=3,
choices=CLK_VALUE,
null=False,
blank=False
)
timesource = models.CharField(
verbose_name='Time Source',
max_length=3,
choices=TIME_VALUE,
null=False,
blank=False
)
clockrate = models.FloatField(
verbose_name='Clock Rate',
blank=False,
null=False,
help_text='Introduce the value in MHz (10^6)'
)
class Meta:
db_table = 'usrp_rx_configurations'
def __str__(self):
return str(self.label)
def get_absolute_url_plot(self):
return reverse('url_plot_usrp_rx_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 Rx Dev 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 Rx 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 Rx restarted OK'
self.device.status = 2
self.device.save()
else:
self.message = 'USRP Rx restart fail'
self.device.status = 4
self.device.save()
except Exception as e:
self.message = 'USRP Rx reset: {}'.format(str(e))
return False
return True
def stop_device(self):
try:
command = self.device.url() + "stoprx"
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 Rx stop: {}'.format(str(e))
self.message = "USRP Rx can't start, please check network/device connection or IP address/port configuration"
self.device.save()
return False
return True
def start_device(self):
try:
usrp = USRPRXConfiguration.objects.get(pk=self)
usrp_daughterboard = usrp.get_daughterboard_display()
usrp_antenna = usrp.get_antenna_display()
usrp_clocksource = usrp.get_clocksource_display()
usrp_timesource = usrp.get_timesource_display()
print(usrp)
payload = {'ip_address': usrp.ip_address, 'daughterboard': usrp_daughterboard, 'antenna': usrp_antenna, 'sample_rate': usrp.samplerate, 'frequency': usrp.frequency,
'datadir': usrp.datadir, 'clock_source': usrp_clocksource, 'time_source': usrp_timesource, 'clock_rate':usrp.clockrate}
print(payload)
r = requests.post(self.device.url("usrprx"), 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 Rx 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 Rx start: {}'.format(str(e))
self.message = "USRP Rx 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 Rx write: {}'.format(payload['command'])
else:
self.message = payload['programming']
if payload['programming'] == 'fail':
self.message = 'USRP Rx 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 Rx configured and started'
else:
self.device.status = 1
self.device.save()
self.message = 'USRP Rx 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 Rx write: {}'.format(str(e))
self.device.save()
return False
return True
def get_absolute_url_import(self):
return reverse('url_import_usrp_rx_conf', args=[str(self.id)])