#from django.db import models # Create your models here. ##from __future__ import unicode_literals # (EP 21/9/22) To allow autoreferencing (ex: AgentCmd.create() returns a AgentCmd) from __future__ import annotations # Stdlib imports from numpy import False_ from src.device_controller.abstract_component.device_controller import DeviceCmd from enum import Enum from datetime import datetime, timedelta, date from dateutil.relativedelta import relativedelta import os import sys from typing import Any, List, Tuple, Optional import re # Django imports from django.core.validators import MaxValueValidator, MinValueValidator # DJANGO imports from django.contrib.auth.models import AbstractUser, UserManager from django.db import models from django.db.models import Q, Max from django.core.validators import MaxValueValidator, MinValueValidator from django.db.models.deletion import DO_NOTHING from django.db.models.expressions import F from django.db.models.query import QuerySet from model_utils import Choices from django.utils import timezone # Project imports # DeviceCommand is used by class Command sys.path.append("../../..") ''' NOT USED - to be removed class PyrosState(Enum): START = 'Starting' PA = 'Passive' INI = "INIT" STAND = "Standby" SCHED_START = 'Scheduler startup' SCHED = 'Scheduler' SCHED_CLOSE = 'Scheduler closing' ''' """ STYLE RULES =========== https://simpleisbetterthancomplex.com/tips/2018/02/10/django-tip-22-designing-better-models.html https://steelkiwi.com/blog/best-practices-working-django-models-python/ - Model name => singular Call it Company instead of Companies. A model definition is the representation of a single object (the object in this example is a company), and not a collection of companies The model definition is a class, so always use CapWords convention (no underscores) E.g. User, Permission, ContentType, etc. - For the model’s attributes use snake_case. E.g. first_name, last_name, etc - Blank and Null Fields (https://simpleisbetterthancomplex.com/tips/2016/07/25/django-tip-8-blank-or-null.html) - Null: It is database-related. Defines if a given database column will accept null values or not. - Blank: It is validation-related. It will be used during forms validation, when calling form.is_valid(). Do not use null=True for text-based fields that are optional. Otherwise, you will end up having two possible values for “no data”, that is: None and an empty string. Having two possible values for “no data” is redundant. The Django convention is to use the empty string, not NULL. Example: # The default values of `null` and `blank` are `False`. class Person(models.Model): name = models.CharField(max_length=255) # Mandatory bio = models.TextField(max_length=500, blank=True) # Optional (don't put null=True) birth_date = models.DateField(null=True, blank=True) # Optional (here you may add null=True) The default values of null and blank are False. Special case, when you need to accept NULL values for a BooleanField, use NullBooleanField instead. - Choices : you can use Choices from the model_utils library. Take model Article, for instance: from model_utils import Choices class Article(models.Model): STATUSES = Choices( (0, 'draft', _('draft')), (1, 'published', _('published')) ) status = models.IntegerField(choices=STATUSES, default=STATUSES.draft) - Reverse Relationships - related_name : Rule of thumb: if you are not sure what would be the related_name, use the plural of the model holding the ForeignKey. ex: class Company: name = models.CharField(max_length=30) class Employee: first_name = models.CharField(max_length=30) last_name = models.CharField(max_length=30) company = models.ForeignKey(Company, on_delete=models.CASCADE, related_name='employees') usage: google = Company.objects.get(name='Google') google.employees.all() You can also use the reverse relationship to modify the company field on the Employee instances: vitor = Employee.objects.get(first_name='Vitor') google = Company.objects.get(name='Google') google.employees.add(vitor) - related_query_name : This kind of relationship also applies to query filters. For example, if I wanted to list all companies that employs people named ‘Vitor’, I could do the following: companies = Company.objects.filter(employee__first_name='Vitor') If you want to customize the name of this relationship, here is how we do it: class Employee: first_name = models.CharField(max_length=30) last_name = models.CharField(max_length=30) company = models.ForeignKey( Company, on_delete=models.CASCADE, related_name='employees', related_query_name='person' ) Then the usage would be: companies = Company.objects.filter(person__first_name='Vitor') To use it consistently, related_name goes as plural and related_query_name goes as singular. GENERAL EXAMPLE ======= from django.db import models from django.urls import reverse class Company(models.Model): # CHOICES PUBLIC_LIMITED_COMPANY = 'PLC' PRIVATE_COMPANY_LIMITED = 'LTD' LIMITED_LIABILITY_PARTNERSHIP = 'LLP' COMPANY_TYPE_CHOICES = ( (PUBLIC_LIMITED_COMPANY, 'Public limited company'), (PRIVATE_COMPANY_LIMITED, 'Private company limited by shares'), (LIMITED_LIABILITY_PARTNERSHIP, 'Limited liability partnership'), ) # DATABASE FIELDS name = models.CharField('name', max_length=30) vat_identification_number = models.CharField('VAT', max_length=20) company_type = models.CharField('type', max_length=3, choices=COMPANY_TYPE_CHOICES) # MANAGERS objects = models.Manager() limited_companies = LimitedCompanyManager() # META CLASS class Meta: verbose_name = 'company' verbose_name_plural = 'companies' # TO STRING METHOD def __str__(self): return self.name # SAVE METHOD def save(self, *args, **kwargs): do_something() super().save(*args, **kwargs) # Call the "real" save() method. do_something_else() # ABSOLUTE URL METHOD def get_absolute_url(self): return reverse('company_details', kwargs={'pk': self.id}) # OTHER METHODS def process_invoices(self): do_something() """ # --- # --- Utility functions # --- ''' def printd(*args, **kwargs): if os.environ.get('PYROS_DEBUG', '0') == '1': print('(MODEL)', *args, **kwargs) ''' def get_or_create_unique_row_from_model(model: models.Model): # return model.objects.get(id=1) if model.objects.exists() else model.objects.create(id=1) return model.objects.first() if model.objects.exists() else model.objects.create(id=1) """ ------------------------ BASE MODEL CLASSES ------------------------ """ class Device(models.Model): name = models.CharField(max_length=45, blank=True, null=True) desc = models.TextField(blank=True, null=True) created = models.DateTimeField(blank=True, null=True, auto_now_add=True) updated = models.DateTimeField(blank=True, null=True, auto_now=True) is_online = models.BooleanField(default=False) status = models.CharField(max_length=11, blank=True, null=True) maintenance_date = models.DateTimeField(blank=True, null=True) class Meta: abstract = True def __str__(self): return (str(self.name)) ''' class Image(models.Model): plan = models.ForeignKey( 'Plan', on_delete=models.CASCADE, related_name="images") nrtanalysis = models.ForeignKey( 'NrtAnalysis', models.DO_NOTHING, blank=True, null=True, related_name="images") name = models.CharField(max_length=45, blank=True, null=True) desc = models.TextField(blank=True, null=True) created = models.DateTimeField(blank=True, null=True, auto_now_add=True) updated = models.DateTimeField(blank=True, null=True, auto_now=True) date_from_gps = models.CharField(max_length=45, blank=True, null=True) level = models.IntegerField(blank=True, null=True) type = models.CharField(max_length=5, blank=True, null=True) quality = models.CharField(max_length=45, blank=True, null=True) flaggps = models.CharField(max_length=45, blank=True, null=True) exposure = models.CharField(max_length=45, blank=True, null=True) tempext = models.CharField(max_length=45, blank=True, null=True) pressure = models.CharField(max_length=45, blank=True, null=True) humidext = models.CharField(max_length=45, blank=True, null=True) wind = models.CharField(max_length=45, blank=True, null=True) wind_dir = models.CharField(max_length=45, blank=True, null=True) dwnimg = models.CharField(max_length=45, blank=True, null=True) dwncata = models.CharField(max_length=45, blank=True, null=True) dwn = models.CharField(max_length=45, blank=True, null=True) level0_fits_name = models.CharField(max_length=45, blank=True, null=True) level1a_fits_name = models.CharField(max_length=45, blank=True, null=True) level1b_fits_name = models.CharField(max_length=45, blank=True, null=True) class Meta: managed = True db_table = 'image' def __str__(self): return (str(self.name)) ''' """ ------------------------ OTHER MODEL CLASSES ------------------------ """ # TODO: A VIRER car remplacé par AgentDeviceStatus class AgentDeviceTelescopeStatus(models.Model): #created = models.DateTimeField('status date', blank=True, null=True, auto_now_add=True) updated = models.DateTimeField( 'status date', blank=True, null=True, auto_now=True) radec = models.CharField('agent mode', max_length=30, blank=True) class Meta: managed = True db_table = 'agent_device_telescope_status' #verbose_name = "agent survey" #verbose_name_plural = "agents survey" """ def __str__(self): return (f"Agent {self.name} at {self.updated} in mode {self.mode} and status {self.status}") """ class AgentDeviceStatus(models.Model): """Table storing various status parameters for EACH Device. Attributes: attr1 (str): Description of `attr1`. attr2 (:obj:`int`, optional): Description of `attr2`. """ #created = models.DateTimeField('status date', blank=True, null=True, auto_now_add=True) agent = models.CharField( 'Name of the agent that saved this parameter', max_length=45, blank=True, null=True) #radec = models.CharField('agent mode', max_length=30, blank=True) status = models.CharField( 'status parameters json dictionnary (ex: {radec:..., speed:...})', max_length=300, blank=True, null=True) date_updated = models.DateTimeField( 'status parameter date', blank=True, null=True, auto_now=True) class Meta: managed = True db_table = 'agent_device_status' verbose_name = "agent device status" verbose_name_plural = "agent devices status" def __str__(self): return (f"Agent {self.agent} last status is ({self.status}) (saved at {self.date_updated})") @classmethod def getStatusForAgent(cls, agent: str) -> str: return cls.objects.filter(agent=agent)[0] if cls.objects.filter(agent=agent).exists() else cls.objects.create(agent=agent) ''' return cls.objects.filter(agent=agent)[0].status if cls.objects.filter(agent=agent).exists() else cls.objects.create(agent=agent).status agent_status = cls.objects.filter(agent=agent) if agent_status.exists(): return agent_status[0].status else: return cls.objects.create(agent=agent) ''' class PlcDeviceStatus(models.Model): device = models.ForeignKey( 'PlcDevice', on_delete=models.CASCADE, related_name='current_status') created = models.DateTimeField( auto_now_add=True, editable=False, blank=True) outside_temp = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) outside_temp_unit = models.CharField(max_length=45, blank=True, null=True) outside_humidity = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) outside_humidity_unit = models.CharField( max_length=45, blank=True, null=True) pressure = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) pressure_unit = models.CharField(max_length=45, blank=True, null=True) rain_rate = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) rain_rate_unit = models.CharField(max_length=45, blank=True, null=True) wind_speed = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) wind_speed_unit = models.CharField(max_length=45, blank=True, null=True) wind_dir = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) wind_dir_unit = models.CharField(max_length=45, blank=True, null=True) dew_point = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) dew_point_unit = models.CharField(max_length=45, blank=True, null=True) analog = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) analog_unit = models.CharField(max_length=45, blank=True, null=True) digital = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) digital_unit = models.CharField(max_length=45, blank=True, null=True) inside_temp = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) inside_temp_unit = models.CharField(max_length=45, blank=True, null=True) inside_humidity = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) inside_humidity_unit = models.CharField( max_length=45, blank=True, null=True) wind_dir_cardinal = models.CharField(max_length=45, blank=True, null=True) wind_dir_cardinal_unit = models.CharField( max_length=45, blank=True, null=True) sensor_temperature = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) sensor_temperature_unit = models.CharField( max_length=45, blank=True, null=True) sky_temperature = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) sky_temperature_unit = models.CharField( max_length=45, blank=True, null=True) status = models.CharField(max_length=45, blank=True, null=True) current = models.DecimalField( max_digits=15, decimal_places=8, blank=True, null=True) current_unit = models.CharField(max_length=45, blank=True, null=True) is_safe = models.BooleanField(default=True) plc_mode = models.CharField(max_length=4, null=True) lights = models.CharField(max_length=3, null=True) shutters = models.CharField(max_length=5, null=True) class Meta: managed = True db_table = 'plc_devices_status' def __str__(self): return (str(self.__dict__)) """ TODO : This function is Ugly, we should change this with a function pointer array and setters getters for each attribute """ def setValue(self, key, value, unit=""): if key == "Temperature_outside": self.outside_temp = value self.outside_temp_unit = unit elif key == "Humidity_outside": self.outside_humidity = value self.outside_humidity_unit = unit elif key == "_Pressure": self.pressure = value self.pressure_unit = unit elif key == "Rain_boolean": # RainRate self.rain_rate = value self.rain_rate_unit = 'boulean' elif key == "Wind_speed": self.wind_speed = value self.wind_speed_unit = unit elif key == "Wind_dir": self.wind_dir = value self.wind_dir_unit = unit elif key == "_DewPoint": self.dew_point = value self.dew_point_unit = unit elif key == "_analog": self.analog = value self.analog_unit = unit elif key == "_digital": self.digital = value self.digital_unit = unit elif key == "_InsideTemp": self.inside_temp = value self.inside_temp_unit = unit elif key == "_InsideHumidity": self.inside_humidity = value self.inside_humidity_unit = unit elif key == "_WindDirCardinal": self.wind_dir_cardinal = value self.wind_dir_cardinal_unit = unit elif key == "_SensorTemperature": self.sensor_temperature = value self.sensor_temperature_unit = unit elif key == "_SkyTemperature": self.sky_temperature = value self.sky_temperature_unit = unit # PM 20190222 try patch elif key == "Error_code": self.status = value elif key == "current": self.current = value self.current_unit = unit elif key == "mode": self.plc_mode = value elif key == "is_safe": self.is_safe = value elif key == "LIGHTS": self.lights = value elif key == "SHUTTERS": self.shutters = value else: # PM 20190222 ignore unrecognized #raise KeyError("Key " + str(key) + " unrecognized") pass # class Plc(Device): # last_update_status = models.DateTimeField(blank=True, null=True) # i # class Meta: # managed = True # db_table = 'plc' class PlcDevice(Device): #device = models.ForeignKey('Plc', on_delete=models.CASCADE, related_name='plc_devices') name = models.CharField(max_length=45, blank=True, null=True) desc = models.TextField(blank=True, null=True) created = models.DateTimeField(blank=True, null=True, auto_now_add=True) updated = models.DateTimeField(blank=True, null=True, auto_now=True) class Meta: managed = True db_table = 'plc_devices' def __str__(self): return str(self.name) class Detector(Device): VIS = "Visible camera" NIR = "Cagire" telescope = models.ForeignKey( 'Telescope', models.DO_NOTHING, related_name="detectors") nb_photo_x = models.IntegerField(blank=True, null=True) nb_photo_y = models.IntegerField(blank=True, null=True) photo_size_x = models.IntegerField(blank=True, null=True) photo_size_y = models.IntegerField(blank=True, null=True) has_shutter = models.BooleanField(default=False) equivalent_foc_len = models.CharField(max_length=45, blank=True, null=True) acq_start = models.DateTimeField(blank=True, null=True) acq_stop = models.DateTimeField(blank=True, null=True) check_temp = models.FloatField(blank=True, null=True) gain = models.FloatField(blank=True, null=True) readout_noise = models.FloatField(blank=True, null=True) readout_time = models.FloatField(blank=True, null=True) idcam_readout_mode = models.IntegerField(blank=True, null=True) class Meta: managed = True db_table = 'detector' def __str__(self): return str(self.name) def device_name(self): return self.__str__() device_name.short_description = "Name" class Dome(Device): DOME = "Dome" open = models.BooleanField(default=False, blank=True) class Meta: managed = True db_table = 'dome' def __str__(self): return str(self.name) def device_name(self): return self.__str__() device_name.short_description = "Name" class Filter(Device): VIS_FILTER_1 = "First visible filter" VIS_FILTER_2 = "Second visible filter" NIR_FILTER_1 = "First infrared filter" NIR_FILTER_2 = "Second infrared filter" filter_wheel = models.ForeignKey( "FilterWheel", models.DO_NOTHING, related_name="filters", blank=True, null=True) category = models.CharField(max_length=1, blank=True, null=True) transmission_curve_doc = models.CharField( max_length=45, blank=True, null=True) class Meta: managed = True db_table = 'filter' def __str__(self): return (str(self.name)) def device_name(self): return self.__str__() device_name.short_description = "Name" class FilterWheel(Device): detector = models.OneToOneField(Detector, on_delete=models.CASCADE, related_name="filter_wheel", blank=True, null=True) class Meta: managed = True db_table = 'filter_wheel' def __str__(self): return (str(self.name)) def device_name(self): return self.__str__() device_name.short_description = "Name" class Telescope(Device): TELESCOPE = "Telescope" mount_type = models.CharField(max_length=9, blank=True, null=True) diameter = models.FloatField(blank=True, null=True) latitude = models.FloatField(blank=True, null=True) longitude = models.FloatField(blank=True, null=True) sens = models.CharField(max_length=1, blank=True, null=True) altitude = models.FloatField(blank=True, null=True) readout_time = models.IntegerField(blank=True, null=True) slew_time = models.IntegerField(blank=True, null=True) slew_dead = models.IntegerField(blank=True, null=True) slew_rate_max = models.FloatField(blank=True, null=True) horizon_type = models.CharField(max_length=45, blank=True, null=True) horizon_def = models.FloatField(blank=True, null=True) lim_dec_max = models.FloatField(blank=True, null=True) lim_dec_min = models.FloatField(blank=True, null=True) lim_ha_rise = models.FloatField(blank=True, null=True) lim_ha_set = models.FloatField(blank=True, null=True) address = models.CharField(max_length=45, blank=True, null=True) night_elev_sun = models.FloatField(blank=True, null=True) mpc_code = models.CharField(max_length=45, blank=True, null=True) class Meta: managed = True db_table = 'telescope' def __str__(self): return (self.name) class TelescopeCommand(models.Model): created = models.DateTimeField(blank=True, null=True, auto_now_add=True) answered = models.DateTimeField(blank=True, null=True) request = models.CharField(blank=False, null=False, max_length=255) answer = models.TextField(null=True, blank=True) class Meta: managed = True db_table = "telescopecommand" def __str__(self): return str(self.request) + str(self.created)