# Django imports from django.contrib.auth.models import User from django.contrib.gis.db import models as gis from django.contrib.gis.db.models import functions from django.core.exceptions import ValidationError from django.core.validators import MaxValueValidator, MinValueValidator from django.db import models from django.db.models import ExpressionWrapper, F from django.utils import timezone def get_default_polarization(): obj, created = Polarization.objects.get_or_create(name="-") return obj.id def get_default_modulation(): obj, created = Modulation.objects.get_or_create(name="-") return obj.id def get_default_standard(): obj, created = Standard.objects.get_or_create(name="-") return obj.id class CustomUser(models.Model): """ Расширенная модель пользователя с ролями. Добавляет систему ролей к стандартной модели User Django. """ ROLE_CHOICES = [ ("admin", "Администратор"), ("moderator", "Модератор"), ("user", "Пользователь"), ] # Связи user = models.OneToOneField( User, on_delete=models.CASCADE, verbose_name="Пользователь", help_text="Связанный пользователь Django", ) # Основные поля role = models.CharField( max_length=20, choices=ROLE_CHOICES, default="user", verbose_name="Роль пользователя", db_index=True, help_text="Роль пользователя в системе", ) def __str__(self): return ( f"{self.user.first_name} {self.user.last_name}" if self.user.first_name and self.user.last_name else self.user.username ) class Meta: verbose_name = "Пользователь" verbose_name_plural = "Пользователи" ordering = ["user__username"] class SigmaParMark(models.Model): """ Модель отметки о наличии сигнала. Используется для фиксации моментов времени когда сигнал был обнаружен или потерян. """ # Основные поля mark = models.BooleanField( null=True, blank=True, verbose_name="Наличие сигнала", help_text="True - сигнал обнаружен, False - сигнал отсутствует", ) timestamp = models.DateTimeField( null=True, blank=True, verbose_name="Время", db_index=True, help_text="Время фиксации отметки", ) def __str__(self): if self.timestamp: timestamp = self.timestamp.strftime("%d.%m.%Y %H:%M") return f"+ {timestamp}" if self.mark else f"- {timestamp}" return "Отметка без времени" class Meta: verbose_name = "Отметка" verbose_name_plural = "Отметки" ordering = ["-timestamp"] class Mirror(models.Model): """ Модель зеркала антенны. Представляет физическое зеркало антенны для приема спутникового сигнала. """ # Основные поля name = models.CharField( max_length=30, unique=True, verbose_name="Имя зеркала", db_index=True, help_text="Уникальное название зеркала антенны", ) def __str__(self): return self.name class Meta: verbose_name = "Зеркало" verbose_name_plural = "Зеркала" ordering = ["name"] class Polarization(models.Model): """ Модель поляризации сигнала. Определяет тип поляризации спутникового сигнала (H, V, L, R и т.д.). """ # Основные поля name = models.CharField( max_length=20, unique=True, verbose_name="Поляризация", db_index=True, help_text="Тип поляризации (H - горизонтальная, V - вертикальная, L - левая круговая, R - правая круговая)", ) def __str__(self): return self.name class Meta: verbose_name = "Поляризация" verbose_name_plural = "Поляризация" ordering = ["name"] class Modulation(models.Model): """ Модель типа модуляции сигнала. Определяет схему модуляции (QPSK, 8PSK, 16APSK и т.д.). """ # Основные поля name = models.CharField( max_length=20, unique=True, verbose_name="Модуляция", db_index=True, help_text="Тип модуляции сигнала (QPSK, 8PSK, 16APSK и т.д.)", ) def __str__(self): return self.name class Meta: verbose_name = "Модуляция" verbose_name_plural = "Модуляции" ordering = ["name"] class Standard(models.Model): """ Модель стандарта передачи данных. Определяет стандарт передачи (DVB-S, DVB-S2, DVB-S2X и т.д.). """ # Основные поля name = models.CharField( max_length=20, unique=True, verbose_name="Стандарт", db_index=True, help_text="Стандарт передачи данных (DVB-S, DVB-S2, DVB-S2X и т.д.)", ) def __str__(self): return self.name class Meta: verbose_name = "Стандарт" verbose_name_plural = "Стандарты" ordering = ["name"] class Satellite(models.Model): """ Модель спутника. Представляет спутник связи с его основными характеристиками. """ # Основные поля name = models.CharField( max_length=100, unique=True, verbose_name="Имя спутника", db_index=True, help_text="Название спутника", ) norad = models.IntegerField( blank=True, null=True, verbose_name="NORAD ID", help_text="Идентификатор NORAD для отслеживания спутника", ) def __str__(self): return self.name class Meta: verbose_name = "Спутник" verbose_name_plural = "Спутники" ordering = ["name"] class ObjItemQuerySet(models.QuerySet): """Custom QuerySet для модели ObjItem с оптимизированными запросами""" def with_related(self): """Оптимизирует запросы, загружая связанные объекты""" return self.select_related( "geo_obj", "updated_by__user", "created_by__user", "lyngsat_source", "parameter_obj", "parameter_obj__id_satellite", "parameter_obj__polarization", "parameter_obj__modulation", "parameter_obj__standard", ) def recent(self, days=30): """Возвращает объекты, созданные за последние N дней""" from datetime import timedelta cutoff_date = timezone.now() - timedelta(days=days) return self.filter(created_at__gte=cutoff_date) def by_user(self, user): """Возвращает объекты, созданные указанным пользователем""" return self.filter(created_by=user) class ObjItemManager(models.Manager): """Custom Manager для модели ObjItem""" def get_queryset(self): return ObjItemQuerySet(self.model, using=self._db) def with_related(self): """Возвращает queryset с предзагруженными связанными объектами""" return self.get_queryset().with_related() def recent(self, days=30): """Возвращает недавно созданные объекты""" return self.get_queryset().recent(days) def by_user(self, user): """Возвращает объекты пользователя""" return self.get_queryset().by_user(user) class ObjItem(models.Model): """ Модель объекта (источника сигнала). Центральная модель, объединяющая информацию о ВЧ параметрах, геолокации и типе источника. """ # Основные поля name = models.CharField( null=True, blank=True, max_length=100, verbose_name="Имя объекта", db_index=True, help_text="Название объекта/источника сигнала", ) # Метаданные created_at = models.DateTimeField( auto_now_add=True, verbose_name="Дата создания", help_text="Дата и время создания записи", ) created_by = models.ForeignKey( CustomUser, on_delete=models.SET_NULL, related_name="objitems_created", null=True, blank=True, verbose_name="Создан пользователем", help_text="Пользователь, создавший запись", ) updated_at = models.DateTimeField( auto_now=True, verbose_name="Дата последнего изменения", help_text="Дата и время последнего изменения", ) updated_by = models.ForeignKey( CustomUser, on_delete=models.SET_NULL, related_name="objitems_updated", null=True, blank=True, verbose_name="Изменен пользователем", help_text="Пользователь, последним изменивший запись", ) lyngsat_source = models.ForeignKey( "lyngsatapp.LyngSat", on_delete=models.SET_NULL, related_name="objitems", null=True, blank=True, verbose_name="Источник LyngSat", help_text="Связанный источник из базы LyngSat (ТВ)", ) # Custom manager objects = ObjItemManager() def __str__(self): return f"Объект {self.name}" if self.name else f"Объект #{self.pk}" class Meta: verbose_name = "Объект" verbose_name_plural = "Объекты" ordering = ["-updated_at"] indexes = [ models.Index(fields=["name"]), models.Index(fields=["-updated_at"]), models.Index(fields=["-created_at"]), ] class Parameter(models.Model): id_satellite = models.ForeignKey( Satellite, on_delete=models.PROTECT, related_name="parameters", verbose_name="Спутник", null=True, ) polarization = models.ForeignKey( Polarization, default=get_default_polarization, on_delete=models.SET_DEFAULT, related_name="polarizations", null=True, blank=True, verbose_name="Поляризация", ) frequency = models.FloatField( default=0, null=True, blank=True, verbose_name="Частота, МГц", db_index=True, # validators=[MinValueValidator(0), MaxValueValidator(50000)], help_text="Центральная частота сигнала", ) freq_range = models.FloatField( default=0, null=True, blank=True, verbose_name="Полоса частот, МГц", # validators=[MinValueValidator(0), MaxValueValidator(1000)], help_text="Полоса частот сигнала", ) bod_velocity = models.FloatField( default=0, null=True, blank=True, verbose_name="Символьная скорость, БОД", # validators=[MinValueValidator(0)], help_text="Символьная скорость должна быть положительной", ) modulation = models.ForeignKey( Modulation, default=get_default_modulation, on_delete=models.SET_DEFAULT, related_name="modulations", null=True, blank=True, verbose_name="Модуляция", ) snr = models.FloatField( default=0, null=True, blank=True, verbose_name="ОСШ", # validators=[MinValueValidator(-50), MaxValueValidator(100)], help_text="Отношение сигнал/шум", ) standard = models.ForeignKey( Standard, default=get_default_standard, on_delete=models.SET_DEFAULT, related_name="standards", null=True, blank=True, verbose_name="Стандарт", ) # id_user_add = models.ForeignKey(CustomUser, on_delete=models.SET_NULL, related_name="parameter_added", verbose_name="Пользователь", null=True, blank=True) objitem = models.OneToOneField( ObjItem, on_delete=models.CASCADE, related_name="parameter_obj", verbose_name="Объект", null=True, blank=True, help_text="Связанный объект" ) # id_sigma_parameter = models.ManyToManyField(SigmaParameter, on_delete=models.SET_NULL, related_name="sigma_parameter", verbose_name="ВЧ с sigma", null=True, blank=True) # id_sigma_parameter = models.ManyToManyField(SigmaParameter, verbose_name="ВЧ с sigma", null=True, blank=True) def clean(self): """Валидация на уровне модели""" super().clean() # Проверка что частота больше полосы частот if self.frequency and self.freq_range: if self.freq_range > self.frequency: raise ValidationError( {"freq_range": "Полоса частот не может быть больше частоты"} ) # Проверка что символьная скорость соответствует полосе частот if self.bod_velocity and self.freq_range: if self.bod_velocity > self.freq_range * 1000000: # Конвертация МГц в Гц raise ValidationError( { "bod_velocity": "Символьная скорость не может превышать полосу частот" } ) def __str__(self): polarization_name = self.polarization.name if self.polarization else "-" modulation_name = self.modulation.name if self.modulation else "-" return f"Источник-{self.frequency}:{self.freq_range} МГц:{polarization_name}:{modulation_name}" class Meta: verbose_name = "ВЧ загрузка" verbose_name_plural = "ВЧ загрузки" indexes = [ models.Index(fields=["id_satellite", "frequency"]), models.Index(fields=["frequency", "polarization"]), ] # constraints = [ # models.UniqueConstraint( # fields=[ # 'polarization', 'frequency', 'freq_range', # 'bod_velocity', 'modulation', 'snr', 'standard' # ], # name='unique_parameter_combination' # ) # ] class SigmaParameter(models.Model): TRANSFERS = [(-1.0, "-"), (9750.0, "9750 МГц"), (10750.0, "10750 МГц")] id_satellite = models.ForeignKey( Satellite, on_delete=models.PROTECT, related_name="sigmapar_sat", verbose_name="Спутник", ) transfer = models.FloatField( choices=TRANSFERS, default=-1.0, verbose_name="Перенос по частоте", help_text="Выберите перенос по частоте", ) status = models.CharField( max_length=20, blank=True, null=True, verbose_name="Статус", help_text="Статус измерения", ) frequency = models.FloatField( default=0, null=True, blank=True, verbose_name="Частота, МГц", db_index=True, # validators=[MinValueValidator(0), MaxValueValidator(50000)], help_text="Центральная частота сигнала", ) transfer_frequency = models.GeneratedField( expression=ExpressionWrapper( F("frequency") + F("transfer"), output_field=models.FloatField() ), output_field=models.FloatField(), db_persist=True, null=True, blank=True, verbose_name="Частота в Ku, МГц", ) freq_range = models.FloatField( default=0, null=True, blank=True, verbose_name="Полоса частот, МГц", # validators=[MinValueValidator(0), MaxValueValidator(1000)], help_text="Полоса частот", ) power = models.FloatField( default=0, null=True, blank=True, verbose_name="Мощность, дБм", # validators=[MinValueValidator(-100), MaxValueValidator(100)], help_text="Мощность сигнала", ) bod_velocity = models.FloatField( default=0, null=True, blank=True, verbose_name="Символьная скорость, БОД", # validators=[MinValueValidator(0)], help_text="Символьная скорость должна быть положительной", ) polarization = models.ForeignKey( Polarization, default=get_default_polarization, on_delete=models.SET_DEFAULT, related_name="polarizations_sigma", null=True, blank=True, verbose_name="Поляризация", ) modulation = models.ForeignKey( Modulation, default=get_default_modulation, on_delete=models.SET_DEFAULT, related_name="modulations_sigma", null=True, blank=True, verbose_name="Модуляция", ) snr = models.FloatField( default=0, null=True, blank=True, verbose_name="ОСШ, Дб", validators=[MinValueValidator(-50), MaxValueValidator(100)], help_text="Отношение сигнал/шум в диапазоне от -50 до 100 дБ", ) standard = models.ForeignKey( Standard, default=get_default_standard, on_delete=models.SET_DEFAULT, related_name="standards_sigma", null=True, blank=True, verbose_name="Стандарт", ) packets = models.BooleanField( null=True, blank=True, verbose_name="Пакетность", help_text="Наличие пакетной передачи", ) datetime_begin = models.DateTimeField( null=True, blank=True, verbose_name="Время начала измерения", help_text="Дата и время начала измерения", ) datetime_end = models.DateTimeField( null=True, blank=True, verbose_name="Время окончания измерения", help_text="Дата и время окончания измерения", ) mark = models.ManyToManyField(SigmaParMark, verbose_name="Отметка", blank=True) parameter = models.ForeignKey( Parameter, on_delete=models.SET_NULL, related_name="sigma_parameter", verbose_name="ВЧ", null=True, blank=True, ) def clean(self): """Валидация на уровне модели""" super().clean() # Проверка что время окончания больше времени начала if self.datetime_begin and self.datetime_end: if self.datetime_end < self.datetime_begin: raise ValidationError( {"datetime_end": "Время окончания должно быть позже времени начала"} ) # Проверка что частота больше полосы частот if self.frequency and self.freq_range: if self.freq_range > self.frequency: raise ValidationError( {"freq_range": "Полоса частот не может быть больше частоты"} ) def __str__(self): modulation_name = self.modulation.name if self.modulation else "-" return f"Sigma-{self.frequency}:{self.freq_range} МГц:{modulation_name}" class Meta: verbose_name = "ВЧ sigma" verbose_name_plural = "ВЧ sigma" class Geo(models.Model): """ Модель геолокационных данных. Хранит информацию о местоположении источника сигнала, включая координаты, данные от различных источников (геолокация, кубсат, оперативники) и расстояния между ними. """ # Основные поля timestamp = models.DateTimeField( null=True, blank=True, verbose_name="Время", db_index=True, help_text="Время фиксации геолокации", ) location = models.CharField( max_length=255, null=True, blank=True, verbose_name="Местоположение", help_text="Текстовое описание местоположения", ) comment = models.CharField( max_length=255, blank=True, verbose_name="Комментарий", help_text="Дополнительные комментарии", ) is_average = models.BooleanField( null=True, blank=True, verbose_name="Усреднённое", help_text="Является ли координата усредненной", ) # Координаты coords = gis.PointField( srid=4326, null=True, blank=True, verbose_name="Координата геолокации", help_text="Основные координаты геолокации (WGS84)", ) coords_kupsat = gis.PointField( srid=4326, null=True, blank=True, verbose_name="Координаты Кубсата", help_text="Координаты, полученные от кубсата (WGS84)", ) coords_valid = gis.PointField( srid=4326, null=True, blank=True, verbose_name="Координаты оперативников", help_text="Координаты, предоставленные оперативным отделом (WGS84)", ) # Вычисляемые поля - расстояния distance_coords_kup = models.GeneratedField( expression=functions.Distance("coords", "coords_kupsat") / 1000, output_field=models.FloatField(), db_persist=True, null=True, blank=True, verbose_name="Расстояние между кубсатом и гео, км", ) distance_coords_valid = models.GeneratedField( expression=functions.Distance("coords", "coords_valid") / 1000, output_field=models.FloatField(), db_persist=True, null=True, blank=True, verbose_name="Расстояние между гео и оперативным отделом, км", ) distance_kup_valid = models.GeneratedField( expression=functions.Distance("coords_valid", "coords_kupsat") / 1000, output_field=models.FloatField(), db_persist=True, null=True, blank=True, verbose_name="Расстояние между кубсатом и оперативным отделом, км", ) # Связи mirrors = models.ManyToManyField( Mirror, related_name="geo_mirrors", verbose_name="Зеркала", blank=True, help_text="Зеркала антенн, использованные для приема", ) objitem = models.OneToOneField( ObjItem, on_delete=models.CASCADE, verbose_name="Объект", related_name="geo_obj", null=True, help_text="Связанный объект", ) def __str__(self): if self.coords: longitude = self.coords.coords[0] latitude = self.coords.coords[1] lon = f"{longitude}E" if longitude > 0 else f"{abs(longitude)}W" lat = f"{latitude}N" if latitude > 0 else f"{abs(latitude)}S" location_str = f", {self.location}" if self.location else "" return f"{lat} {lon}{location_str}" return f"Гео #{self.pk}" class Meta: verbose_name = "Гео" verbose_name_plural = "Гео" ordering = ["-timestamp"] indexes = [ models.Index(fields=["-timestamp"]), models.Index(fields=["location"]), ] constraints = [ models.UniqueConstraint( fields=["timestamp", "coords"], name="unique_geo_combination" ) ]