Select Git revision
create_db.py
Forked from
uffd / uffd
Source project has a limited visibility.
maps.py 6.30 KiB
import abc
import json
from rest_framework.response import Response
from rest_framework.views import APIView
from django.contrib.gis.gdal import CoordTransform, SpatialReference
from core.models.assemblies import Assembly
from core.models.conference import ConferenceExportCache
from core.models.map import MapPOI
from api.permissions import IsConferenceService, IsSuperUser
from api.views.mixins import ConferenceSlugMixin
_cts = {} # cache of CoordTransforms (if needed)
def get_field_geojson(value, srid: int, ct_cache: dict | None = None):
if value.srid != srid:
if ct_cache is None:
ct_cache = {}
if value.srid not in ct_cache:
srs = SpatialReference(srid)
ct_cache[value.srid] = CoordTransform(value.srs, srs)
value.transform(ct_cache[value.srid])
return json.loads(value.geojson)
class PoiExportView(ConferenceSlugMixin, APIView):
def get_geojson(self):
srid = 4326 # RFC7946 mandates WGS84
ct_cache = {}
features = []
result = {
'type': 'FeatureCollection',
'features': features,
}
for poi in MapPOI.objects.filter(conference=self.conference, visible=True).exclude(location_point=None):
feature = {
'type': 'Feature',
'id': str(poi.id),
'geometry': get_field_geojson(poi.location_point, srid=srid, ct_cache=ct_cache),
'properties': {
'floor': poi.get_location_floor_index(),
'type': 'poi',
'name': poi.name,
'official': poi.is_official,
},
}
features.append(feature)
return result
def get(self, request, *args, **kwargs):
cache_id = 'geojson-poi'
return ConferenceExportCache.handle_http_request(
request=request,
conference=self.conference,
type=ConferenceExportCache.Type.MAP,
ident=cache_id,
content_type='application/geo+json',
result_func=lambda: json.dumps(self.get_geojson()),
)
class AssembliesExportView(ConferenceSlugMixin, APIView, metaclass=abc.ABCMeta):
geometry_field = None
def get_queryset(self):
exportable_states = [*Assembly.PLACED_STATES, Assembly.State.HIDDEN]
return Assembly.objects.filter(conference=self.conference, state__in=exportable_states)
def get_geometry_field(self, obj):
return getattr(obj, self.geometry_field)
def get_geojson(self):
srid = 4326 # RFC7946 mandates WGS84
ct_cache = {}
features = []
result = {
'type': 'FeatureCollection',
'features': features,
}
for assembly in self.get_queryset().select_related('parent').all():
geometry = self.get_geometry_field(assembly)
if geometry is None:
continue
feature = {
'type': 'Feature',
'id': assembly.slug,
'geometry': get_field_geojson(geometry, srid=srid, ct_cache=ct_cache),
'properties': {
'floor': assembly.get_location_floor_index(),
'type': 'assembly',
'name': assembly.name,
'official': assembly.is_official,
'cluster': assembly.is_cluster,
'parent': assembly.parent.slug if assembly.parent else None,
},
}
features.append(feature)
return result
def get(self, request, *args, **kwargs):
cache_id = 'geojson-' + self.geometry_field
return ConferenceExportCache.handle_http_request(
request=request,
conference=self.conference,
type=ConferenceExportCache.Type.MAP,
ident=cache_id,
content_type='application/geo+json',
result_func=lambda: json.dumps(self.get_geojson()),
)
class AssembliesPoiExportView(AssembliesExportView):
geometry_field = 'location_point'
class AssembliesAreasExportView(AssembliesExportView):
geometry_field = 'location_boundaries'
class C3NavExportView(ConferenceSlugMixin, APIView):
permission_classes = [IsConferenceService | IsSuperUser]
required_service_classes = ['c3nav']
def get(self, request, *args, **kwargs):
from core.templatetags.hub_absolute import hub_absolute
data = []
exportable_states = [*Assembly.PLACED_STATES, Assembly.State.HIDDEN]
qs = self.conference.assemblies.filter(state__in=exportable_states)
if request.GET.get('all') != '1':
qs = qs.exclude(location_data__point=None, location_data__boundaries=None)
for assembly in qs: # type: Assembly
loc_data = assembly.location_data or {}
data.append(
{
'type': 'assembly',
'id': str(assembly.pk),
'slug': assembly.slug,
'name': assembly.name,
'is_official': assembly.is_official,
'description': {'de': assembly.description_de, 'en': assembly.description_en},
'public_url': hub_absolute('plainui:assembly', assembly_slug=assembly.slug),
'parent_id': assembly.parent_id,
'children': assembly.children.filter(state__in=exportable_states).values_list('slug', flat=True) if assembly.is_cluster else None,
'floor': assembly.get_location_floor_index(),
'location': loc_data.get('point'), # assembly.get_location_point_xy(),
'polygons': loc_data.get('boundaries'), # assembly.get_location_boundaries_xy(),
}
)
for poi in self.conference.pois.filter(visible=True): # type: MapPOI
data.append(
{
'type': 'poi',
'id': str(poi.pk),
'name': {'de': poi.name_de, 'en': poi.name_en},
'is_official': poi.is_official,
'description': {'de': poi.description_de, 'en': poi.description_en},
'floor': poi.get_location_floor_index(),
'location': poi.get_location_point_xy(),
}
)
return Response(data)