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 django.db.models import Q, QuerySet
from core.models.assemblies import Assembly
from core.models.conference import Conference, 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)
q_publishable_location_states = Q(location_state__in=[Assembly.LocationState.PREVIEW, Assembly.LocationState.FINAL])
def get_exportable_assemblies(conference: Conference) -> QuerySet[Assembly]:
"""Fetches all assemblies in the given conference with publish-able location."""
exportable_states = [*Assembly.PUBLIC_STATES, Assembly.State.HIDDEN]
return Assembly.objects.filter(conference=conference, state__in=exportable_states).filter(q_publishable_location_states)
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,
entry_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):
return get_exportable_assemblies(conference=self.conference)
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,
entry_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 # pylint: disable=import-outside-toplevel
data = []
qs = get_exportable_assemblies(self.conference)
if request.GET.get('all') != '1':
qs = qs.exclude(location_data__point=None, location_data__boundaries=None)
for assembly in qs.all(): # 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, i18n=False),
'parent_id': assembly.parent_id,
'children': get_exportable_assemblies(conference=self.conference).filter(parent=assembly).values_list('slug', flat=True)
if assembly.is_cluster
else None,
'rooms': assembly.rooms.filter(q_publishable_location_states).exclude(blocked=True).values_list('slug', flat=True),
'floor': assembly.get_location_floor_index(),
'is_preview': assembly.location_state != Assembly.LocationState.FINAL,
'location': loc_data.get('point'), # assembly.get_location_point_xy(),
'polygons': loc_data.get('boundaries'), # assembly.get_location_boundaries_xy(),
'tags': assembly.tags.filter(tag__is_public=True).values_list('tag__slug', flat=True),
}
)
for project in assembly.projects.conference_accessible(self.conference):
data.append(
{
'type': 'project',
'id': str(project.pk),
'slug': project.slug,
'name': project.name,
'description': {'de': project.description_de, 'en': project.description_en},
'public_url': hub_absolute('plainui:project', slug=project.slug, i18n=False),
'assembly_id': str(assembly.pk),
'floor': assembly.get_location_floor_index(), # TODO: allow project-specific location floor
'is_preview': assembly.location_state != Assembly.LocationState.FINAL,
'location': loc_data.get('point'), # TODO: allow project-specific location point
'location_text': project.location,
'tags': project.tags.filter(tag__is_public=True).values_list('tag__slug', flat=True),
}
)
for room in self.conference.rooms.filter(q_publishable_location_states).exclude(blocked=True):
data.append(
{
'type': 'room',
'id': str(room.pk),
'name': room.name,
'slug': room.slug,
'is_official': room.is_official,
'in_public_fahrplan': room.is_public_fahrplan,
'description': {'de': room.description_de, 'en': room.description_en},
'public_url': hub_absolute('plainui:room', slug=room.slug, i18n=False),
'is_preview': room.location_state != Assembly.LocationState.FINAL,
'floor': room.get_location_floor_index(),
'location': room.get_location_point_xy(),
'polygons': room.get_location_boundaries_xy(),
'tags': room.tags.filter(tag__is_public=True).values_list('tag__slug', flat=True),
}
)
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},
'is_preview': False,
'floor': poi.get_location_floor_index(),
'location': poi.get_location_point_xy(),
}
)
return Response(data)