# models_D3_Seq_submit_and_planning #from django.db import models ##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 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 from common.models import PyrosUser, ScientificProgram, Period # 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 Request(models.Model): # pyros_user = models.ForeignKey(PyrosUser, on_delete=models.DO_NOTHING, related_name="requests") # #pyros_user = models.ForeignKey('PyrosUser', on_delete=models.DO_NOTHING, related_name="requests") # scientific_program = models.ForeignKey(ScientificProgram, on_delete=models.DO_NOTHING, related_name="requests", blank=True, null=True) # #scientific_program = models.ForeignKey('ScientificProgram', on_delete=models.DO_NOTHING, related_name="requests", blank=True, null=True) # 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_alert = models.BooleanField(default=False) # target_type = models.CharField(max_length=8, blank=True, null=True) # status = models.CharField(max_length=10, blank=True, null=True) # autodeposit = models.BooleanField(default=False) # checkpoint = models.CharField(max_length=45, blank=True, null=True) # flag = models.CharField(max_length=45, blank=True, null=True) # complete = models.BooleanField(default=False) # submitted = models.BooleanField(default=False) # class Meta: # managed = True # db_table = 'request' # def __str__(self): # return (str(self.name)) """ ------------------------ OTHER MODEL CLASSES ------------------------ """ class Album(models.Model): sequence = models.ForeignKey('Sequence', on_delete=models.CASCADE, related_name="albums") # detector = models.ForeignKey( # 'Detector', models.DO_NOTHING, related_name="albums", blank=True, null=True) #name_of_channel = models.CharField(blank=True,null=True,max_length=150) 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) complete = models.BooleanField(default=False) class Meta: managed = True db_table = 'album' #verbose_name_plural = "Albums" def __str__(self): return (str(self.name)) # class Alert(Request): # request = models.OneToOneField(Request, on_delete=models.CASCADE, default='', parent_link=True) # strategyobs = models.ForeignKey('StrategyObs', models.DO_NOTHING, related_name="alerts", blank=True, null=True) # voevent_file = models.CharField(max_length=45, blank=True, null=True) # author = models.CharField(max_length=45, blank=True, null=True) # burst_jd = models.DecimalField( # max_digits=15, decimal_places=8, blank=True, null=True) # burst_ra = models.FloatField(max_length=45, blank=True, null=True) # burst_dec = models.FloatField(max_length=45, blank=True, null=True) # astro_coord_system = models.CharField(max_length=45, blank=True, null=True) # jd_send = models.DecimalField( # max_digits=15, decimal_places=8, blank=True, null=True) # jd_received = models.DecimalField( # max_digits=15, decimal_places=8, blank=True, null=True) # trig_id = models.IntegerField(blank=True, null=True) # error_radius = models.FloatField(max_length=45, blank=True, null=True) # defly_not_grb = models.BooleanField(default=False) # editor = models.CharField(max_length=45, blank=True, null=True) # soln_status = models.CharField(max_length=45, blank=True, null=True) # pkt_ser_num = models.IntegerField(blank=True, null=True) # class Meta: # managed = True # db_table = 'alert' # def __str__(self): # return str(self.trig_id) # def request_name(self): # return self.__str__() # request_name.short_description = "Name" class NrtAnalysis(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) analysis = models.TextField(blank=True, null=True) class Meta: managed = True db_table = 'nrtanalysis' verbose_name_plural = "Nrt analyzes" def __str__(self): return (str(self.name)) """ class Plan(models.Model): album = models.ForeignKey(Album, on_delete=models.CASCADE, related_name="plans") filter = models.ForeignKey(Filter, models.DO_NOTHING, related_name="plans", blank=True, null=True) name = models.CharField(max_length=45, blank=True, null=True) desc = models.CharField(max_length=45, 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) duration = models.FloatField(default=0, blank=True, null=True) position = models.CharField(max_length=45, blank=True, null=True) exposure_time = models.FloatField(blank=True, null=True) nb_images = models.IntegerField(blank=True, null=True) dithering = models.BooleanField(default=False) complete = models.BooleanField(default=False) class Meta: managed = True db_table = 'plan' def __str__(self): return (str(self.name)) """ class Plan(models.Model): album = models.ForeignKey( Album, on_delete=models.CASCADE, related_name="plans") created = models.DateTimeField(blank=True, null=True, auto_now_add=True) updated = models.DateTimeField(blank=True, null=True, auto_now=True) duration = models.FloatField(default=0, blank=True, null=True) nb_images = models.PositiveIntegerField( blank=True, null=True, validators=[MinValueValidator(1)]) config_attributes = models.JSONField(blank=True, null=True) complete = models.BooleanField(default=False) class Meta: db_table = "plan" def __str__(self) -> str: return f"Plan of Album {self.album.name} has {self.nb_images} image(s)" 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) # 12 parameters (coscrval1,sincrval1,coscrval2,sincrval2,`cd11rad`,`cd12rad`,`cd21rad`,`cd22rad`,`crpix1`,`crpix2`,`naxis1`,`naxis2`) sky_field = models.JSONField() # File type, i.e. calibration file or image (L0, L1, L2,...) ftype = 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) 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)) # class Schedule(models.Model): # created = models.DateTimeField(blank=True, null=True, auto_now_add=True) # plan_start = models.DecimalField( # default=0.0, max_digits=15, decimal_places=8) # plan_end = models.DecimalField( # default=0.0, max_digits=15, decimal_places=8) # flag = models.CharField(max_length=45, blank=True, null=True) # # class Meta: # managed = True # db_table = 'schedule' # # def __str__(self): # return (str(self.created)) class Schedule(models.Model): sequences = models.ManyToManyField( 'Sequence', through='ScheduleHasSequences', related_name='schedules') created = models.DateTimeField(blank=True, null=True, auto_now_add=True) plan_night_start = models.DecimalField( default=0.0, max_digits=15, decimal_places=8) plan_end = models.DecimalField( default=0.0, max_digits=15, decimal_places=8) plan_start = models.DecimalField( default=0.0, max_digits=15, decimal_places=8) flag = models.CharField(max_length=45, blank=True, null=True) class Meta: managed = True db_table = 'schedule' verbose_name_plural = "Schedules" def __str__(self): return (str(self.created)) class Sequence(models.Model): """ Definition of Status enum values """ INVALID = "INVL" DRAFT = "DRAFT" # INCOMPLETE = "INCPL" # COMPLETE = "CPL" TOBEPLANNED = "TBP" PLANNED = "PLND" UNPLANNABLE = "UNPLN" REJECTED = "RJTD" REC_RUNNING = "RUN" REC_FINISHED = "EXD" REC_CANCELED = "CNCLD" PROC_RUNNING = "PROC_RUN" PROC_CANCELED = "PROC_CNCLD" PROC_FINISHED = "PROC_EXD" # PENDING = "PNDG" # EXECUTING = "EXING" # EXECUTED = "EXD" CANCELLED = "CNCLD" STATUS_CHOICES = ( (INVALID, "Invalid"), (DRAFT, "DRAFT"), (TOBEPLANNED, "To be planned"), (PLANNED, "Planned"), (UNPLANNABLE, "Unplannable"), (REJECTED, "Rejected"), (REC_RUNNING, "Recording running"), (REC_FINISHED, "Recording finished"), (REC_CANCELED, "Recording canceled"), (PROC_RUNNING, "Processing running"), (PROC_FINISHED, "Processing finished"), (PROC_CANCELED, "Processing canceled"), (CANCELLED, "Cancelled"), ) START_EXPO_PREF_CHOICES = ( ('IMMEDIATE', 'IMMEDIATE'), ('BEST_ELEVATION', 'BEST_ELEVATION'), ('NO_CONSTRAINT', 'NO_CONSTRAINT'), ) start_expo_pref = models.CharField(max_length=50, blank=False, null=True, choices=START_EXPO_PREF_CHOICES, default=START_EXPO_PREF_CHOICES[0]) # request = models.ForeignKey( # Request, on_delete=models.CASCADE, related_name="sequences") pyros_user = models.ForeignKey(PyrosUser, on_delete=models.DO_NOTHING, related_name="sequences", blank=True, null=True) #pyros_user = models.ForeignKey('PyrosUser', on_delete=models.DO_NOTHING, related_name="sequences", blank=True, null=True) #scientific_program = models.ForeignKey('ScientificProgram', on_delete=models.DO_NOTHING, related_name="sequences", blank=True, null=True) scientific_program = models.ForeignKey(ScientificProgram, on_delete=models.DO_NOTHING, related_name="sequences", blank=True, null=True) name = models.CharField(max_length=45, blank=True, null=True, unique=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) #last_modified_by = models.ForeignKey('PyrosUser', on_delete=models.DO_NOTHING, related_name="+", blank=True, null=True) last_modified_by = models.ForeignKey(PyrosUser, on_delete=models.DO_NOTHING, related_name="+", blank=True, null=True) is_alert = models.BooleanField(default=False) status = models.CharField( max_length=11, blank=True, null=True, choices=STATUS_CHOICES) target_coords = models.CharField(max_length=100, blank=True, null=True) with_drift = models.BooleanField(default=False) priority = models.IntegerField(blank=True, null=True) analysis_method = models.CharField(max_length=45, blank=True, null=True) moon_min = models.IntegerField(blank=True, null=True) alt_min = models.IntegerField(blank=True, null=True) type = models.CharField(max_length=6, blank=True, null=True) img_current = models.CharField(max_length=45, blank=True, null=True) img_total = models.CharField(max_length=45, blank=True, null=True) not_obs = models.BooleanField(default=False) obsolete = models.BooleanField(default=False) processing = models.BooleanField(default=False) flag = models.CharField(max_length=45, blank=True, null=True) period = models.ForeignKey(Period, on_delete=models.DO_NOTHING, related_name="sequences", blank=True, null=True) #period = models.ForeignKey("Period", on_delete=models.DO_NOTHING, related_name="sequences", blank=True, null=True) start_date = models.DateTimeField( blank=True, null=True, default=timezone.now, editable=True) end_date = models.DateTimeField( blank=True, null=True, default=timezone.now, editable=True) # jd1 et jd2 = julian day start / end jd1 = models.DecimalField(default=0.0, max_digits=15, decimal_places=8) jd2 = models.DecimalField(default=0.0, max_digits=15, decimal_places=8) tolerance_before = models.CharField( max_length=50, default="1s", blank=True, null=True) tolerance_after = models.CharField( max_length=50, default="1min", blank=True, null=True) #t_prefered = models.DecimalField(default=-1.0, max_digits=15, decimal_places=8) duration = models.DecimalField( default=-1.0, max_digits=15, decimal_places=8) # décomposer duration en duration pointing + duration album overhead = models.DecimalField(default=0, max_digits=15, decimal_places=8) submitted = models.BooleanField(default=False) config_attributes = models.JSONField(blank=True, null=True) ra = models.FloatField(blank=True, null=True) dec = models.FloatField(blank=True, null=True) complete = models.BooleanField(default=False, null=True, blank=True) class Meta: managed = True db_table = 'sequence' def __str__(self): return (str(self.name)) class ScheduleHasSequences(models.Model): # (EP) TODO: C'est pas un pb d'utiliser 2 fois le meme nom "shs" pour 2 choses differentes ???!!! schedule = models.ForeignKey( 'Schedule', on_delete=models.CASCADE, related_name="shs") sequence = models.ForeignKey( 'Sequence', on_delete=models.CASCADE, related_name="shs") status = models.CharField( max_length=11, blank=True, null=True, choices=Sequence.STATUS_CHOICES) desc = models.CharField(max_length=45, blank=True, null=True) tsp = models.DecimalField(default=-1.0, max_digits=15, decimal_places=8) tep = models.DecimalField(default=-1.0, max_digits=15, decimal_places=8) deltaTL = models.DecimalField( default=-1.0, max_digits=15, decimal_places=8) deltaTR = models.DecimalField( default=-1.0, max_digits=15, decimal_places=8) class Meta: managed = True db_table = 'schedule_has_sequences' class StrategyObs(models.Model): name = models.CharField(max_length=45, blank=True, null=True) desc = models.TextField(blank=True, null=True) xml_file = models.CharField(max_length=45, blank=True, null=True) is_default = models.BooleanField(default=False) class Meta: managed = True db_table = 'strategyobs' verbose_name_plural = "Strategy obs" def __str__(self): return (str(self.name))