"""Django database model for SatNOGS DB""" import logging import re from os import path from uuid import uuid4 import satnogsdecoders from django.conf import settings from django.contrib.auth import get_user_model from django.core.cache import cache from django.core.exceptions import ValidationError from django.core.validators import MaxLengthValidator, MaxValueValidator, MinLengthValidator, \ MinValueValidator, URLValidator from django.db import models from django.db.models import OuterRef, Subquery from django.utils.timezone import now from django_countries.fields import CountryField from markdown import markdown from nanoid import generate from shortuuidfield import ShortUUIDField LOGGER = logging.getLogger('db') DATA_SOURCES = ['manual', 'network', 'sids'] SATELLITE_STATUS = ['alive', 'dead', 'future', 're-entered'] TRANSMITTER_STATUS = ['active', 'inactive', 'invalid'] TRANSMITTER_TYPE = ['Transmitter', 'Transceiver', 'Transponder'] SERVICE_TYPE = [ 'Aeronautical', 'Amateur', 'Broadcasting', 'Earth Exploration', 'Fixed', 'Inter-satellite', 'Maritime', 'Meteorological', 'Mobile', 'Radiolocation', 'Radionavigational', 'Space Operation', 'Space Research', 'Standard Frequency and Time Signal', 'Unknown' ] IARU_COORDINATION_STATUS = ['IARU Coordinated', 'IARU Declined', 'IARU Uncoordinated', 'N/A'] BAD_COORDINATIONS = ['IARU Declined', 'IARU Uncoordinated'] # 'violations' URL_REGEX = r"(?:http(s)?:\/\/)?[\w.-]+(?:\.[\w\.-]+)+[\w\-\._~:/?#[\]@!\$&'\(\)\*\+,;=.]+$" MIN_FREQ = 0 MAX_FREQ = 40000000000 MIN_FREQ_MSG = "Ensure this value is greater than or equal to 0Hz" MAX_FREQ_MSG = "Ensure this value is less than or equal to 40Ghz" def _name_exported_frames(instance, filename): # pylint: disable=W0613 """Returns path for a exported frames file""" return path.join('download/', filename) def _name_payload_frame(instance, filename): # pylint: disable=W0613 """Returns a unique, timestamped path and filename for a payload :param filename: the original filename submitted :returns: path string with timestamped subfolders and filename """ today = now() folder = 'payload_frames/{0}/{1}/{2}/'.format(today.year, today.month, today.day) ext = 'raw' filename = '{0}_{1}.{2}'.format(filename, uuid4().hex, ext) return path.join(folder, filename) class Mode(models.Model): """A satellite transmitter RF mode. For example: FM""" name = models.CharField(max_length=25, unique=True) def __str__(self): return self.name class Operator(models.Model): """Satellite Owner/Operator""" name = models.CharField(max_length=255, unique=True) names = models.TextField(blank=True) description = models.TextField(blank=True) website = models.URLField( blank=True, validators=[URLValidator(schemes=['http', 'https'], regex=URL_REGEX)] ) def __str__(self): return self.name def validate_sat_id(value): """Validate a Satellite Identifier""" if not re.compile(r'^[A-Z]{4,4}(?:-\d\d\d\d){4,4}$').match(value): raise ValidationError( '%(value)s is not a valid Satellite Identifier. Satellite Identifier should have the \ format "CCCC-NNNN-NNNN-NNNN-NNNN" where C is {A-Z} and N is {0-9}', params={'value': value}, ) def generate_sat_id(): """Generate Satellite Identifier""" numeric = "0123456789" uppercase = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" first_segment = generate(uppercase, 4) second_segment = generate(numeric, 4) third_segment = generate(numeric, 4) fourth_segment = generate(numeric, 4) fifth_segment = generate(numeric, 4) return "{0}-{1}-{2}-{3}-{4}".format( first_segment, second_segment, third_segment, fourth_segment, fifth_segment ) def get_default_itu_notification_field(): """Generate default value for itu_notification field of TransmitterEntry model""" return {'urls': []} class SatelliteIdentifier(models.Model): """Model for Satellite Identifier.""" sat_id = models.CharField( default=generate_sat_id, unique=True, max_length=24, validators=[validate_sat_id] ) created = models.DateTimeField(auto_now_add=True) def __str__(self): return '{0}'.format(self.sat_id) class SatelliteEntry(models.Model): """Model for all the satellite entries.""" satellite_identifier = models.ForeignKey( SatelliteIdentifier, null=True, related_name='satellite_entries', on_delete=models.PROTECT ) norad_cat_id = models.PositiveIntegerField(blank=True, null=True) norad_follow_id = models.PositiveIntegerField(blank=True, null=True) name = models.CharField(max_length=45) names = models.TextField(blank=True) description = models.TextField(blank=True) dashboard_url = models.URLField( blank=True, null=True, max_length=200, validators=[URLValidator(schemes=['http', 'https'], regex=URL_REGEX)] ) image = models.ImageField(upload_to='satellites', blank=True, help_text='Ideally: 250x250') status = models.CharField( choices=list(zip(SATELLITE_STATUS, SATELLITE_STATUS)), max_length=10, default='alive' ) decayed = models.DateTimeField(null=True, blank=True) # new fields below, metasat etc # countries is multiple for edge cases like ISS/Zarya countries = CountryField(blank=True, multiple=True, blank_label='(select countries)') website = models.URLField( blank=True, validators=[URLValidator(schemes=['http', 'https'], regex=URL_REGEX)] ) launched = models.DateTimeField(null=True, blank=True) deployed = models.DateTimeField(null=True, blank=True) operator = models.ForeignKey( Operator, blank=True, null=True, on_delete=models.SET_NULL, related_name='satellite_operator' ) # Fields related to suggestion reviews reviewer = models.ForeignKey( get_user_model(), related_name='reviewed_satellites', blank=True, null=True, on_delete=models.SET_NULL ) reviewed = models.DateTimeField(blank=True, null=True, help_text='Timestamp of review') approved = models.BooleanField(default=False) created = models.DateTimeField(default=now, help_text='Timestamp of creation/edit') created_by = models.ForeignKey( get_user_model(), related_name='created_satellites', null=True, on_delete=models.SET_NULL ) citation = models.CharField( max_length=512, default='CITATION NEEDED - https://xkcd.com/285/', help_text='A reference (preferrably URL) for this entry or edit' ) class Meta: ordering = ['norad_cat_id'] verbose_name_plural = 'Satellite Entries' unique_together = ("satellite_identifier", "reviewed") def get_description(self): """Returns the markdown-processed satellite description :returns: the markdown-processed satellite description """ return markdown(self.description) def get_image(self): """Returns an image for the satellite :returns: the saved image for the satellite, or a default """ if self.image and hasattr(self.image, 'url'): image = self.image.url else: image = settings.SATELLITE_DEFAULT_IMAGE return image @property def countries_str(self): """Returns countries for this Satellite in comma seperated string format :returns: countries for this Satellite in comma seperated string format """ return ','.join(map(str, self.countries)) def __str__(self): return '{0} - {1}'.format(self.norad_cat_id, self.name) class SatelliteSuggestionManager(models.Manager): # pylint: disable=R0903 """Django Manager for SatelliteSuggestions SatelliteSuggestions are SatelliteEntry objects that have been submitted (suggested) but not yet reviewed """ def get_queryset(self): # pylint: disable=R0201 """Returns SatelliteEntries that have not been reviewed""" return SatelliteEntry.objects.filter(reviewed__isnull=True) class SatelliteSuggestion(SatelliteEntry): """Proxy model for unreviewed SatelliteEntry objects""" objects = SatelliteSuggestionManager() class Meta: proxy = True permissions = ( ('approve_satellitesuggestion', 'Can approve/reject satellite suggestions'), ) class SatelliteManager(models.Manager): # pylint: disable=R0903 """Django Manager for Satellites Satellite objects rarely used without referencing their SatelliteIdentifier and SatelliteEntry foreign key fields, thus follow these relationships to get the related-objects data by using select_related(). """ def get_queryset(self): # pylint: disable=R0201 """Returns SatelliteEntries that have not been reviewed""" return super().get_queryset().select_related('satellite_identifier', 'satellite_entry') class Satellite(models.Model): """ Model for the lastest satellite entry for each Satellite Identifier.""" objects = SatelliteManager() satellite_identifier = models.OneToOneField( SatelliteIdentifier, related_name='satellite', on_delete=models.CASCADE ) satellite_entry = models.ForeignKey(SatelliteEntry, null=True, on_delete=models.SET_NULL) associated_satellite = models.ForeignKey( 'self', null=True, related_name='associated_with', on_delete=models.PROTECT ) last_modified = models.DateTimeField(auto_now=True) class Meta: permissions = (('merge_satellites', 'Can merge satellites'), ) def __str__(self): if self.satellite_entry: name = self.satellite_entry.name norad_cat_id = self.satellite_entry.norad_cat_id else: name = '-' norad_cat_id = '-' return '{1} ({2}) | {0}'.format(self.satellite_identifier.sat_id, name, norad_cat_id) @property def transmitters(self): """Returns valid transmitters for this Satellite :returns: the valid transmitters for this Satellite """ transmitters = Transmitter.objects.filter(satellite=self.id).exclude(status='invalid') return transmitters @property def transmitter_suggestion_count(self): """Returns number of pending transmitter suggestions for this Satellite :returns: number of pending transmitter suggestions for this Satellite """ pending_count = TransmitterSuggestion.objects.filter(satellite=self.id).count() return pending_count @property def satellite_suggestion_count(self): """Returns number of pending satellite suggestions for this Satellite :returns: number of pending satellite suggestions for this Satellite """ pending_count = SatelliteSuggestion.objects.filter( satellite_identifier=self.satellite_identifier ).count() return pending_count @property def telemetry_data_count(self): """Returns number of DemodData for this Satellite :returns: number of DemodData for this Satellite """ cached_satellite = cache.get(self.id) if cached_satellite: data_count = cached_satellite['count'] else: satellites_list = list(self.associated_with.all().values_list('pk', flat=True)) satellites_list.append(self.pk) data_count = DemodData.objects.filter(satellite__in=satellites_list).count() return data_count @property def telemetry_decoder_count(self): """Returns number of Telemetry objects for this Satellite :returns: number of Telemetry objects for this Satellite """ decoder_count = Telemetry.objects.filter(satellite=self.id).exclude(decoder='').count() return decoder_count @property def latest_data(self): """Returns the latest DemodData for this Satellite :returns: dict with most recent DemodData for this Satellite """ satellites_list = list(self.associated_with.all().values_list('pk', flat=True)) satellites_list.append(self.pk) data = DemodData.objects.filter(satellite__in=satellites_list).order_by('-id')[:1] if data: latest_datum = data[0] return { 'data_id': latest_datum.data_id, 'payload_frame': latest_datum.payload_frame, 'timestamp': latest_datum.timestamp, 'is_decoded': latest_datum.is_decoded, 'station': latest_datum.station, 'observer': latest_datum.observer, } return None @property def needs_help(self): """Returns a boolean based on whether or not this Satellite could use some editorial help based on a configurable threshold :returns: bool """ score = 0 if self.satellite_entry.description and self.satellite_entry.description != '': score += 1 if self.satellite_entry.countries and self.satellite_entry.countries != '': score += 1 if self.satellite_entry.website and self.satellite_entry.website != '': score += 1 if self.satellite_entry.names and self.satellite_entry.names != '': score += 1 if self.satellite_entry.launched and self.satellite_entry.launched != '': score += 1 if self.satellite_entry.operator and self.satellite_entry.operator != '': score += 1 if self.satellite_entry.image and self.satellite_entry.image != '': score += 1 return score <= 2 @property def has_bad_transmitter(self): """Returns a boolean based on whether or not this Satellite has a transmitter associated with it that is considered uncoordinated or otherwise bad :returns: bool """ violation = cache.get("violator_" + self.satellite_identifier.sat_id) if violation is not None: return violation['status'] result = False for transmitter in Transmitter.objects.filter(satellite=self.id).exclude(status='invalid'): if transmitter.bad_transmitter: result = True break cache.set( "violator_" + str(self.satellite_entry.norad_cat_id), { 'status': result, 'id': str(self.id) }, None ) cache.set( "violator_" + self.satellite_identifier.sat_id, { 'status': result, 'id': str(self.id) }, None ) for merged_satellite in self.associated_with.all(): cache.set( "violator_" + merged_satellite.satellite_identifier.sat_id, { 'status': result, 'id': str(self.id) }, None ) return result class TransmitterEntry(models.Model): """Model for satellite transmitters.""" uuid = ShortUUIDField(db_index=True) description = models.TextField( help_text='Short description for this entry, like: UHF 9k6 AFSK Telemetry' ) status = models.CharField( choices=list(zip(TRANSMITTER_STATUS, TRANSMITTER_STATUS)), max_length=8, default='active', help_text='Functional state of this transmitter' ) type = models.CharField( choices=list(zip(TRANSMITTER_TYPE, TRANSMITTER_TYPE)), max_length=11, default='Transmitter' ) uplink_low = models.BigIntegerField( blank=True, null=True, validators=[ MinValueValidator(MIN_FREQ, message=MIN_FREQ_MSG), MaxValueValidator(MAX_FREQ, message=MAX_FREQ_MSG) ], help_text='Frequency (in Hz) for the uplink, or bottom of the uplink range for a \ transponder' ) uplink_high = models.BigIntegerField( blank=True, null=True, validators=[ MinValueValidator(MIN_FREQ, message=MIN_FREQ_MSG), MaxValueValidator(MAX_FREQ, message=MAX_FREQ_MSG) ], help_text='Frequency (in Hz) for the top of the uplink range for a transponder' ) uplink_drift = models.IntegerField( blank=True, null=True, validators=[MinValueValidator(-99999), MaxValueValidator(99999)], help_text='Receiver drift from the published uplink frequency, stored in parts \ per billion (PPB)' ) downlink_low = models.BigIntegerField( blank=True, null=True, validators=[ MinValueValidator(MIN_FREQ, message=MIN_FREQ_MSG), MaxValueValidator(MAX_FREQ, message=MAX_FREQ_MSG) ], help_text='Frequency (in Hz) for the downlink, or bottom of the downlink range \ for a transponder' ) downlink_high = models.BigIntegerField( blank=True, null=True, validators=[ MinValueValidator(MIN_FREQ, message=MIN_FREQ_MSG), MaxValueValidator(MAX_FREQ, message=MAX_FREQ_MSG) ], help_text='Frequency (in Hz) for the top of the downlink range for a transponder' ) downlink_drift = models.IntegerField( blank=True, null=True, validators=[MinValueValidator(-99999), MaxValueValidator(99999)], help_text='Transmitter drift from the published downlink frequency, stored in \ parts per billion (PPB)' ) downlink_mode = models.ForeignKey( Mode, blank=True, null=True, on_delete=models.SET_NULL, related_name='transmitter_downlink_entries', help_text='Modulation mode for the downlink' ) uplink_mode = models.ForeignKey( Mode, blank=True, null=True, on_delete=models.SET_NULL, related_name='transmitter_uplink_entries', help_text='Modulation mode for the uplink' ) invert = models.BooleanField( default=False, help_text='True if this is an inverted transponder' ) baud = models.FloatField( validators=[MinValueValidator(0)], blank=True, null=True, help_text='The number of modulated symbols that the transmitter sends every second' ) satellite = models.ForeignKey( Satellite, null=True, related_name='transmitter_entries', on_delete=models.SET_NULL ) citation = models.CharField( max_length=512, default='CITATION NEEDED - https://xkcd.com/285/', help_text='A reference (preferrably URL) for this entry or edit' ) service = models.CharField( choices=zip(SERVICE_TYPE, SERVICE_TYPE), max_length=34, default='Unknown', help_text='The published usage category for this transmitter' ) iaru_coordination = models.CharField( choices=list(zip(IARU_COORDINATION_STATUS, IARU_COORDINATION_STATUS)), max_length=20, default='N/A', help_text='IARU frequency coordination status for this transmitter' ) iaru_coordination_url = models.URLField( blank=True, help_text='URL for more details on this frequency coordination', validators=[URLValidator(schemes=['http', 'https'], regex=URL_REGEX)] ) itu_notification = models.JSONField(default=get_default_itu_notification_field) reviewer = models.ForeignKey( get_user_model(), related_name='reviewed_transmitters', blank=True, null=True, on_delete=models.SET_NULL ) reviewed = models.DateTimeField(blank=True, null=True, help_text='Timestamp of review') approved = models.BooleanField(default=False) created = models.DateTimeField(default=now, help_text='Timestamp of creation/edit') created_by = models.ForeignKey( get_user_model(), related_name='created_transmitters', null=True, on_delete=models.SET_NULL ) # NOTE: future fields will need to be added to forms.py and to # api/serializers.py @property def bad_transmitter(self): """Returns a boolean that indicates whether this transmitter should be flagged as bad with regard to frequency coordination or rejection :returns: bool """ if self.iaru_coordination in BAD_COORDINATIONS: return True return False class Meta: unique_together = ("uuid", "reviewed") verbose_name_plural = 'Transmitter entries' def __str__(self): return self.description def clean(self): if self.type == TRANSMITTER_TYPE[0]: if self.uplink_low is not None or self.uplink_high is not None \ or self.uplink_drift is not None: raise ValidationError("Uplink shouldn't be filled in for a transmitter") if self.downlink_high: raise ValidationError( "Downlink high frequency shouldn't be filled in for a transmitter" ) elif self.type == TRANSMITTER_TYPE[1]: if self.uplink_high is not None or self.downlink_high is not None: raise ValidationError("Frequency range shouldn't be filled in for a transceiver") elif self.type == TRANSMITTER_TYPE[2]: if self.downlink_low is not None and self.downlink_high is not None: if self.downlink_low > self.downlink_high: raise ValidationError( "Downlink low frequency must be lower or equal \ than downlink high frequency" ) if self.uplink_low is not None and self.uplink_high is not None: if self.uplink_low > self.uplink_high: raise ValidationError( "Uplink low frequency must be lower or equal \ than uplink high frequency" ) class TransmitterSuggestionManager(models.Manager): # pylint: disable=R0903 """Django Manager for TransmitterSuggestions TransmitterSuggestions are TransmitterEntry objects that have been submitted (suggested) but not yet reviewed """ def get_queryset(self): # pylint: disable=R0201 """Returns TransmitterEntries that have not been reviewed""" return TransmitterEntry.objects.filter(reviewed__isnull=True) class TransmitterSuggestion(TransmitterEntry): """TransmitterSuggestion is an unreviewed TransmitterEntry object""" objects = TransmitterSuggestionManager() def __str__(self): return self.description class Meta: proxy = True permissions = ( ('approve_transmittersuggestion', 'Can approve/reject transmitter suggestions'), ) class TransmitterManager(models.Manager): # pylint: disable=R0903 """Django Manager for Transmitter objects""" def get_queryset(self): """Returns query of TransmitterEntries :returns: the latest revision of a TransmitterEntry for each TransmitterEntry uuid associated with this Satellite that is both reviewed and approved """ subquery = TransmitterEntry.objects.filter( reviewed__isnull=False, approved=True ).filter(uuid=OuterRef('uuid')).order_by('-reviewed') return super().get_queryset().filter( reviewed__isnull=False, approved=True ).select_related('satellite', 'downlink_mode', 'uplink_mode').filter(reviewed=Subquery(subquery.values('reviewed')[:1])) class Transmitter(TransmitterEntry): """Associates a generic Transmitter object with their TransmitterEntries that are managed by TransmitterManager """ objects = TransmitterManager() def __str__(self): return self.description class Meta: proxy = True class Tle(models.Model): """Model for TLEs.""" tle0 = models.CharField( max_length=69, blank=True, validators=[MinLengthValidator(1), MaxLengthValidator(69)] ) tle1 = models.CharField( max_length=69, blank=True, validators=[MinLengthValidator(69), MaxLengthValidator(69)] ) tle2 = models.CharField( max_length=69, blank=True, validators=[MinLengthValidator(69), MaxLengthValidator(69)] ) tle_source = models.CharField(max_length=300, blank=True) updated = models.DateTimeField(auto_now=True, blank=True) satellite = models.ForeignKey( Satellite, null=True, blank=True, related_name='tle_sets', on_delete=models.SET_NULL ) class Meta: ordering = ['-updated'] indexes = [ models.Index(fields=['-updated']), ] permissions = [('access_all_tles', 'Access all TLEs')] def __str__(self): return '{:d} - {:s}'.format(self.id, self.tle0) @property def str_array(self): """Return TLE in string array format""" # tle fields are unicode, pyephem and others expect python strings return [str(self.tle0), str(self.tle1), str(self.tle2)] class LatestTleSet(models.Model): """LatestTleSet holds the latest entry of a Satellite Tle Set""" satellite = models.OneToOneField( Satellite, related_name='latest_tle_set', on_delete=models.CASCADE ) latest = models.ForeignKey(Tle, null=True, related_name='latest', on_delete=models.SET_NULL) latest_distributable = models.ForeignKey( Tle, null=True, related_name='latest_distributable', on_delete=models.SET_NULL ) last_modified = models.DateTimeField(auto_now=True) class Telemetry(models.Model): """Model for satellite telemetry decoders.""" satellite = models.ForeignKey( Satellite, null=True, related_name='telemetries', on_delete=models.SET_NULL ) name = models.CharField(max_length=45) schema = models.TextField(blank=True) decoder = models.CharField(max_length=200, blank=True) class Meta: ordering = ['satellite__satellite_entry__norad_cat_id'] verbose_name_plural = 'Telemetries' def __str__(self): return self.name def get_kaitai_fields(self): """Return an empty-value dict of fields for this kaitai.io struct Beware the overuse of "decoder" in satnogsdecoders and "decoder" the field above in this Telemetry model""" results = {} try: decoder_class = getattr(satnogsdecoders.decoder, self.decoder.capitalize()) results = satnogsdecoders.decoder.get_fields(decoder_class, empty=True) except AttributeError: pass return results class DemodData(models.Model): """Model for satellite for observation data.""" satellite = models.ForeignKey( Satellite, null=True, related_name='telemetry_data', on_delete=models.SET_NULL ) transmitter = models.ForeignKey( TransmitterEntry, null=True, blank=True, on_delete=models.SET_NULL ) app_source = models.CharField( choices=list(zip(DATA_SOURCES, DATA_SOURCES)), max_length=7, default='sids' ) observation_id = models.IntegerField(blank=True, null=True) station_id = models.IntegerField(blank=True, null=True) data_id = models.PositiveIntegerField(blank=True, null=True) payload_frame = models.FileField(upload_to=_name_payload_frame, blank=True, null=True) payload_decoded = models.TextField(blank=True) payload_telemetry = models.ForeignKey( Telemetry, null=True, blank=True, on_delete=models.SET_NULL ) station = models.CharField(max_length=45, default='Unknown') observer = models.CharField(max_length=60, blank=True) lat = models.FloatField(validators=[MaxValueValidator(90), MinValueValidator(-90)], default=0) lng = models.FloatField( validators=[MaxValueValidator(180), MinValueValidator(-180)], default=0 ) is_decoded = models.BooleanField(default=False, db_index=True) timestamp = models.DateTimeField(null=True, db_index=True) version = models.CharField(max_length=45, blank=True) class Meta: ordering = ['-timestamp'] def __str__(self): return 'data-for-{0}'.format(self.satellite.satellite_identifier.sat_id) def display_frame(self): """Returns the contents of the saved frame file for this DemodData :returns: the contents of the saved frame file for this DemodData """ try: with open(self.payload_frame.path) as frame_file: return frame_file.read() except IOError as err: LOGGER.error( err, exc_info=True, extra={ 'payload frame path': self.payload_frame.path, } ) return None except ValueError: # unlikely to happen in prod, but if an entry is made without a file return None class ExportedFrameset(models.Model): """Model for exported frames.""" created = models.DateTimeField(auto_now_add=True) user = models.ForeignKey(get_user_model(), null=True, on_delete=models.SET_NULL) satellite = models.ForeignKey(Satellite, null=True, on_delete=models.SET_NULL) exported_file = models.FileField(upload_to=_name_exported_frames, blank=True, null=True) start = models.DateTimeField(blank=True, null=True) end = models.DateTimeField(blank=True, null=True) class Artifact(models.Model): """Model for observation artifacts.""" artifact_file = models.FileField(upload_to='artifacts/', blank=True, null=True) network_obs_id = models.BigIntegerField(blank=True, null=True) def __str__(self): return 'artifact-{0}'.format(self.id)