team-3/src/c3nav/mapdata/models/locations.py
2019-12-27 20:36:22 +01:00

617 lines
24 KiB
Python

import string
from contextlib import suppress
from datetime import timedelta
from decimal import Decimal
from operator import attrgetter
from django.conf import settings
from django.core.cache import cache
from django.core.validators import MaxValueValidator, MinValueValidator, RegexValidator
from django.db import models, transaction
from django.db.models import FieldDoesNotExist, Prefetch
from django.urls import reverse
from django.utils import timezone
from django.utils.crypto import get_random_string
from django.utils.functional import cached_property
from django.utils.text import format_lazy
from django.utils.translation import ugettext_lazy as _
from django.utils.translation import ungettext_lazy
from c3nav.mapdata.fields import I18nField
from c3nav.mapdata.grid import grid
from c3nav.mapdata.models.access import AccessRestrictionMixin
from c3nav.mapdata.models.base import SerializableMixin, TitledMixin
from c3nav.mapdata.utils.fields import LocationById
from c3nav.mapdata.utils.models import get_submodels
class LocationSlugManager(models.Manager):
def get_queryset(self):
result = super().get_queryset()
if self.model == LocationSlug:
for model in get_submodels(Location) + [LocationRedirect]:
result = result.select_related(model._meta.default_related_name)
try:
model._meta.get_field('space')
except FieldDoesNotExist:
pass
else:
result = result.select_related(model._meta.default_related_name+'__space')
return result
def select_related_target(self):
if self.model != LocationSlug:
raise TypeError
qs = self.get_queryset()
qs = qs.select_related('redirect__target', *('redirect__target__'+model._meta.default_related_name
for model in get_submodels(Location) + [LocationRedirect]))
return qs
validate_slug = RegexValidator(
r'^[a-z0-9]+(--?[a-z0-9]+)*\Z',
# Translators: "letters" means latin letters: a-z and A-Z.
_('Enter a valid location slug consisting of lowercase letters, numbers or hyphens, '
'not starting or ending with hyphens or containing consecutive hyphens.'),
'invalid'
)
class LocationSlug(SerializableMixin, models.Model):
LOCATION_TYPE_CODES = {
'Level': 'l',
'Space': 's',
'Area': 'a',
'POI': 'p',
'LocationGroup': 'g'
}
LOCATION_TYPE_BY_CODE = {code: model_name for model_name, code in LOCATION_TYPE_CODES.items()}
slug = models.SlugField(_('Slug'), unique=True, null=True, blank=True, max_length=50, validators=[validate_slug])
objects = LocationSlugManager()
def get_child(self, instance=None):
for model in get_submodels(Location)+[LocationRedirect]:
with suppress(AttributeError):
return getattr(instance or self, model._meta.default_related_name)
return None
def get_slug(self):
return self.slug
def _serialize(self, **kwargs):
result = super()._serialize(**kwargs)
result['slug'] = self.get_slug()
return result
def details_display(self, **kwargs):
result = super().details_display(**kwargs)
result['display'].insert(2, (_('Slug'), str(self.get_slug())))
return result
@cached_property
def order(self):
return (-1, 0)
class Meta:
verbose_name = _('Location with Slug')
verbose_name_plural = _('Location with Slug')
default_related_name = 'locationslugs'
class Location(LocationSlug, AccessRestrictionMixin, TitledMixin, models.Model):
can_search = models.BooleanField(default=True, verbose_name=_('can be searched'))
can_describe = models.BooleanField(default=True, verbose_name=_('can describe'))
icon = models.CharField(_('icon'), max_length=32, null=True, blank=True, help_text=_('any material icons name'))
class Meta:
abstract = True
def serialize(self, detailed=True, describe_only=False, **kwargs):
result = super().serialize(detailed=detailed, **kwargs)
if not detailed:
fields = ('id', 'type', 'slug', 'title', 'subtitle', 'icon', 'point', 'bounds', 'grid_square',
'locations', 'on_top_of', 'label_settings', 'label_override', 'add_search', 'dynamic')
result = {name: result[name] for name in fields if name in result}
return result
def _serialize(self, search=False, **kwargs):
result = super()._serialize(**kwargs)
result['subtitle'] = str(self.subtitle)
result['icon'] = self.get_icon()
result['can_search'] = self.can_search
result['can_describe'] = self.can_search
if search:
result['add_search'] = ' '.join((
*(redirect.slug for redirect in self.redirects.all()),
*self.other_titles,
))
return result
def details_display(self, **kwargs):
result = super().details_display(**kwargs)
result['display'].extend([
(_('searchable'), _('Yes') if self.can_search else _('No')),
(_('can describe'), _('Yes') if self.can_describe else _('No')),
(_('icon'), self.get_icon()),
])
return result
def get_slug(self):
if self.slug is None:
code = self.LOCATION_TYPE_CODES.get(self.__class__.__name__)
if code is not None:
return code+':'+str(self.id)
return self.slug
@property
def subtitle(self):
return ''
@property
def grid_square(self):
return None
def get_color(self, instance=None):
# dont filter in the query here so prefetch_related works
result = self.get_color_sorted(instance)
return None if result is None else result[1]
def get_color_sorted(self, instance=None):
# dont filter in the query here so prefetch_related works
if instance is None:
instance = self
for group in instance.groups.all():
if group.color and getattr(group.category, 'allow_'+self.__class__._meta.default_related_name):
return (0, group.category.priority, group.hierarchy, group.priority), group.color
return None
def get_icon(self):
return self.icon or None
class SpecificLocation(Location, models.Model):
groups = models.ManyToManyField('mapdata.LocationGroup', verbose_name=_('Location Groups'), blank=True)
label_settings = models.ForeignKey('mapdata.LabelSettings', null=True, blank=True, on_delete=models.PROTECT,
verbose_name=_('label settings'))
label_override = I18nField(_('Label override'), plural_name='label_overrides', blank=True, fallback_any=True)
class Meta:
abstract = True
def _serialize(self, detailed=True, **kwargs):
result = super()._serialize(detailed=detailed, **kwargs)
if grid.enabled:
grid_square = self.grid_square
if grid_square is not None:
result['grid_square'] = grid_square or None
if detailed:
groups = {}
for group in self.groups.all():
groups.setdefault(group.category, []).append(group.pk)
groups = {category.name: (items[0] if items else None) if category.single else items
for category, items in groups.items()
if getattr(category, 'allow_'+self.__class__._meta.default_related_name)}
result['groups'] = groups
label_settings = self.get_label_settings()
if label_settings:
result['label_settings'] = label_settings.serialize(detailed=False)
if self.label_overrides:
# todo: what if only one language is set?
result['label_override'] = self.label_override
return result
def get_label_settings(self):
if self.label_settings:
return self.label_settings
for group in self.groups.all():
if group.label_settings:
return group.label_settings
return None
def details_display(self, **kwargs):
result = super().details_display(**kwargs)
groupcategories = {}
for group in self.groups.all():
groupcategories.setdefault(group.category, []).append(group)
if grid.enabled:
grid_square = self.grid_square
if grid_square is not None:
grid_square_title = (_('Grid Squares') if grid_square and '-' in grid_square else _('Grid Square'))
result['display'].insert(3, (grid_square_title, grid_square or None))
for category, groups in sorted(groupcategories.items(), key=lambda item: item[0].priority):
result['display'].insert(3, (
category.title if category.single else category.title_plural,
tuple({
'id': group.pk,
'slug': group.get_slug(),
'title': group.title,
'can_search': group.can_search,
} for group in sorted(groups, key=attrgetter('priority'), reverse=True))
))
return result
@cached_property
def describing_groups(self):
groups = tuple(self.groups.all() if 'groups' in getattr(self, '_prefetched_objects_cache', ()) else ())
groups = tuple(group for group in groups if group.can_describe)
return groups
@property
def subtitle(self):
subtitle = self.describing_groups[0].title if self.describing_groups else self.__class__._meta.verbose_name
if self.grid_square:
return '%s, %s' % (subtitle, self.grid_square)
return subtitle
@cached_property
def order(self):
groups = tuple(self.groups.all())
if not groups:
return (0, 0, 0)
return (0, groups[0].category.priority, groups[0].priority)
def get_icon(self):
icon = super().get_icon()
if icon:
return icon
for group in self.groups.all():
if group.icon and getattr(group.category, 'allow_' + self.__class__._meta.default_related_name):
return group.icon
return None
class LocationGroupCategory(SerializableMixin, models.Model):
name = models.SlugField(_('Name'), unique=True, max_length=50)
single = models.BooleanField(_('single selection'), default=False)
title = I18nField(_('Title'), plural_name='titles', fallback_any=True)
title_plural = I18nField(_('Title (Plural)'), plural_name='titles_plural', fallback_any=True)
help_text = I18nField(_('Help text'), plural_name='help_texts', fallback_any=True, fallback_value='')
allow_levels = models.BooleanField(_('allow levels'), db_index=True, default=True)
allow_spaces = models.BooleanField(_('allow spaces'), db_index=True, default=True)
allow_areas = models.BooleanField(_('allow areas'), db_index=True, default=True)
allow_pois = models.BooleanField(_('allow pois'), db_index=True, default=True)
allow_dynamic_locations = models.BooleanField(_('allow dynamic locations'), db_index=True, default=True)
priority = models.IntegerField(default=0, db_index=True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.orig_priority = self.priority
class Meta:
verbose_name = _('Location Group Category')
verbose_name_plural = _('Location Group Categories')
default_related_name = 'locationgroupcategories'
ordering = ('-priority', )
def _serialize(self, detailed=True, **kwargs):
result = super()._serialize(detailed=detailed, **kwargs)
result['name'] = self.name
if detailed:
result['titles'] = self.titles
result['title'] = self.title
return result
def register_changed_geometries(self):
from c3nav.mapdata.models.geometry.space import SpaceGeometryMixin
query = self.groups.all()
for model in get_submodels(SpecificLocation):
related_name = model._meta.default_related_name
subquery = model.objects.all()
if issubclass(model, SpaceGeometryMixin):
subquery = subquery.select_related('space')
query.prefetch_related(Prefetch('groups__'+related_name, subquery))
for group in query:
group.register_changed_geometries(do_query=False)
def save(self, *args, **kwargs):
if self.pk and self.priority != self.orig_priority:
self.register_changed_geometries()
super().save(*args, **kwargs)
class LocationGroupManager(models.Manager):
def get_queryset(self):
return super().get_queryset().select_related('category')
class LocationGroup(Location, models.Model):
category = models.ForeignKey(LocationGroupCategory, related_name='groups', on_delete=models.PROTECT,
verbose_name=_('Category'))
priority = models.IntegerField(default=0, db_index=True)
hierarchy = models.IntegerField(default=0, db_index=True, verbose_name=_('hierarchy'))
label_settings = models.ForeignKey('mapdata.LabelSettings', null=True, blank=True, on_delete=models.PROTECT,
verbose_name=_('label settings'),
help_text=_('unless location specifies otherwise'))
can_report_missing = models.BooleanField(default=False, verbose_name=_('for missing locations'),
help_text=_('can be used when reporting a missing location'))
color = models.CharField(null=True, blank=True, max_length=32, verbose_name=_('background color'))
objects = LocationGroupManager()
class Meta:
verbose_name = _('Location Group')
verbose_name_plural = _('Location Groups')
default_related_name = 'locationgroups'
ordering = ('-category__priority', '-priority')
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.orig_priority = self.priority
self.orig_hierarchy = self.hierarchy
self.orig_category_id = self.category_id
self.orig_color = self.color
def _serialize(self, simple_geometry=False, **kwargs):
result = super()._serialize(simple_geometry=simple_geometry, **kwargs)
result['category'] = self.category_id
result['color'] = self.color
if simple_geometry:
result['locations'] = tuple(obj.pk for obj in getattr(self, 'locations', ()))
return result
def details_display(self, editor_url=True, **kwargs):
result = super().details_display(**kwargs)
result['display'].insert(3, (_('Category'), self.category.title))
result['display'].extend([
(_('color'), self.color),
(_('priority'), self.priority),
])
if editor_url:
result['editor_url'] = reverse('editor.locationgroups.edit', kwargs={'pk': self.pk})
return result
@property
def title_for_forms(self):
attributes = []
if self.can_search:
attributes.append(_('search'))
if self.can_describe:
attributes.append(_('describe'))
if self.color:
attributes.append(_('color'))
if not attributes:
attributes.append(_('internal'))
return self.title + ' ('+', '.join(str(s) for s in attributes)+')'
def register_changed_geometries(self, do_query=True):
from c3nav.mapdata.models.geometry.space import SpaceGeometryMixin
for model in get_submodels(SpecificLocation):
query = getattr(self, model._meta.default_related_name).all()
if do_query:
if issubclass(model, SpaceGeometryMixin):
query = query.select_related('space')
for obj in query:
obj.register_change(force=True)
@property
def subtitle(self):
result = self.category.title
if hasattr(self, 'locations'):
return format_lazy(_('{category_title}, {num_locations}'),
category_title=result,
num_locations=(ungettext_lazy('%(num)d location', '%(num)d locations', 'num') %
{'num': len(self.locations)}))
return result
@cached_property
def order(self):
return (1, self.category.priority, self.priority)
def save(self, *args, **kwargs):
if self.pk and (self.orig_color != self.color or
self.priority != self.orig_priority or
self.hierarchy != self.orig_hierarchy or
self.category_id != self.orig_category_id):
self.register_changed_geometries()
super().save(*args, **kwargs)
def delete(self, *args, **kwargs):
self.register_changed_geometries()
super().delete(*args, **kwargs)
class LocationRedirect(LocationSlug):
target = models.ForeignKey(LocationSlug, related_name='redirects', on_delete=models.CASCADE,
verbose_name=_('target'))
def _serialize(self, include_type=True, **kwargs):
result = super()._serialize(include_type=include_type, **kwargs)
if type(self.target) == LocationSlug:
result['target'] = self.target.get_child().slug
else:
result['target'] = self.target.slug
if include_type:
result['type'] = 'redirect'
result.pop('id')
return result
class Meta:
default_related_name = 'redirect'
class LabelSettings(SerializableMixin, models.Model):
title = I18nField(_('Title'), plural_name='titles', fallback_any=True)
min_zoom = models.DecimalField(_('min zoom'), max_digits=3, decimal_places=1, default=-10,
validators=[MinValueValidator(Decimal('-10')),
MaxValueValidator(Decimal('10'))])
max_zoom = models.DecimalField(_('max zoom'), max_digits=3, decimal_places=1, default=10,
validators=[MinValueValidator(Decimal('-10')),
MaxValueValidator(Decimal('10'))])
font_size = models.IntegerField(_('font size'), default=12,
validators=[MinValueValidator(12),
MaxValueValidator(30)])
def _serialize(self, detailed=True, **kwargs):
result = super()._serialize(detailed=detailed, **kwargs)
if detailed:
result['titles'] = self.titles
if self.min_zoom > -10:
result['min_zoom'] = self.min_zoom
if self.max_zoom < 10:
result['max_zoom'] = self.max_zoom
result['font_size'] = self.font_size
return result
class Meta:
verbose_name = _('Label Settings')
verbose_name_plural = _('Label Settings')
default_related_name = 'labelsettings'
ordering = ('min_zoom', '-font_size')
class CustomLocationProxyMixin:
def get_custom_location(self):
raise NotImplementedError
def serialize_position(self):
raise NotImplementedError
class DynamicLocation(CustomLocationProxyMixin, SpecificLocation, models.Model):
position_secret = models.CharField(_('position secret'), max_length=32, null=True, blank=True)
class Meta:
verbose_name = _('Dynamic location')
verbose_name_plural = _('Dynamic locations')
default_related_name = 'dynamic_locations'
def _serialize(self, **kwargs):
"""custom_location = self.get_custom_location()
print(custom_location)
result = {} if custom_location is None else custom_location.serialize(**kwargs)
super_result = super()._serialize(**kwargs)
super_result['subtitle'] = '%s %s, %s' % (_('(moving)'), result['title'], result['subtitle'])
result.update(super_result)"""
result = super()._serialize(**kwargs)
result['dynamic'] = True
return result
def serialize_position(self):
custom_location = self.get_custom_location()
if custom_location is None:
return {
'available': False,
'title': self.title,
'subtitle': '%s %s, %s' % (_('currently unavailable'), _('(moving)'), self.subtitle)
}
result = custom_location.serialize(simple_geometry=True)
result.update({
'available': True,
'id': self.pk,
'slug': self.slug,
'coordinates': custom_location.pk,
'icon': self.get_icon(),
'title': self.title,
'subtitle': '%s %s%s, %s' % (
_('(moving)'),
('%s, ' % self.subtitle) if self.describing_groups else '',
result['title'],
result['subtitle']
),
})
return result
def get_custom_location(self):
if not self.position_secret:
return None
try:
return Position.objects.get(secret=self.position_secret).get_custom_location()
except Position.DoesNotExist:
return None
def details_display(self, editor_url=True, **kwargs):
result = super().details_display(**kwargs)
if editor_url:
result['editor_url'] = reverse('editor.dynamic_locations.edit', kwargs={'pk': self.pk})
return result
def get_position_secret():
return get_random_string(32, string.ascii_letters+string.digits)
def get_position_api_secret():
return get_random_string(64, string.ascii_letters+string.digits)
class Position(CustomLocationProxyMixin, models.Model):
owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
name = models.CharField(_('name'), max_length=32)
secret = models.CharField(_('secret'), unique=True, max_length=32, default=get_position_secret)
last_coordinates_update = models.DateTimeField(_('last coordinates update'), null=True)
timeout = models.PositiveSmallIntegerField(_('timeout (in seconds)'), default=0, help_text=_('0 for no timeout'))
coordinates_id = models.CharField(_('coordinates'), null=True, max_length=48)
api_secret = models.CharField(_('api secret'), max_length=64, default=get_position_api_secret)
coordinates = LocationById()
class Meta:
verbose_name = _('Dynamic position')
verbose_name_plural = _('Dynamic position')
default_related_name = 'dynamic_positions'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if self.timeout and self.last_coordinates_update:
end_time = self.last_coordinates_update + timedelta(seconds=self.timeout)
if timezone.now() >= end_time:
self.cordinates = None
self.last_coordinates_update = end_time
def get_custom_location(self):
return self.coordinates
@classmethod
def user_has_positions(cls, user):
if not user.is_authenticated:
return False
cache_key = 'user_has_positions:%d' % user.pk
result = cache.get(cache_key, None)
if result is None:
result = cls.objects.filter(owner=user).exists()
cache.set(cache_key, result, 600)
return result
def serialize_position(self):
custom_location = self.get_custom_location()
if custom_location is None:
return {
'id': self.secret,
'slug': self.secret,
'available': False,
'icon': 'my_location',
'title': self.name,
'subtitle': _('currently unavailable'),
}
result = custom_location.serialize(simple_geometry=True)
result.update({
'available': True,
'id': 'p:%s' % self.secret,
'slug': 'p:%s' % self.secret,
'coordinates': custom_location.pk,
'icon': 'my_location',
'title': self.name,
'subtitle': '%s, %s, %s' % (
_('Position'),
result['title'],
result['subtitle']
),
})
return result
def save(self, *args, **kwargs):
with transaction.atomic():
super().save(*args, **kwargs)
transaction.on_commit(lambda: cache.delete('user_has_positions:%d' % self.owner_id))
def delete(self, *args, **kwargs):
with transaction.atomic():
super().delete(*args, **kwargs)
transaction.on_commit(lambda: cache.delete('user_has_positions:%d' % self.owner_id))