1# (c) 2018-2020 

2# MPIB <https://www.mpib-berlin.mpg.de/>, 

3# MPI-CBS <https://www.cbs.mpg.de/>, 

4# MPIP <http://www.psych.mpg.de/> 

5# 

6# This file is part of Castellum. 

7# 

8# Castellum is free software; you can redistribute it and/or modify it 

9# under the terms of the GNU Affero General Public License as published 

10# by the Free Software Foundation; either version 3 of the License, or 

11# (at your option) any later version. 

12# 

13# Castellum is distributed in the hope that it will be useful, but 

14# WITHOUT ANY WARRANTY; without even the implied warranty of 

15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 

16# Affero General Public License for more details. 

17# 

18# You should have received a copy of the GNU Affero General Public 

19# License along with Castellum. If not, see 

20# <http://www.gnu.org/licenses/>. 

21 

22import functools 

23from collections import OrderedDict 

24 

25from django.contrib.postgres.fields import JSONField 

26from django.core.exceptions import ValidationError 

27from django.core.serializers.json import DjangoJSONEncoder 

28from django.core.validators import MaxValueValidator 

29from django.db import models 

30from django.urls import reverse 

31from django.utils.translation import gettext_lazy as _ 

32 

33from parler.managers import TranslatableManager 

34from parler.models import TranslatableModel 

35from parler.models import TranslatedFields 

36from parler.utils.context import switch_language 

37 

38from castellum.studies.models import StudyType 

39from castellum.subjects.models import Subject 

40from castellum.utils.models import TimeStampedModel 

41 

42from .. import attribute_fields 

43from ..attribute_fields import ANSWER_DECLINED 

44 

45UNCATEGORIZED = "UNCATEGORIZED" 

46 

47 

48@functools.lru_cache(maxsize=2) 

49def get_description_by_statistics_rank(rank): 

50 try: 

51 return AttributeDescription.objects.get(statistics_rank=rank) 

52 except AttributeDescription.DoesNotExist: 

53 return None 

54 

55 

56class ImportMixin: 

57 def export(self): 

58 data = {} 

59 for field in self.import_fields: 

60 data[field] = getattr(self, field) 

61 for field in self.import_translated_fields: 

62 data[field] = {} 

63 for translation in self.translations.all(): 

64 for field in self.import_translated_fields: 

65 data[field][translation.language_code] = getattr(translation, field) 

66 return data 

67 

68 @classmethod 

69 def _import(cls, data, **kwargs): 

70 obj = cls(**kwargs) 

71 for key in data: 

72 if key in cls.import_translated_fields: 

73 for lang, value in data[key].items(): 

74 with switch_language(obj, lang): 

75 setattr(obj, key, value) 

76 elif key in cls.import_fields: 

77 setattr(obj, key, data[key]) 

78 obj.save() 

79 return obj 

80 

81 

82class AttributeSet(TimeStampedModel): 

83 subject = models.OneToOneField(Subject, on_delete=models.CASCADE, editable=False) 

84 data = JSONField(encoder=DjangoJSONEncoder) 

85 study_type_disinterest = models.ManyToManyField( 

86 StudyType, 

87 verbose_name=_('Does not want to participate in the following study types:'), 

88 blank=True, 

89 related_name='+', 

90 ) 

91 

92 class Meta: 

93 verbose_name = _('Attribute set') 

94 verbose_name_plural = _('Attribute sets') 

95 

96 def get_absolute_url(self): 

97 return reverse('subjects:detail', args=[self.subject.pk]) 

98 

99 def get_data(self): 

100 qs = AttributeDescription.objects.all() 

101 return {desc.json_key: desc.field.from_json(self.data.get(desc.json_key)) for desc in qs} 

102 

103 def get_verbose_name(self, pk): 

104 description = AttributeDescription.objects.get(pk=pk) 

105 return description.label 

106 

107 def get_display(self, pk): 

108 description = AttributeDescription.objects.get(pk=pk) 

109 value = self.data.get(description.json_key) 

110 return description.field.get_display(value) 

111 

112 def get_field_names(self): 

113 return AttributeDescription.objects.values_list('pk', flat=True) 

114 

115 def get_statistics_bucket(self, rank): 

116 description = get_description_by_statistics_rank(rank) 

117 if not description: 117 ↛ 118line 117 didn't jump to line 118, because the condition on line 117 was never true

118 return None 

119 value = description.field.from_json(self.data.get(description.json_key)) 

120 

121 if not value or value == ANSWER_DECLINED: 

122 return None 

123 

124 return description.field.get_statistics_bucket(value) 

125 

126 def get_completeness(self): 

127 # Needs to match ``filter_queries.completeness_expr()`` 

128 completed = len([k for k, v in self.data.items() if v not in [None, '']]) 

129 total = AttributeDescription.objects.count() 

130 return completed, total 

131 

132 

133class AttributeCategory(ImportMixin, TranslatableModel): 

134 import_fields = ['order'] 

135 import_translated_fields = ['label'] 

136 

137 order = models.IntegerField(default=0) 

138 

139 translations = TranslatedFields( 

140 label=models.CharField(max_length=64) 

141 ) 

142 

143 class Meta: 

144 ordering = ['order'] 

145 verbose_name = _('Attribute category') 

146 verbose_name_plural = _('Attribute categories') 

147 

148 def __str__(self): 

149 return self.label 

150 

151 def export(self): 

152 data = super().export() 

153 data['descriptions'] = [desc.export() for desc in self.attributedescription_set.all()] 

154 return data 

155 

156 @classmethod 

157 def _import(cls, data): 

158 category = super()._import(data) 

159 for description in data.get('descriptions', []): 

160 AttributeDescription._import(description, category=category) 

161 return category 

162 

163 

164class AttributeDescriptionManager(TranslatableManager): 

165 def allowed_read(self, user, obj=None): 

166 privacy_level = user.get_privacy_level(obj=obj) 

167 return self.filter(privacy_level_read__lte=privacy_level) 

168 

169 def allowed_write(self, user, obj=None): 

170 privacy_level = user.get_privacy_level(obj=obj) 

171 return self.filter(privacy_level_write__lte=privacy_level) 

172 

173 def by_category(self, qs=None): 

174 qs = qs or self.all() 

175 qs = qs.select_related('category').order_by('category__order', 'order') 

176 

177 categories = OrderedDict() 

178 categories[UNCATEGORIZED] = [] 

179 for description in qs: 

180 category = description.category or UNCATEGORIZED 

181 categories.setdefault(category, []) 

182 categories[category].append(description) 

183 return categories 

184 

185 

186class AttributeDescription(ImportMixin, TranslatableModel): 

187 FIELD_TYPE_CHOICES = ( 

188 ('CharField', _('CharField')), 

189 ('ChoiceField', _('ChoiceField')), 

190 ('MultipleChoiceField', _('MultipleChoiceField')), 

191 ('OrderedChoiceField', _('OrderedChoiceField')), 

192 ('IntegerField', _('IntegerField')), 

193 ('BooleanField', _('BooleanField')), 

194 ('DateField', _('DateField')), 

195 ('AgeField', _('AgeField')), 

196 ) 

197 STATISTICS_RANK_CHOICES = ( 

198 ('primary', _('primary')), 

199 ('secondary', _('secondary')), 

200 ) 

201 STATISTICS_TYPES = ['ChoiceField', 'OrderedChoiceField', 'BooleanField', 'AgeField'] 

202 

203 import_fields = ['field_type', 'privacy_level_read', 'privacy_level_write', 'order'] 

204 import_translated_fields = ['label', '_filter_label', 'help_text'] 

205 

206 translations = TranslatedFields( 

207 label=models.CharField(max_length=64), 

208 _filter_label=models.CharField(max_length=64, blank=True, default=''), 

209 help_text=models.TextField(blank=True, default=''), 

210 ) 

211 

212 field_type = models.CharField(max_length=64, choices=FIELD_TYPE_CHOICES) 

213 privacy_level_read = models.PositiveSmallIntegerField( 

214 default=0, validators=[MaxValueValidator(2)] 

215 ) 

216 privacy_level_write = models.PositiveSmallIntegerField( 

217 default=0, validators=[MaxValueValidator(2)] 

218 ) 

219 order = models.IntegerField(default=0) 

220 category = models.ForeignKey( 

221 AttributeCategory, on_delete=models.SET_NULL, null=True, blank=True) 

222 statistics_rank = models.CharField( 

223 max_length=16, choices=STATISTICS_RANK_CHOICES, null=True, blank=True, unique=True) 

224 

225 objects = AttributeDescriptionManager() 

226 

227 class Meta: 

228 ordering = ['order'] 

229 

230 def __str__(self): 

231 if self.category: 231 ↛ 232line 231 didn't jump to line 232, because the condition on line 231 was never true

232 return '{}:{}'.format(self.category, self.label) 

233 return self.label 

234 

235 def clean(self): 

236 if self.privacy_level_read > self.privacy_level_write: 

237 raise ValidationError( 

238 _('Privacy level for reading may not be higher than for writing.'), code='invalid') 

239 

240 if self.statistics_rank and self.field_type not in self.STATISTICS_TYPES: 

241 msg = _('Statistics rank can only be set for the following field types: %s') 

242 raise ValidationError(msg, code='invalid', params=', '.join(self.STATISTICS_TYPES)) 

243 

244 @property 

245 def json_key(self): 

246 # can not be a number because then queries like `data__8` would expect an array. 

247 return 'd{}'.format(self.pk) 

248 

249 @property 

250 def filter_label(self): 

251 return self._filter_label or self.label 

252 

253 @property 

254 def field(self): 

255 MAPPING = { 

256 'CharField': attribute_fields.TextAttributeField, 

257 'ChoiceField': attribute_fields.ChoiceAttributeField, 

258 'MultipleChoiceField': attribute_fields.MultipleChoiceAttributeField, 

259 'OrderedChoiceField': attribute_fields.OrderedChoiceAttributeField, 

260 'IntegerField': attribute_fields.NumberAttributeField, 

261 'BooleanField': attribute_fields.BooleanAttributeField, 

262 'DateField': attribute_fields.DateAttributeField, 

263 'AgeField': attribute_fields.AgeAttributeField, 

264 } 

265 return MAPPING[self.field_type](self) 

266 

267 @classmethod 

268 def get_statistics_buckets(cls, rank): 

269 description = get_description_by_statistics_rank(rank) 

270 if not description: 270 ↛ 271line 270 didn't jump to line 271, because the condition on line 270 was never true

271 return [(None, None)] 

272 

273 return description.field.get_statistics_buckets() + [ 

274 (None, _('Other')), 

275 ] 

276 

277 def export(self): 

278 data = super().export() 

279 data['choices'] = [choice.export() for choice in self.attributechoice_set.all()] 

280 return data 

281 

282 @classmethod 

283 def _import(cls, data, category=None): 

284 description = super()._import(data, category=category) 

285 for index, choice in enumerate(data.get('choices', [])): 

286 AttributeChoice._import(choice, description=description, order=index * 10) 

287 return description 

288 

289 

290class AttributeChoice(ImportMixin, TranslatableModel): 

291 import_fields = [] 

292 import_translated_fields = ['label'] 

293 

294 description = models.ForeignKey(AttributeDescription, on_delete=models.CASCADE) 

295 order = models.IntegerField(default=0) 

296 translations = TranslatedFields( 

297 label=models.CharField(max_length=64) 

298 ) 

299 

300 class Meta: 

301 ordering = ['order']