models.py
578 lines
| 19.3 KiB
| text/x-python
|
PythonLexer
|
r100 | from django.shortcuts import render, redirect, get_object_or_404, HttpResponse | |
|
r85 | from datetime import datetime | |
|
r0 | from django.db import models | |
|
r6 | from polymorphic import PolymorphicModel | |
|
r0 | ||
|
r32 | from django.core.urlresolvers import reverse | |
|
r108 | ||
|
r41 | CONF_STATES = ( | |
|
r47 | (0, 'Disconnected'), | |
(1, 'Connected'), | |||
|
r72 | (2, 'Running'), | |
) | |||
EXP_STATES = ( | |||
|
r84 | (0,'Error'), #RED | |
(1,'Configurated'), #BLUE | |||
(2,'Running'), #GREEN | |||
(3,'Waiting'), #YELLOW | |||
(4,'Not Configured'), #WHITE | |||
|
r47 | ) | |
|
r41 | ||
|
r21 | CONF_TYPES = ( | |
|
r47 | (0, 'Active'), | |
(1, 'Historical'), | |||
) | |||
|
r16 | ||
DEV_STATES = ( | |||
|
r47 | (0, 'No connected'), | |
(1, 'Connected'), | |||
(2, 'Configured'), | |||
(3, 'Running'), | |||
) | |||
|
r2 | ||
|
r13 | DEV_TYPES = ( | |
|
r47 | ('', 'Select a device type'), | |
('rc', 'Radar Controller'), | |||
|
r106 | ('rc_mix', 'Radar Controller (Mix)'), | |
|
r47 | ('dds', 'Direct Digital Synthesizer'), | |
('jars', 'Jicamarca Radar Acquisition System'), | |||
('usrp', 'Universal Software Radio Peripheral'), | |||
('cgs', 'Clock Generator System'), | |||
('abs', 'Automatic Beam Switching'), | |||
) | |||
DEV_PORTS = { | |||
'rc' : 2000, | |||
|
r106 | 'rc_mix': 2000, | |
|
r47 | 'dds' : 2000, | |
'jars' : 2000, | |||
'usrp' : 2000, | |||
'cgs' : 8080, | |||
'abs' : 8080 | |||
} | |||
|
r2 | ||
|
r49 | RADAR_STATES = ( | |
(0, 'No connected'), | |||
|
r84 | (1, 'Connected'), | |
|
r49 | (2, 'Configured'), | |
(3, 'Running'), | |||
(4, 'Scheduled'), | |||
) | |||
|
r13 | # Create your models here. | |
|
r53 | ||
class Location(models.Model): | |||
name = models.CharField(max_length = 30) | |||
description = models.TextField(blank=True, null=True) | |||
class Meta: | |||
db_table = 'db_location' | |||
def __unicode__(self): | |||
return u'%s' % self.name | |||
|
r89 | def get_absolute_url(self): | |
return reverse('url_device', args=[str(self.id)]) | |||
|
r2 | class DeviceType(models.Model): | |
|
r13 | name = models.CharField(max_length = 10, choices = DEV_TYPES, default = 'rc') | |
description = models.TextField(blank=True, null=True) | |||
|
r2 | ||
class Meta: | |||
|
r13 | db_table = 'db_device_types' | |
|
r2 | ||
def __unicode__(self): | |||
|
r22 | return u'%s' % self.get_name_display() | |
|
r2 | ||
class Device(models.Model): | |||
|
r53 | device_type = models.ForeignKey(DeviceType, on_delete=models.CASCADE) | |
location = models.ForeignKey(Location, on_delete=models.CASCADE) | |||
|
r41 | ||
|
r9 | name = models.CharField(max_length=40, default='') | |
|
r6 | ip_address = models.GenericIPAddressField(protocol='IPv4', default='0.0.0.0') | |
|
r13 | port_address = models.PositiveSmallIntegerField(default=2000) | |
|
r9 | description = models.TextField(blank=True, null=True) | |
|
r16 | status = models.PositiveSmallIntegerField(default=0, choices=DEV_STATES) | |
|
r2 | ||
class Meta: | |||
|
r13 | db_table = 'db_devices' | |
|
r2 | ||
def __unicode__(self): | |||
|
r22 | return u'%s | %s' % (self.name, self.ip_address) | |
|
r47 | ||
|
r89 | def get_status(self): | |
|
r47 | return self.status | |
|
r89 | def get_absolute_url(self): | |
return reverse('url_device', args=[str(self.id)]) | |||
|
r13 | ||
class Campaign(models.Model): | |||
|
r85 | template = models.BooleanField(default=False) | |
name = models.CharField(max_length=60, unique=True) | |||
|
r13 | start_date = models.DateTimeField(blank=True, null=True) | |
end_date = models.DateTimeField(blank=True, null=True) | |||
tags = models.CharField(max_length=40) | |||
description = models.TextField(blank=True, null=True) | |||
|
r85 | experiments = models.ManyToManyField('Experiment', blank=True) | |
|
r13 | ||
class Meta: | |||
db_table = 'db_campaigns' | |||
|
r85 | ordering = ('name',) | |
|
r13 | ||
def __unicode__(self): | |||
|
r22 | return u'%s' % (self.name) | |
|
r72 | ||
|
r89 | def get_absolute_url(self): | |
return reverse('url_campaign', args=[str(self.id)]) | |||
|
r100 | def parms_to_dict(self): | |
import json | |||
parameters = {} | |||
|
r102 | exp_parameters = {} | |
experiments = Experiment.objects.filter(campaign = self) | |||
|
r100 | ||
|
r102 | i=1 | |
for experiment in experiments: | |||
exp_parameters['experiment-'+str(i)] = json.loads(experiment.parms_to_dict()) | |||
i += 1 | |||
|
r100 | ||
|
r103 | ||
|
r108 | parameters['experiments'] = exp_parameters | |
parameters['end_date'] = self.end_date.strftime("%Y-%m-%d") | |||
parameters['start_date'] = self.start_date.strftime("%Y-%m-%d") | |||
parameters['campaign'] = self.__unicode__() | |||
parameters['tags'] =self.tags | |||
|
r103 | ||
parameters = json.dumps(parameters, indent=2, sort_keys=False) | |||
|
r100 | ||
return parameters | |||
|
r108 | def import_from_file(self, fp): | |
import os, json | |||
parms = {} | |||
path, ext = os.path.splitext(fp.name) | |||
if ext == '.json': | |||
parms = json.load(fp) | |||
return parms | |||
|
r109 | def dict_to_parms(self, parms, CONF_MODELS): | |
experiments = Experiment.objects.filter(campaign = self) | |||
configurations = Configuration.objects.filter(experiment = experiments) | |||
if configurations: | |||
for configuration in configurations: | |||
configuration.delete() | |||
if experiments: | |||
for experiment in experiments: | |||
experiment.delete() | |||
for parms_exp in parms['experiments']: | |||
location = Location.objects.get(name = parms['experiments'][parms_exp]['radar']) | |||
new_exp = Experiment( | |||
name = parms['experiments'][parms_exp]['experiment'], | |||
location = location, | |||
start_time = parms['experiments'][parms_exp]['start_time'], | |||
end_time = parms['experiments'][parms_exp]['end_time'], | |||
) | |||
new_exp.save() | |||
new_exp.dict_to_parms(parms['experiments'][parms_exp],CONF_MODELS) | |||
new_exp.save() | |||
|
r110 | self.name = parms['campaign'] | |
self.start_date = parms['start_date'] | |||
self.end_date = parms['end_date'] | |||
self.tags = parms['tags'] | |||
|
r109 | self.experiments.add(new_exp) | |
|
r110 | self.save() | |
return self | |||
|
r109 | ||
|
r108 | def get_absolute_url(self): | |
return reverse('url_campaign', args=[str(self.id)]) | |||
def get_absolute_url_edit(self): | |||
return reverse('url_edit_campaign', args=[str(self.id)]) | |||
|
r100 | def get_absolute_url_export(self): | |
return reverse('url_export_campaign', args=[str(self.id)]) | |||
|
r108 | def get_absolute_url_import(self): | |
return reverse('url_import_campaign', args=[str(self.id)]) | |||
|
r100 | ||
|
r85 | ||
|
r84 | class RunningExperiment(models.Model): | |
radar = models.OneToOneField('Location', on_delete=models.CASCADE) | |||
|
r92 | running_experiment = models.ManyToManyField('Experiment', blank = True) | |
|
r84 | status = models.PositiveSmallIntegerField(default=0, choices=RADAR_STATES) | |
|
r48 | ||
|
r49 | ||
|
r2 | class Experiment(models.Model): | |
|
r85 | template = models.BooleanField(default=False) | |
name = models.CharField(max_length=40, default='', unique=True) | |||
|
r53 | location = models.ForeignKey('Location', null=True, blank=True, on_delete=models.CASCADE) | |
|
r13 | start_time = models.TimeField(default='00:00:00') | |
end_time = models.TimeField(default='23:59:59') | |||
|
r84 | status = models.PositiveSmallIntegerField(default=0, choices=EXP_STATES) | |
|
r2 | ||
class Meta: | |||
|
r13 | db_table = 'db_experiments' | |
|
r91 | ordering = ('template', 'name') | |
|
r2 | ||
def __unicode__(self): | |||
|
r91 | if self.template: | |
return u'%s (template)' % (self.name) | |||
else: | |||
return u'%s' % (self.name) | |||
|
r6 | ||
|
r89 | @property | |
def radar(self): | |||
return self.location | |||
|
r85 | def clone(self, **kwargs): | |
confs = Configuration.objects.filter(experiment=self, type=0) | |||
self.pk = None | |||
self.name = '{} [{:%Y/%m/%d}]'.format(self.name, datetime.now()) | |||
for attr, value in kwargs.items(): | |||
setattr(self, attr, value) | |||
self.save() | |||
for conf in confs: | |||
conf.clone(experiment=self, template=False) | |||
return self | |||
|
r84 | ||
def get_status(self): | |||
configurations = Configuration.objects.filter(experiment=self) | |||
exp_status=[] | |||
for conf in configurations: | |||
print conf.status_device() | |||
exp_status.append(conf.status_device()) | |||
if not exp_status: #No Configuration | |||
self.status = 4 | |||
self.save() | |||
return | |||
total = 1 | |||
for e_s in exp_status: | |||
total = total*e_s | |||
if total == 0: #Error | |||
status = 0 | |||
elif total == (3**len(exp_status)): #Running | |||
status = 2 | |||
else: | |||
status = 1 #Configurated | |||
self.status = status | |||
self.save() | |||
def status_color(self): | |||
color = 'danger' | |||
if self.status == 0: | |||
color = "danger" | |||
elif self.status == 1: | |||
color = "info" | |||
elif self.status == 2: | |||
|
r91 | color = "success" | |
|
r84 | elif self.status == 3: | |
color = "warning" | |||
else: | |||
color = "muted" | |||
return color | |||
|
r89 | def get_absolute_url(self): | |
return reverse('url_experiment', args=[str(self.id)]) | |||
|
r100 | def parms_to_dict(self): | |
import json | |||
configurations = Configuration.objects.filter(experiment=self) | |||
conf_parameters = {} | |||
parameters={} | |||
for configuration in configurations: | |||
if 'cgs' in configuration.device.device_type.name: | |||
conf_parameters['cgs'] = configuration.parms_to_dict() | |||
if 'dds' in configuration.device.device_type.name: | |||
conf_parameters['dds'] = configuration.parms_to_dict() | |||
if 'rc' in configuration.device.device_type.name: | |||
conf_parameters['rc'] = configuration.parms_to_dict() | |||
if 'jars' in configuration.device.device_type.name: | |||
conf_parameters['jars'] = configuration.parms_to_dict() | |||
if 'usrp' in configuration.device.device_type.name: | |||
conf_parameters['usrp'] = configuration.parms_to_dict() | |||
if 'abs' in configuration.device.device_type.name: | |||
conf_parameters['abs'] = configuration.parms_to_dict() | |||
parameters['configurations'] = conf_parameters | |||
|
r108 | parameters['end_time'] = self.end_time.strftime("%H:%M:%S") | |
parameters['start_time'] = self.start_time.strftime("%H:%M:%S") | |||
parameters['radar'] = self.radar.name | |||
parameters['experiment'] = self.name | |||
|
r100 | parameters = json.dumps(parameters, indent=2) | |
return parameters | |||
|
r108 | def import_from_file(self, fp): | |
import os, json | |||
parms = {} | |||
path, ext = os.path.splitext(fp.name) | |||
if ext == '.json': | |||
parms = json.load(fp) | |||
return parms | |||
def dict_to_parms(self, parms, CONF_MODELS): | |||
configurations = Configuration.objects.filter(experiment=self) | |||
if configurations: | |||
for configuration in configurations: | |||
configuration.delete() | |||
for conf_type in parms['configurations']: | |||
#--For ABS Device: | |||
#--For USRP Device: | |||
#--For JARS Device: | |||
#--For RC Device: | |||
if conf_type == 'rc': | |||
device = get_object_or_404(Device, pk=parms['configurations']['rc']['device_id']) | |||
DevConfModel = CONF_MODELS[conf_type] | |||
confrc_form = DevConfModel( | |||
experiment = self, | |||
name = 'RC', | |||
device=device, | |||
) | |||
confrc_form.dict_to_parms(parms['configurations']['rc']) | |||
confrc_form.save() | |||
#--For DDS Device: | |||
if conf_type == 'dds': | |||
device = get_object_or_404(Device, pk=parms['configurations']['dds']['device_id']) | |||
DevConfModel = CONF_MODELS[conf_type] | |||
confdds_form = DevConfModel( | |||
experiment = self, | |||
name = 'DDS', | |||
device=device, | |||
) | |||
confdds_form.dict_to_parms(parms['configurations']['dds']) | |||
confdds_form.save() | |||
#--For CGS Device: | |||
if conf_type == 'cgs': | |||
device = get_object_or_404(Device, pk=parms['configurations']['cgs']['device_id']) | |||
DevConfModel = CONF_MODELS[conf_type] | |||
confcgs_form = DevConfModel( | |||
experiment = self, | |||
name = 'CGS', | |||
device=device, | |||
) | |||
confcgs_form.dict_to_parms(parms['configurations']['cgs']) | |||
confcgs_form.save() | |||
|
r110 | ||
location = Location.objects.get(name = parms['radar']) | |||
self.name = parms['experiment'] | |||
self.location = location | |||
self.start_time = parms['start_time'] | |||
self.end_time = parms['end_time'] | |||
self.save() | |||
return self | |||
|
r108 | ||
def get_absolute_url_edit(self): | |||
return reverse('url_edit_experiment', args=[str(self.id)]) | |||
def get_absolute_url_import(self): | |||
return reverse('url_import_experiment', args=[str(self.id)]) | |||
|
r100 | def get_absolute_url_export(self): | |
return reverse('url_export_experiment', args=[str(self.id)]) | |||
|
r85 | ||
|
r6 | class Configuration(PolymorphicModel): | |
|
r2 | ||
|
r47 | template = models.BooleanField(default=False) | |
name = models.CharField(verbose_name="Configuration Name", max_length=40, default='') | |||
|
r53 | experiment = models.ForeignKey('Experiment', null=True, blank=True, on_delete=models.CASCADE) | |
|
r106 | device = models.ForeignKey(Device, null=True, on_delete=models.CASCADE) | |
|
r41 | ||
|
r21 | type = models.PositiveSmallIntegerField(default=0, choices=CONF_TYPES) | |
|
r16 | ||
|
r22 | created_date = models.DateTimeField(auto_now_add=True) | |
programmed_date = models.DateTimeField(auto_now=True) | |||
parameters = models.TextField(default='{}') | |||
|
r53 | message = "" | |
|
r2 | class Meta: | |
|
r13 | db_table = 'db_configurations' | |
|
r2 | ||
def __unicode__(self): | |||
|
r106 | ||
return u'[%s]: %s' % (self.device.name, self.name) | |||
|
r79 | ||
|
r85 | def clone(self, **kwargs): | |
|
r79 | ||
|
r85 | self.pk = None | |
self.id = None | |||
for attr, value in kwargs.items(): | |||
setattr(self, attr, value) | |||
self.save() | |||
return self | |||
|
r53 | ||
def parms_to_dict(self): | |||
parameters = {} | |||
for key in self.__dict__.keys(): | |||
parameters[key] = getattr(self, key) | |||
return parameters | |||
|
r57 | def parms_to_text(self): | |
raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |||
return '' | |||
def parms_to_binary(self): | |||
raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |||
return '' | |||
|
r53 | def dict_to_parms(self, parameters): | |
if type(parameters) != type({}): | |||
return | |||
for key in parameters.keys(): | |||
setattr(self, key, parameters[key]) | |||
def export_to_file(self, format="json"): | |||
import json | |||
|
r32 | ||
|
r57 | content_type = '' | |
|
r53 | ||
if format == 'text': | |||
content_type = 'text/plain' | |||
|
r57 | filename = '%s_%s.%s' %(self.device.device_type.name, self.name, self.device.device_type.name) | |
content = self.parms_to_text() | |||
|
r53 | ||
if format == 'binary': | |||
content_type = 'application/octet-stream' | |||
|
r57 | filename = '%s_%s.bin' %(self.device.device_type.name, self.name) | |
content = self.parms_to_binary() | |||
if not content_type: | |||
content_type = 'application/json' | |||
filename = '%s_%s.json' %(self.device.device_type.name, self.name) | |||
|
r79 | content = json.dumps(self.parms_to_dict(), indent=2) | |
|
r53 | ||
fields = {'content_type':content_type, | |||
'filename':filename, | |||
'content':content | |||
} | |||
return fields | |||
|
r57 | def import_from_file(self, fp): | |
import os, json | |||
parms = {} | |||
path, ext = os.path.splitext(fp.name) | |||
|
r53 | ||
|
r57 | if ext == '.json': | |
parms = json.load(fp) | |||
|
r53 | ||
|
r57 | return parms | |
|
r53 | ||
def status_device(self): | |||
|
r57 | raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |
|
r53 | ||
return None | |||
def stop_device(self): | |||
|
r57 | raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |
|
r53 | ||
return None | |||
def start_device(self): | |||
|
r57 | raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |
|
r53 | ||
return None | |||
def write_device(self, parms): | |||
|
r57 | raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |
|
r53 | ||
return None | |||
def read_device(self): | |||
|
r57 | raise NotImplementedError, "This method should be implemented in %s Configuration model" %str(self.device.device_type.name).upper() | |
|
r53 | ||
return None | |||
|
r47 | def get_absolute_url(self): | |
|
r23 | return reverse('url_%s_conf' % self.device.device_type.name, args=[str(self.id)]) | |
|
r30 | ||
def get_absolute_url_edit(self): | |||
return reverse('url_edit_%s_conf' % self.device.device_type.name, args=[str(self.id)]) | |||
def get_absolute_url_import(self): | |||
|
r53 | return reverse('url_import_dev_conf', args=[str(self.id)]) | |
|
r30 | ||
def get_absolute_url_export(self): | |||
|
r53 | return reverse('url_export_dev_conf', args=[str(self.id)]) | |
|
r30 | ||
def get_absolute_url_write(self): | |||
|
r53 | return reverse('url_write_dev_conf', args=[str(self.id)]) | |
|
r30 | ||
def get_absolute_url_read(self): | |||
|
r53 | return reverse('url_read_dev_conf', args=[str(self.id)]) | |
|
r49 | ||
|
r53 | def get_absolute_url_start(self): | |
return reverse('url_start_dev_conf', args=[str(self.id)]) | |||
|
r49 | ||
|
r53 | def get_absolute_url_stop(self): | |
return reverse('url_stop_dev_conf', args=[str(self.id)]) | |||
def get_absolute_url_status(self): | |||
return reverse('url_status_dev_conf', args=[str(self.id)]) |