geralt/dbstorage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Процесс переделки

Browse Source
This commit is contained in:
Кошкин Сергей
2025-11-12 17:53:25 +03:00
parent 73ce06deec
commit 7126974aed
21 changed files with 1927 additions and 1922 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
dbapp/mapsapp/migrations/0001_initial.py
View File

@@ -1,37 +1,44 @@
# Generated by Django 5.2.7 on 2025-10-31 13:36
import django.db.models.deletion
import django.db.models.expressions
import django.db.models.functions.math
import mainapp.models
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('mainapp', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='Transponders',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(blank=True, max_length=30, null=True, verbose_name='Название транспондера')),
('downlink', models.FloatField(blank=True, null=True, verbose_name='Downlink')),
('frequency_range', models.FloatField(blank=True, null=True, verbose_name='Полоса')),
('uplink', models.FloatField(blank=True, null=True, verbose_name='Uplink')),
('zone_name', models.CharField(blank=True, max_length=255, null=True, verbose_name='Название зоны')),
('transfer', models.GeneratedField(db_persist=True, expression=models.ExpressionWrapper(django.db.models.functions.math.Abs(django.db.models.expressions.CombinedExpression(models.F('downlink'), '-', models.F('uplink'))), output_field=models.FloatField()), null=True, output_field=models.FloatField(), verbose_name='Перенос')),
('polarization', models.ForeignKey(blank=True, default=mainapp.models.get_default_polarization, null=True, on_delete=django.db.models.deletion.SET_DEFAULT, related_name='tran_polarizations', to='mainapp.polarization', verbose_name='Поляризация')),
('sat_id', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='tran_satellite', to='mainapp.satellite', verbose_name='Спутник')),
],
options={
'verbose_name': 'Транспондер',
'verbose_name_plural': 'Транспондеры',
},
),
]
# Generated by Django 5.2.7 on 2025-11-12 14:21
import django.db.models.deletion
import django.db.models.expressions
import django.db.models.functions.math
import mainapp.models
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('mainapp', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='Transponders',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(blank=True, db_index=True, help_text='Название транспондера', max_length=30, null=True, verbose_name='Название транспондера')),
('downlink', models.FloatField(blank=True, null=True, verbose_name='Downlink')),
('frequency_range', models.FloatField(blank=True, null=True, verbose_name='Полоса')),
('uplink', models.FloatField(blank=True, null=True, verbose_name='Uplink')),
('zone_name', models.CharField(blank=True, db_index=True, help_text='Название зоны покрытия транспондера', max_length=255, null=True, verbose_name='Название зоны')),
('snr', models.FloatField(blank=True, help_text='Полоса частот в МГц (0-1000)', null=True, verbose_name='Полоса')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='Дата и время создания записи', verbose_name='Дата создания')),
('updated_at', models.DateTimeField(auto_now=True, help_text='Дата и время последнего изменения', verbose_name='Дата последнего изменения')),
('transfer', models.GeneratedField(db_persist=True, expression=models.ExpressionWrapper(django.db.models.functions.math.Abs(django.db.models.expressions.CombinedExpression(models.F('downlink'), '-', models.F('uplink'))), output_field=models.FloatField()), null=True, output_field=models.FloatField(), verbose_name='Перенос')),
('created_by', models.ForeignKey(blank=True, help_text='Пользователь, создавший запись', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='transponder_created', to='mainapp.customuser', verbose_name='Создан пользователем')),
('polarization', models.ForeignKey(blank=True, default=mainapp.models.get_default_polarization, help_text='Поляризация сигнала', null=True, on_delete=django.db.models.deletion.SET_DEFAULT, related_name='tran_polarizations', to='mainapp.polarization', verbose_name='Поляризация')),
('sat_id', models.ForeignKey(help_text='Спутник, которому принадлежит транспондер', on_delete=django.db.models.deletion.PROTECT, related_name='tran_satellite', to='mainapp.satellite', verbose_name='Спутник')),
('updated_by', models.ForeignKey(blank=True, help_text='Пользователь, последним изменивший запись', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='transponder_updated', to='mainapp.customuser', verbose_name='Изменен пользователем')),
],
options={
'verbose_name': 'Транспондер',
'verbose_name_plural': 'Транспондеры',
'ordering': ['sat_id', 'downlink'],
'indexes': [models.Index(fields=['sat_id', 'downlink'], name='mapsapp_tra_sat_id__3e3fd7_idx'), models.Index(fields=['sat_id', 'zone_name'], name='mapsapp_tra_sat_id__305ae7_idx')],
},
),
]

64
dbapp/mapsapp/migrations/0002_alter_transponders_options_and_more.py
View File

@@ -1,64 +0,0 @@
# Generated by Django 5.2.7 on 2025-11-07 20:58
import django.core.validators
import django.db.models.deletion
import mainapp.models
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('mainapp', '0006_alter_customuser_options_alter_geo_options_and_more'),
('mapsapp', '0001_initial'),
]
operations = [
migrations.AlterModelOptions(
name='transponders',
options={'ordering': ['sat_id', 'downlink'], 'verbose_name': 'Транспондер', 'verbose_name_plural': 'Транспондеры'},
),
migrations.AlterField(
model_name='transponders',
name='downlink',
field=models.FloatField(blank=True, help_text='Частота downlink в МГц (0-50000)', null=True, validators=[django.core.validators.MinValueValidator(0), django.core.validators.MaxValueValidator(50000)], verbose_name='Downlink'),
),
migrations.AlterField(
model_name='transponders',
name='frequency_range',
field=models.FloatField(blank=True, help_text='Полоса частот в МГц (0-1000)', null=True, validators=[django.core.validators.MinValueValidator(0), django.core.validators.MaxValueValidator(1000)], verbose_name='Полоса'),
),
migrations.AlterField(
model_name='transponders',
name='name',
field=models.CharField(blank=True, db_index=True, help_text='Название транспондера', max_length=30, null=True, verbose_name='Название транспондера'),
),
migrations.AlterField(
model_name='transponders',
name='polarization',
field=models.ForeignKey(blank=True, default=mainapp.models.get_default_polarization, help_text='Поляризация сигнала', null=True, on_delete=django.db.models.deletion.SET_DEFAULT, related_name='tran_polarizations', to='mainapp.polarization', verbose_name='Поляризация'),
),
migrations.AlterField(
model_name='transponders',
name='sat_id',
field=models.ForeignKey(help_text='Спутник, которому принадлежит транспондер', on_delete=django.db.models.deletion.PROTECT, related_name='tran_satellite', to='mainapp.satellite', verbose_name='Спутник'),
),
migrations.AlterField(
model_name='transponders',
name='uplink',
field=models.FloatField(blank=True, help_text='Частота uplink в МГц (0-50000)', null=True, validators=[django.core.validators.MinValueValidator(0), django.core.validators.MaxValueValidator(50000)], verbose_name='Uplink'),
),
migrations.AlterField(
model_name='transponders',
name='zone_name',
field=models.CharField(blank=True, db_index=True, help_text='Название зоны покрытия транспондера', max_length=255, null=True, verbose_name='Название зоны'),
),
migrations.AddIndex(
model_name='transponders',
index=models.Index(fields=['sat_id', 'downlink'], name='mapsapp_tra_sat_id__3e3fd7_idx'),
),
migrations.AddIndex(
model_name='transponders',
index=models.Index(fields=['sat_id', 'zone_name'], name='mapsapp_tra_sat_id__305ae7_idx'),
),
]

69
dbapp/mapsapp/models.py
View File

@@ -6,7 +6,7 @@ from django.db.models import ExpressionWrapper, F
from django.db.models.functions import Abs
# Local imports
from mainapp.models import Polarization, Satellite, get_default_polarization
from mainapp.models import Polarization, Satellite, get_default_polarization, CustomUser
class Transponders(models.Model):
@@ -29,22 +29,22 @@ class Transponders(models.Model):
blank=True,
null=True,
verbose_name="Downlink",
validators=[MinValueValidator(0), MaxValueValidator(50000)],
help_text="Частота downlink в МГц (0-50000)"
# validators=[MinValueValidator(0), MaxValueValidator(50000)],
# help_text="Частота downlink в МГц (0-50000)"
)
frequency_range = models.FloatField(
blank=True,
null=True,
verbose_name="Полоса",
validators=[MinValueValidator(0), MaxValueValidator(1000)],
help_text="Полоса частот в МГц (0-1000)"
# validators=[MinValueValidator(0), MaxValueValidator(1000)],
# help_text="Полоса частот в МГц (0-1000)"
)
uplink = models.FloatField(
blank=True,
null=True,
verbose_name="Uplink",
validators=[MinValueValidator(0), MaxValueValidator(50000)],
help_text="Частота uplink в МГц (0-50000)"
# validators=[MinValueValidator(0), MaxValueValidator(50000)],
# help_text="Частота uplink в МГц (0-50000)"
)
zone_name = models.CharField(
max_length=255,
@@ -54,6 +54,41 @@ class Transponders(models.Model):
db_index=True,
help_text="Название зоны покрытия транспондера"
)
snr = models.FloatField(
blank=True,
null=True,
verbose_name="Полоса",
# validators=[MinValueValidator(0), MaxValueValidator(1000)],
help_text="Полоса частот в МГц (0-1000)"
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name="Дата создания",
help_text="Дата и время создания записи",
)
created_by = models.ForeignKey(
CustomUser,
on_delete=models.SET_NULL,
related_name="transponder_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="transponder_updated",
null=True,
blank=True,
verbose_name="Изменен пользователем",
help_text="Пользователь, последним изменивший запись",
)
# Связи
polarization = models.ForeignKey(
@@ -88,17 +123,17 @@ class Transponders(models.Model):
verbose_name="Перенос"
)
def clean(self):
"""Валидация на уровне модели"""
super().clean()
# def clean(self):
# """Валидация на уровне модели"""
# super().clean()
# Проверка что downlink и uplink заданы
if self.downlink and self.uplink:
# Обычно uplink выше downlink для спутниковой связи
if self.uplink < self.downlink:
raise ValidationError({
'uplink': 'Частота uplink обычно выше частоты downlink'
})
# # Проверка что downlink и uplink заданы
# if self.downlink and self.uplink:
# # Обычно uplink выше downlink для спутниковой связи
# if self.uplink < self.downlink:
# raise ValidationError({
# 'uplink': 'Частота uplink обычно выше частоты downlink'
# })
def __str__(self):
if self.name:

334
dbapp/mapsapp/utils.py
View File

@@ -1,165 +1,169 @@
# Standard library imports
import json
import re
from io import BytesIO
# Third-party imports
import requests
# Local imports
from mainapp.models import Polarization, Satellite
from .models import Transponders
def search_satellite_on_page(data: dict, satellite_name: str):
for pos, value in data.get('page', {}).get('positions').items():
for name in value['satellites']:
if name['other_names'] is None:
name['other_names'] = ''
if satellite_name.lower() in name['name'].lower() or satellite_name.lower() in name['other_names'].lower():
return pos, name['id']
return '', ''
def get_footprint_data(position: str = 62) -> dict:
"""Возвращает словарь с данным по footprint для спутников на выбранной долготе"""
response = requests.get(f"https://www.satbeams.com/footprints?position={position}")
response.raise_for_status()
match = re.search(r'var data = ({.*?});', response.text, re.DOTALL)
if match:
json_str = match.group(1)
try:
data = json.loads(json_str)
return data.get("page", {}).get("footprint_data", {}).get("beams",[])
except json.JSONDecodeError as e:
print("Ошибка парсинга JSON:", e)
else:
print("Нужных данных не найдено")
return {}
def get_all_page_data(url:str = 'https://www.satbeams.com/footprints') -> dict:
"""Возвращает словарь с данными по всем спутникам на странице"""
response = requests.get(url)
response.raise_for_status()
match = re.search(r'var data = ({.*?});', response.text, re.DOTALL)
if match:
json_str = match.group(1)
try:
data = json.loads(json_str)
# Файл json на диске для достоверности
with open('data.json', 'w') as jf:
json.dump(data, jf, indent=2)
return data
except json.JSONDecodeError as e:
print("Ошибка парсинга JSON:", e)
else:
print("Нужных данных не найдено")
return {}
def get_names_footprints_for_satellite(footprint_data: dict, sat_id: str) -> list[str]:
names = []
for beam in footprint_data:
if 'ku' in beam['band'].lower() and sat_id in beam['satellite_id']:
names.append(
{
"name": beam['name'],
"fullname": beam['fullname'][8:]
}
)
return names
def get_band_names(satellite_name: str) -> list[str]:
data = get_all_page_data()
pos, sat_id = search_satellite_on_page(data, satellite_name)
footprints = get_footprint_data(pos)
names = get_names_footprints_for_satellite(footprints, sat_id)
return names
def parse_transponders_from_json(filepath: str):
with open(filepath, encoding="utf-8") as jf:
data = json.load(jf)
for sat_name, trans_zone in data["satellites"].items():
for zone, trans in trans_zone.items():
for tran in trans:
f_b, f_e = tran["freq"][0].split("-")
f = round((float(f_b) + float(f_e))/2, 3)
f_range = round(abs(float(f_e) - float(f_b)), 3)
tran_obj = Transponders.objects.create(
name=tran["name"],
frequency=f,
frequency_range=f_range,
zone_name=zone,
polarization=Polarization.objects.get(name=tran["pol"]),
sat_id=Satellite.objects.get(name__iexact=sat_name)
)
tran_obj.save()
# Third-party imports (additional)
from lxml import etree
def parse_transponders_from_xml(data_in: BytesIO):
tree = etree.parse(data_in)
ns = {
'i': 'http://www.w3.org/2001/XMLSchema-instance',
'ns': 'http://schemas.datacontract.org/2004/07/Geolocation.Domain.Utils.Repository.SatellitesSerialization.Memos',
'tr': 'http://schemas.datacontract.org/2004/07/Geolocation.Common.Extensions'
}
satellites = tree.xpath('//ns:SatelliteMemo', namespaces=ns)
for sat in satellites[:]:
name = sat.xpath('./ns:name/text()', namespaces=ns)[0]
if name == 'X' or 'DONT USE' in name:
continue
norad = sat.xpath('./ns:norad/text()', namespaces=ns)
beams = sat.xpath('.//ns:BeamMemo', namespaces=ns)
zones = {}
for zone in beams:
zone_name = zone.xpath('./ns:name/text()', namespaces=ns)[0] if zone.xpath('./ns:name/text()', namespaces=ns) else '-'
zones[zone.xpath('./ns:id/text()', namespaces=ns)[0]] = {
"name": zone_name,
"pol": zone.xpath('./ns:polarization/text()', namespaces=ns)[0],
}
transponders = sat.xpath('.//ns:TransponderMemo', namespaces=ns)
for transponder in transponders:
tr_id = transponder.xpath('./ns:downlinkBeamId/text()', namespaces=ns)[0]
downlink_start = float(transponder.xpath('./ns:downlinkFrequency/tr:start/text()', namespaces=ns)[0])
downlink_end = float(transponder.xpath('./ns:downlinkFrequency/tr:end/text()', namespaces=ns)[0])
uplink_start = float(transponder.xpath('./ns:uplinkFrequency/tr:start/text()', namespaces=ns)[0])
uplink_end = float(transponder.xpath('./ns:uplinkFrequency/tr:end/text()', namespaces=ns)[0])
tr_data = zones[tr_id]
# p = tr_data['pol'][0] if tr_data['pol'] else '-'
match tr_data['pol']:
case 'Horizontal':
pol = 'Горизонтальная'
case 'Vertical':
pol = 'Вертикальная'
case 'CircularRight':
pol = 'Правая'
case 'CircularLeft':
pol = 'Левая'
case _:
pol = '-'
tr_name = transponder.xpath('./ns:name/text()', namespaces=ns)[0]
pol_obj, _ = Polarization.objects.get_or_create(name=pol)
sat_obj, _ = Satellite.objects.get_or_create(
name=name,
defaults={
"norad": int(norad[0]) if norad else -1
})
trans_obj, _ = Transponders.objects.get_or_create(
polarization=pol_obj,
downlink=(downlink_start+downlink_end)/2/1000000,
uplink=(uplink_start+uplink_end)/2/1000000,
frequency_range=abs(downlink_end-downlink_start)/1000000,
name=tr_name,
defaults={
"zone_name": tr_data['name'],
"sat_id": sat_obj,
}
)
trans_obj.save()
# Standard library imports
import json
import re
from io import BytesIO
# Third-party imports
import requests
# Local imports
from mainapp.models import Polarization, Satellite
from .models import Transponders
def search_satellite_on_page(data: dict, satellite_name: str):
for pos, value in data.get('page', {}).get('positions').items():
for name in value['satellites']:
if name['other_names'] is None:
name['other_names'] = ''
if satellite_name.lower() in name['name'].lower() or satellite_name.lower() in name['other_names'].lower():
return pos, name['id']
return '', ''
def get_footprint_data(position: str = 62) -> dict:
"""Возвращает словарь с данным по footprint для спутников на выбранной долготе"""
response = requests.get(f"https://www.satbeams.com/footprints?position={position}")
response.raise_for_status()
match = re.search(r'var data = ({.*?});', response.text, re.DOTALL)
if match:
json_str = match.group(1)
try:
data = json.loads(json_str)
return data.get("page", {}).get("footprint_data", {}).get("beams",[])
except json.JSONDecodeError as e:
print("Ошибка парсинга JSON:", e)
else:
print("Нужных данных не найдено")
return {}
def get_all_page_data(url:str = 'https://www.satbeams.com/footprints') -> dict:
"""Возвращает словарь с данными по всем спутникам на странице"""
response = requests.get(url)
response.raise_for_status()
match = re.search(r'var data = ({.*?});', response.text, re.DOTALL)
if match:
json_str = match.group(1)
try:
data = json.loads(json_str)
# Файл json на диске для достоверности
with open('data.json', 'w') as jf:
json.dump(data, jf, indent=2)
return data
except json.JSONDecodeError as e:
print("Ошибка парсинга JSON:", e)
else:
print("Нужных данных не найдено")
return {}
def get_names_footprints_for_satellite(footprint_data: dict, sat_id: str) -> list[str]:
names = []
for beam in footprint_data:
if 'ku' in beam['band'].lower() and sat_id in beam['satellite_id']:
names.append(
{
"name": beam['name'],
"fullname": beam['fullname'][8:]
}
)
return names
def get_band_names(satellite_name: str) -> list[str]:
data = get_all_page_data()
pos, sat_id = search_satellite_on_page(data, satellite_name)
footprints = get_footprint_data(pos)
names = get_names_footprints_for_satellite(footprints, sat_id)
return names
def parse_transponders_from_json(filepath: str):
with open(filepath, encoding="utf-8") as jf:
data = json.load(jf)
for sat_name, trans_zone in data["satellites"].items():
for zone, trans in trans_zone.items():
for tran in trans:
f_b, f_e = tran["freq"][0].split("-")
f = round((float(f_b) + float(f_e))/2, 3)
f_range = round(abs(float(f_e) - float(f_b)), 3)
tran_obj = Transponders.objects.create(
name=tran["name"],
frequency=f,
frequency_range=f_range,
zone_name=zone,
polarization=Polarization.objects.get(name=tran["pol"]),
sat_id=Satellite.objects.get(name__iexact=sat_name)
)
tran_obj.save()
# Third-party imports (additional)
from lxml import etree
def parse_transponders_from_xml(data_in: BytesIO, user=None):
tree = etree.parse(data_in)
ns = {
'i': 'http://www.w3.org/2001/XMLSchema-instance',
'ns': 'http://schemas.datacontract.org/2004/07/Geolocation.Domain.Utils.Repository.SatellitesSerialization.Memos',
'tr': 'http://schemas.datacontract.org/2004/07/Geolocation.Common.Extensions'
}
satellites = tree.xpath('//ns:SatelliteMemo', namespaces=ns)
for sat in satellites[:]:
name = sat.xpath('./ns:name/text()', namespaces=ns)[0]
if name == 'X' or 'DONT USE' in name:
continue
norad = sat.xpath('./ns:norad/text()', namespaces=ns)
beams = sat.xpath('.//ns:BeamMemo', namespaces=ns)
zones = {}
for zone in beams:
zone_name = zone.xpath('./ns:name/text()', namespaces=ns)[0] if zone.xpath('./ns:name/text()', namespaces=ns) else '-'
zones[zone.xpath('./ns:id/text()', namespaces=ns)[0]] = {
"name": zone_name,
"pol": zone.xpath('./ns:polarization/text()', namespaces=ns)[0],
}
transponders = sat.xpath('.//ns:TransponderMemo', namespaces=ns)
for transponder in transponders:
tr_id = transponder.xpath('./ns:downlinkBeamId/text()', namespaces=ns)[0]
downlink_start = float(transponder.xpath('./ns:downlinkFrequency/tr:start/text()', namespaces=ns)[0])
downlink_end = float(transponder.xpath('./ns:downlinkFrequency/tr:end/text()', namespaces=ns)[0])
uplink_start = float(transponder.xpath('./ns:uplinkFrequency/tr:start/text()', namespaces=ns)[0])
uplink_end = float(transponder.xpath('./ns:uplinkFrequency/tr:end/text()', namespaces=ns)[0])
tr_data = zones[tr_id]
# p = tr_data['pol'][0] if tr_data['pol'] else '-'
match tr_data['pol']:
case 'Horizontal':
pol = 'Горизонтальная'
case 'Vertical':
pol = 'Вертикальная'
case 'CircularRight':
pol = 'Правая'
case 'CircularLeft':
pol = 'Левая'
case _:
pol = '-'
tr_name = transponder.xpath('./ns:name/text()', namespaces=ns)[0]
pol_obj, _ = Polarization.objects.get_or_create(name=pol)
sat_obj, _ = Satellite.objects.get_or_create(
name=name,
defaults={
"norad": int(norad[0]) if norad else -1
})
trans_obj, created = Transponders.objects.get_or_create(
polarization=pol_obj,
downlink=(downlink_start+downlink_end)/2/1000000,
uplink=(uplink_start+uplink_end)/2/1000000,
frequency_range=abs(downlink_end-downlink_start)/1000000,
name=tr_name,
defaults={
"zone_name": tr_data['name'],
"sat_id": sat_obj,
}
)
if user:
if created:
trans_obj.created_by = user
trans_obj.updated_by = user
trans_obj.save()

165
dbapp/mapsapp/utils.py.backup Normal file
View File

@@ -0,0 +1,165 @@
# Standard library imports
import json
import re
from io import BytesIO
# Third-party imports
import requests
# Local imports
from mainapp.models import Polarization, Satellite
from .models import Transponders
def search_satellite_on_page(data: dict, satellite_name: str):
for pos, value in data.get('page', {}).get('positions').items():
for name in value['satellites']:
if name['other_names'] is None:
name['other_names'] = ''
if satellite_name.lower() in name['name'].lower() or satellite_name.lower() in name['other_names'].lower():
return pos, name['id']
return '', ''
def get_footprint_data(position: str = 62) -> dict:
"""Возвращает словарь с данным по footprint для спутников на выбранной долготе"""
response = requests.get(f"https://www.satbeams.com/footprints?position={position}")
response.raise_for_status()
match = re.search(r'var data = ({.*?});', response.text, re.DOTALL)
if match:
json_str = match.group(1)
try:
data = json.loads(json_str)
return data.get("page", {}).get("footprint_data", {}).get("beams",[])
except json.JSONDecodeError as e:
print("Ошибка парсинга JSON:", e)
else:
print("Нужных данных не найдено")
return {}
def get_all_page_data(url:str = 'https://www.satbeams.com/footprints') -> dict:
"""Возвращает словарь с данными по всем спутникам на странице"""
response = requests.get(url)
response.raise_for_status()
match = re.search(r'var data = ({.*?});', response.text, re.DOTALL)
if match:
json_str = match.group(1)
try:
data = json.loads(json_str)
# Файл json на диске для достоверности
with open('data.json', 'w') as jf:
json.dump(data, jf, indent=2)
return data
except json.JSONDecodeError as e:
print("Ошибка парсинга JSON:", e)
else:
print("Нужных данных не найдено")
return {}
def get_names_footprints_for_satellite(footprint_data: dict, sat_id: str) -> list[str]:
names = []
for beam in footprint_data:
if 'ku' in beam['band'].lower() and sat_id in beam['satellite_id']:
names.append(
{
"name": beam['name'],
"fullname": beam['fullname'][8:]
}
)
return names
def get_band_names(satellite_name: str) -> list[str]:
data = get_all_page_data()
pos, sat_id = search_satellite_on_page(data, satellite_name)
footprints = get_footprint_data(pos)
names = get_names_footprints_for_satellite(footprints, sat_id)
return names
def parse_transponders_from_json(filepath: str):
with open(filepath, encoding="utf-8") as jf:
data = json.load(jf)
for sat_name, trans_zone in data["satellites"].items():
for zone, trans in trans_zone.items():
for tran in trans:
f_b, f_e = tran["freq"][0].split("-")
f = round((float(f_b) + float(f_e))/2, 3)
f_range = round(abs(float(f_e) - float(f_b)), 3)
tran_obj = Transponders.objects.create(
name=tran["name"],
frequency=f,
frequency_range=f_range,
zone_name=zone,
polarization=Polarization.objects.get(name=tran["pol"]),
sat_id=Satellite.objects.get(name__iexact=sat_name)
)
tran_obj.save()
# Third-party imports (additional)
from lxml import etree
def parse_transponders_from_xml(data_in: BytesIO):
tree = etree.parse(data_in)
ns = {
'i': 'http://www.w3.org/2001/XMLSchema-instance',
'ns': 'http://schemas.datacontract.org/2004/07/Geolocation.Domain.Utils.Repository.SatellitesSerialization.Memos',
'tr': 'http://schemas.datacontract.org/2004/07/Geolocation.Common.Extensions'
}
satellites = tree.xpath('//ns:SatelliteMemo', namespaces=ns)
for sat in satellites[:]:
name = sat.xpath('./ns:name/text()', namespaces=ns)[0]
if name == 'X' or 'DONT USE' in name:
continue
norad = sat.xpath('./ns:norad/text()', namespaces=ns)
beams = sat.xpath('.//ns:BeamMemo', namespaces=ns)
zones = {}
for zone in beams:
zone_name = zone.xpath('./ns:name/text()', namespaces=ns)[0] if zone.xpath('./ns:name/text()', namespaces=ns) else '-'
zones[zone.xpath('./ns:id/text()', namespaces=ns)[0]] = {
"name": zone_name,
"pol": zone.xpath('./ns:polarization/text()', namespaces=ns)[0],
}
transponders = sat.xpath('.//ns:TransponderMemo', namespaces=ns)
for transponder in transponders:
tr_id = transponder.xpath('./ns:downlinkBeamId/text()', namespaces=ns)[0]
downlink_start = float(transponder.xpath('./ns:downlinkFrequency/tr:start/text()', namespaces=ns)[0])
downlink_end = float(transponder.xpath('./ns:downlinkFrequency/tr:end/text()', namespaces=ns)[0])
uplink_start = float(transponder.xpath('./ns:uplinkFrequency/tr:start/text()', namespaces=ns)[0])
uplink_end = float(transponder.xpath('./ns:uplinkFrequency/tr:end/text()', namespaces=ns)[0])
tr_data = zones[tr_id]
# p = tr_data['pol'][0] if tr_data['pol'] else '-'
match tr_data['pol']:
case 'Horizontal':
pol = 'Горизонтальная'
case 'Vertical':
pol = 'Вертикальная'
case 'CircularRight':
pol = 'Правая'
case 'CircularLeft':
pol = 'Левая'
case _:
pol = '-'
tr_name = transponder.xpath('./ns:name/text()', namespaces=ns)[0]
pol_obj, _ = Polarization.objects.get_or_create(name=pol)
sat_obj, _ = Satellite.objects.get_or_create(
name=name,
defaults={
"norad": int(norad[0]) if norad else -1
})
trans_obj, _ = Transponders.objects.get_or_create(
polarization=pol_obj,
downlink=(downlink_start+downlink_end)/2/1000000,
uplink=(uplink_start+uplink_end)/2/1000000,
frequency_range=abs(downlink_end-downlink_start)/1000000,
name=tr_name,
defaults={
"zone_name": tr_data['name'],
"sat_id": sat_obj,
}
)
trans_obj.save()