548 lines
22 KiB
Python
548 lines
22 KiB
Python
from django.shortcuts import render, redirect
|
|
from django.contrib import messages
|
|
from django.http import JsonResponse, HttpResponse
|
|
from django.views.decorators.http import require_GET
|
|
from django.contrib.admin.views.decorators import staff_member_required
|
|
from django.utils.decorators import method_decorator
|
|
from django.views import View
|
|
from django.views.generic import TemplateView, FormView
|
|
from django.contrib.auth.mixins import UserPassesTestMixin
|
|
from django.db import models
|
|
import pandas as pd
|
|
from .utils import (
|
|
fill_data_from_df,
|
|
add_satellite_list,
|
|
get_points_from_csv,
|
|
get_vch_load_from_html,
|
|
compare_and_link_vch_load,
|
|
kub_report
|
|
)
|
|
from mapsapp.utils import parse_transponders_from_json, parse_transponders_from_xml
|
|
from .forms import LoadExcelData, LoadCsvData, UploadFileForm, VchLinkForm, UploadVchLoad, NewEventForm
|
|
from .models import ObjItem
|
|
from .clusters import get_clusters
|
|
from io import BytesIO
|
|
|
|
|
|
|
|
class AddSatellitesView(View):
|
|
def get(self, request):
|
|
add_satellite_list()
|
|
return redirect('home')
|
|
|
|
# class AddTranspondersView(View):
|
|
# def get(self, request):
|
|
# try:
|
|
# parse_transponders_from_json(BASE_DIR / "transponders.json")
|
|
# except FileNotFoundError:
|
|
# print("Файл не найден")
|
|
# return redirect('home')
|
|
|
|
class AddTranspondersView(FormView):
|
|
template_name = 'mainapp/transponders_upload.html'
|
|
form_class = UploadFileForm
|
|
|
|
def form_valid(self, form):
|
|
uploaded_file = self.request.FILES['file']
|
|
try:
|
|
content = uploaded_file.read()
|
|
parse_transponders_from_xml(BytesIO(content))
|
|
messages.success(self.request, "Файл успешно обработан")
|
|
except ValueError as e:
|
|
messages.error(self.request, f"Ошибка при чтении таблиц: {e}")
|
|
except Exception as e:
|
|
messages.error(self.request, f"Неизвестная ошибка: {e}")
|
|
return redirect('add_trans')
|
|
|
|
def form_invalid(self, form):
|
|
messages.error(self.request, "Форма заполнена некорректно.")
|
|
return super().form_invalid(form)
|
|
|
|
class HomePageView(TemplateView):
|
|
template_name = 'mainapp/actions.html'
|
|
|
|
|
|
class LoadExcelDataView(FormView):
|
|
template_name = 'mainapp/add_data_from_excel.html'
|
|
form_class = LoadExcelData
|
|
|
|
def form_valid(self, form):
|
|
uploaded_file = self.request.FILES['file']
|
|
selected_sat = form.cleaned_data['sat_choice']
|
|
number = form.cleaned_data['number_input']
|
|
|
|
try:
|
|
import io
|
|
df = pd.read_excel(io.BytesIO(uploaded_file.read()))
|
|
if number > 0:
|
|
df = df.head(number)
|
|
result = fill_data_from_df(df, selected_sat)
|
|
|
|
messages.success(self.request, f"Данные успешно загружены! Обработано строк: {result}")
|
|
return redirect('load_excel_data')
|
|
except Exception as e:
|
|
messages.error(self.request, f"Ошибка при обработке файла: {str(e)}")
|
|
return redirect('load_excel_data')
|
|
|
|
def form_invalid(self, form):
|
|
messages.error(self.request, "Форма заполнена некорректно.")
|
|
return super().form_invalid(form)
|
|
|
|
|
|
from django.views.generic import View
|
|
from django.core.paginator import Paginator
|
|
from django.db.models import Prefetch
|
|
from .models import Satellite, ObjItem, Parameter, Geo
|
|
|
|
class GetLocationsView(View):
|
|
def get(self, request, sat_id):
|
|
locations = ObjItem.objects.filter(parameters_obj__id_satellite=sat_id)
|
|
if not locations:
|
|
return JsonResponse({'error': 'Объектов не найдено'}, status=400)
|
|
|
|
features = []
|
|
for loc in locations:
|
|
param = loc.parameters_obj.get()
|
|
features.append({
|
|
"type": "Feature",
|
|
"geometry": {
|
|
"type": "Point",
|
|
"coordinates": [loc.geo_obj.coords[0], loc.geo_obj.coords[1]]
|
|
},
|
|
"properties": {
|
|
"pol": param.polarization.name,
|
|
"freq": param.frequency*1000000,
|
|
"name": f"{loc.name}",
|
|
"id": loc.geo_obj.id
|
|
}
|
|
})
|
|
|
|
return JsonResponse({
|
|
"type": "FeatureCollection",
|
|
"features": features
|
|
})
|
|
|
|
class LoadCsvDataView(FormView):
|
|
template_name = 'mainapp/add_data_from_csv.html'
|
|
form_class = LoadCsvData
|
|
|
|
def form_valid(self, form):
|
|
uploaded_file = self.request.FILES['file']
|
|
try:
|
|
# Read the file content and pass it directly to the function
|
|
content = uploaded_file.read()
|
|
if isinstance(content, bytes):
|
|
content = content.decode('utf-8')
|
|
|
|
get_points_from_csv(content)
|
|
messages.success(self.request, f"Данные успешно загружены!")
|
|
return redirect('load_csv_data')
|
|
except Exception as e:
|
|
messages.error(self.request, f"Ошибка при обработке файла: {str(e)}")
|
|
return redirect('load_csv_data')
|
|
|
|
def form_invalid(self, form):
|
|
messages.error(self.request, "Форма заполнена некорректно.")
|
|
return super().form_invalid(form)
|
|
|
|
|
|
from collections import defaultdict
|
|
|
|
@method_decorator(staff_member_required, name='dispatch')
|
|
class ShowMapView(UserPassesTestMixin, View):
|
|
def test_func(self):
|
|
return self.request.user.is_staff
|
|
|
|
def get(self, request):
|
|
ids = request.GET.get('ids', '')
|
|
points = []
|
|
if ids:
|
|
id_list = [int(x) for x in ids.split(',') if x.isdigit()]
|
|
locations = ObjItem.objects.filter(id__in=id_list).prefetch_related(
|
|
'parameters_obj__id_satellite',
|
|
'parameters_obj__polarization',
|
|
'parameters_obj__modulation',
|
|
'parameters_obj__standard',
|
|
'geo_obj'
|
|
)
|
|
for obj in locations:
|
|
param = obj.parameters_obj.get()
|
|
points.append({
|
|
'name': f"{obj.name}",
|
|
'freq': f"{param.frequency} [{param.freq_range}] МГц",
|
|
'point': (obj.geo_obj.coords.x, obj.geo_obj.coords.y)
|
|
})
|
|
else:
|
|
return redirect('admin')
|
|
grouped = defaultdict(list)
|
|
for p in points:
|
|
grouped[p["name"]].append({
|
|
'point': p["point"],
|
|
'frequency': p["freq"]
|
|
})
|
|
|
|
groups = [
|
|
{
|
|
"name": name,
|
|
"points": coords_list
|
|
}
|
|
for name, coords_list in grouped.items()
|
|
]
|
|
|
|
|
|
context = {
|
|
'groups': groups,
|
|
}
|
|
return render(request, 'admin/map_custom.html', context)
|
|
|
|
|
|
class ClusterTestView(View):
|
|
def get(self, request):
|
|
objs = ObjItem.objects.filter(name__icontains="! Astra 4A 12654,040 [1,962] МГц H")
|
|
coords = []
|
|
for obj in objs:
|
|
coords.append((obj.id_geo.coords[1], obj.id_geo.coords[0]))
|
|
get_clusters(coords)
|
|
|
|
return JsonResponse({"success": "ок"})
|
|
|
|
class UploadVchLoadView(FormView):
|
|
template_name = 'mainapp/upload_html.html'
|
|
form_class = UploadVchLoad
|
|
|
|
def form_valid(self, form):
|
|
selected_sat = form.cleaned_data['sat_choice']
|
|
uploaded_file = self.request.FILES['file']
|
|
try:
|
|
get_vch_load_from_html(uploaded_file, selected_sat)
|
|
messages.success(self.request, "Файл успешно обработан")
|
|
except ValueError as e:
|
|
messages.error(self.request, f"Ошибка при чтении таблиц: {e}")
|
|
except Exception as e:
|
|
messages.error(self.request, f"Неизвестная ошибка: {e}")
|
|
return redirect('vch_load')
|
|
|
|
def form_invalid(self, form):
|
|
messages.error(self.request, "Форма заполнена некорректно.")
|
|
return super().form_invalid(form)
|
|
|
|
|
|
class LinkVchSigmaView(FormView):
|
|
template_name = 'mainapp/link_vch.html'
|
|
form_class = VchLinkForm
|
|
|
|
def form_valid(self, form):
|
|
freq = form.cleaned_data['value1']
|
|
freq_range = form.cleaned_data['value2']
|
|
# ku_range = float(form.cleaned_data['ku_range'])
|
|
sat_id = form.cleaned_data['sat_choice']
|
|
# print(freq, freq_range, ku_range, sat_id.pk)
|
|
count_all, link_count = compare_and_link_vch_load(sat_id, freq, freq_range, 1)
|
|
messages.success(self.request, f"Привязано {link_count} из {count_all} объектов")
|
|
return redirect('link_vch_sigma')
|
|
|
|
def form_invalid(self, form):
|
|
return self.render_to_response(self.get_context_data(form=form))
|
|
|
|
|
|
class ProcessKubsatView(FormView):
|
|
template_name = 'mainapp/process_kubsat.html'
|
|
form_class = NewEventForm
|
|
|
|
def form_valid(self, form):
|
|
# selected_sat = form.cleaned_data['sat_choice']
|
|
# selected_pol = form.cleaned_data['pol_choice']
|
|
uploaded_file = self.request.FILES['file']
|
|
try:
|
|
content = uploaded_file.read()
|
|
df = kub_report(BytesIO(content))
|
|
output = BytesIO()
|
|
with pd.ExcelWriter(output, engine='openpyxl') as writer:
|
|
df.to_excel(writer, index=False, sheet_name='Результат')
|
|
output.seek(0)
|
|
|
|
response = HttpResponse(
|
|
output.getvalue(),
|
|
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
|
)
|
|
response['Content-Disposition'] = f'attachment; filename="kubsat_report.xlsx"'
|
|
|
|
messages.success(self.request, "Событие успешно обработано!")
|
|
return response
|
|
except Exception as e:
|
|
messages.error(self.request, f"Ошибка при обработке файла: {str(e)}")
|
|
return redirect('kubsat_excel')
|
|
# return redirect('kubsat_excel')
|
|
|
|
def form_invalid(self, form):
|
|
messages.error(self.request, "Форма заполнена некорректно.")
|
|
return super().form_invalid(form)
|
|
|
|
class ObjItemListView(View):
|
|
def get(self, request):
|
|
# Get satellites that have associated objects, sorted alphabetically
|
|
satellites = Satellite.objects.filter(parameters__objitems__isnull=False).distinct().order_by('name')
|
|
|
|
# Get selected satellite from query parameters
|
|
selected_sat_id = request.GET.get('satellite_id')
|
|
page_number = request.GET.get('page', 1)
|
|
items_per_page = request.GET.get('items_per_page', '25') # Default to 25 items per page
|
|
|
|
# Get filter parameters
|
|
freq_min = request.GET.get('freq_min')
|
|
freq_max = request.GET.get('freq_max')
|
|
range_min = request.GET.get('range_min')
|
|
range_max = request.GET.get('range_max')
|
|
snr_min = request.GET.get('snr_min')
|
|
snr_max = request.GET.get('snr_max')
|
|
bod_min = request.GET.get('bod_min')
|
|
bod_max = request.GET.get('bod_max')
|
|
search_query = request.GET.get('search')
|
|
selected_modulations = request.GET.getlist('modulation')
|
|
selected_polarizations = request.GET.getlist('polarization')
|
|
selected_satellites = request.GET.getlist('satellite_id')
|
|
has_kupsat = request.GET.get('has_kupsat')
|
|
has_valid = request.GET.get('has_valid')
|
|
|
|
try:
|
|
items_per_page = int(items_per_page)
|
|
except ValueError:
|
|
items_per_page = 25
|
|
|
|
# Only filter objects by selected satellite if provided (no data shown by default)
|
|
objects = ObjItem.objects.none() # Initially empty
|
|
|
|
if selected_satellites or selected_sat_id:
|
|
# Handle single satellite from old parameter or multiple from new parameter
|
|
if selected_sat_id and not selected_satellites:
|
|
# For backward compatibility - if only single satellite parameter is provided
|
|
try:
|
|
selected_sat_id_single = int(selected_sat_id)
|
|
selected_satellites = [selected_sat_id_single]
|
|
except ValueError:
|
|
selected_satellites = []
|
|
|
|
# Start with the basic filter if any satellites are selected
|
|
if selected_satellites:
|
|
# Start with the basic filter
|
|
objects = ObjItem.objects.select_related(
|
|
'id_user_add__user',
|
|
'geo_obj'
|
|
).prefetch_related(
|
|
'parameters_obj__id_satellite',
|
|
'parameters_obj__polarization',
|
|
'parameters_obj__modulation',
|
|
'parameters_obj__standard'
|
|
).filter(parameters_obj__id_satellite_id__in=selected_satellites)
|
|
else:
|
|
# If no satellites are selected, start with all objects
|
|
objects = ObjItem.objects.select_related(
|
|
'id_user_add__user',
|
|
'geo_obj'
|
|
).prefetch_related(
|
|
'parameters_obj__id_satellite',
|
|
'parameters_obj__polarization',
|
|
'parameters_obj__modulation',
|
|
'parameters_obj__standard'
|
|
)
|
|
|
|
# Apply additional filters
|
|
# Frequency filter
|
|
if freq_min is not None and freq_min.strip() != '':
|
|
try:
|
|
freq_min_val = float(freq_min)
|
|
objects = objects.filter(parameters_obj__frequency__gte=freq_min_val)
|
|
except ValueError:
|
|
pass
|
|
if freq_max is not None and freq_max.strip() != '':
|
|
try:
|
|
freq_max_val = float(freq_max)
|
|
objects = objects.filter(parameters_obj__frequency__lte=freq_max_val)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Range filter
|
|
if range_min is not None and range_min.strip() != '':
|
|
try:
|
|
range_min_val = float(range_min)
|
|
objects = objects.filter(parameters_obj__freq_range__gte=range_min_val)
|
|
except ValueError:
|
|
pass
|
|
if range_max is not None and range_max.strip() != '':
|
|
try:
|
|
range_max_val = float(range_max)
|
|
objects = objects.filter(parameters_obj__freq_range__lte=range_max_val)
|
|
except ValueError:
|
|
pass
|
|
|
|
# SNR filter
|
|
if snr_min is not None and snr_min.strip() != '':
|
|
try:
|
|
snr_min_val = float(snr_min)
|
|
objects = objects.filter(parameters_obj__snr__gte=snr_min_val)
|
|
except ValueError:
|
|
pass
|
|
if snr_max is not None and snr_max.strip() != '':
|
|
try:
|
|
snr_max_val = float(snr_max)
|
|
objects = objects.filter(parameters_obj__snr__lte=snr_max_val)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Symbol rate filter
|
|
if bod_min is not None and bod_min.strip() != '':
|
|
try:
|
|
bod_min_val = float(bod_min)
|
|
objects = objects.filter(parameters_obj__bod_velocity__gte=bod_min_val)
|
|
except ValueError:
|
|
pass
|
|
if bod_max is not None and bod_max.strip() != '':
|
|
try:
|
|
bod_max_val = float(bod_max)
|
|
objects = objects.filter(parameters_obj__bod_velocity__lte=bod_max_val)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Modulation filter
|
|
if selected_modulations:
|
|
objects = objects.filter(parameters_obj__modulation__id__in=selected_modulations)
|
|
|
|
# Polarization filter
|
|
if selected_polarizations:
|
|
objects = objects.filter(parameters_obj__polarization__id__in=selected_polarizations)
|
|
|
|
# Kupsat coords filter
|
|
if has_kupsat == '1': # has coords
|
|
objects = objects.filter(geo_obj__coords_kupsat__isnull=False)
|
|
elif has_kupsat == '0': # no coords
|
|
objects = objects.filter(geo_obj__coords_kupsat__isnull=True)
|
|
|
|
# Valid coords filter
|
|
if has_valid == '1': # has coords
|
|
objects = objects.filter(geo_obj__coords_valid__isnull=False)
|
|
elif has_valid == '0': # no coords
|
|
objects = objects.filter(geo_obj__coords_valid__isnull=True)
|
|
|
|
# Add search functionality - search only in name and location fields to avoid spatial lookup errors
|
|
if search_query:
|
|
search_query = search_query.strip()
|
|
if search_query:
|
|
# Search in name and location fields to match displayed text
|
|
objects = objects.filter(
|
|
models.Q(name__icontains=search_query) |
|
|
models.Q(geo_obj__location__icontains=search_query)
|
|
)
|
|
else:
|
|
selected_sat_id = None
|
|
|
|
# Add pagination
|
|
paginator = Paginator(objects, items_per_page)
|
|
page_obj = paginator.get_page(page_number)
|
|
|
|
# Prepare the data to include calculated fields for the template
|
|
processed_objects = []
|
|
for obj in page_obj:
|
|
# Get the first parameter
|
|
param = obj.parameters_obj.first() if obj.parameters_obj.exists() else None
|
|
|
|
# Process geo coordinates
|
|
geo_coords = "-"
|
|
kupsat_coords = "-"
|
|
valid_coords = "-"
|
|
distance_geo_kup = "-"
|
|
distance_geo_valid = "-"
|
|
distance_kup_valid = "-"
|
|
|
|
if obj.geo_obj:
|
|
# Format geo coordinates
|
|
if obj.geo_obj.coords:
|
|
longitude = obj.geo_obj.coords.coords[0]
|
|
latitude = obj.geo_obj.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"
|
|
geo_coords = f"{lat} {lon}"
|
|
|
|
# Format kupsat coordinates
|
|
if obj.geo_obj.coords_kupsat:
|
|
longitude = obj.geo_obj.coords_kupsat.coords[0]
|
|
latitude = obj.geo_obj.coords_kupsat.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"
|
|
kupsat_coords = f"{lat} {lon}"
|
|
elif obj.geo_obj.coords_kupsat is not None:
|
|
kupsat_coords = "-"
|
|
|
|
# Format valid coordinates
|
|
if obj.geo_obj.coords_valid:
|
|
longitude = obj.geo_obj.coords_valid.coords[0]
|
|
latitude = obj.geo_obj.coords_valid.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"
|
|
valid_coords = f"{lat} {lon}"
|
|
elif obj.geo_obj.coords_valid is not None:
|
|
valid_coords = "-"
|
|
|
|
# Format distances
|
|
if obj.geo_obj.distance_coords_kup is not None:
|
|
distance_geo_kup = f"{obj.geo_obj.distance_coords_kup:.3f}"
|
|
|
|
if obj.geo_obj.distance_coords_valid is not None:
|
|
distance_geo_valid = f"{obj.geo_obj.distance_coords_valid:.3f}"
|
|
|
|
if obj.geo_obj.distance_kup_valid is not None:
|
|
distance_kup_valid = f"{obj.geo_obj.distance_kup_valid:.3f}"
|
|
|
|
processed_objects.append({
|
|
'id': obj.id,
|
|
'name': obj.name or "-",
|
|
'satellite_name': param.id_satellite.name if param and param.id_satellite else "-",
|
|
'frequency': f"{param.frequency:.3f}" if param and param.frequency else "-",
|
|
'freq_range': f"{param.freq_range:.3f}" if param and param.freq_range else "-",
|
|
'polarization': param.polarization.name if param and param.polarization else "-",
|
|
'bod_velocity': f"{param.bod_velocity:.3f}" if param and param.bod_velocity else "-",
|
|
'modulation': param.modulation.name if param and param.modulation else "-",
|
|
'snr': f"{param.snr:.3f}" if param and param.snr else "-",
|
|
'geo_coords': geo_coords,
|
|
'kupsat_coords': kupsat_coords,
|
|
'valid_coords': valid_coords,
|
|
'distance_geo_kup': distance_geo_kup,
|
|
'distance_geo_valid': distance_geo_valid,
|
|
'distance_kup_valid': distance_kup_valid,
|
|
'obj': obj
|
|
})
|
|
|
|
# Get all modulations and polarizations for filter dropdowns
|
|
from .models import Modulation, Polarization
|
|
modulations = Modulation.objects.all()
|
|
polarizations = Polarization.objects.all()
|
|
|
|
context = {
|
|
'satellites': satellites,
|
|
'selected_satellite_id': selected_sat_id,
|
|
'page_obj': page_obj,
|
|
'processed_objects': processed_objects,
|
|
'items_per_page': items_per_page,
|
|
'available_items_per_page': [10, 25, 50, 100],
|
|
# Filter values
|
|
'freq_min': freq_min,
|
|
'freq_max': freq_max,
|
|
'range_min': range_min,
|
|
'range_max': range_max,
|
|
'snr_min': snr_min,
|
|
'snr_max': snr_max,
|
|
'bod_min': bod_min,
|
|
'bod_max': bod_max,
|
|
'search_query': search_query,
|
|
'selected_modulations': [int(x) for x in selected_modulations if x.isdigit()],
|
|
'selected_polarizations': [int(x) for x in selected_polarizations if x.isdigit()],
|
|
'selected_satellites': [int(x) for x in selected_satellites if x.isdigit()],
|
|
'has_kupsat': has_kupsat,
|
|
'has_valid': has_valid,
|
|
# For filter dropdowns
|
|
'modulations': modulations,
|
|
'polarizations': polarizations,
|
|
# Enable full width layout
|
|
'full_width_page': True,
|
|
}
|
|
|
|
return render(request, 'mainapp/objitem_list.html', context) |