Skip to content
Snippets Groups Projects
Select Git revision
  • 6881f9e3efaeaaa6679d29d01efedfec7c763459
  • main default protected
  • staging
3 results

main.py

Blame
  • 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)