"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 24.01.22
"""
import json
from django.contrib.gis.geos import MultiPolygon
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import QuerySet
from api.utils.serializer.serializer import AbstractModelAPISerializer
from codelist.models import KonovaCode
from codelist.settings import CODELIST_COMPENSATION_ACTION_ID, CODELIST_BIOTOPES_ID, CODELIST_PROCESS_TYPE_ID, \
CODELIST_LAW_ID, CODELIST_REGISTRATION_OFFICE_ID, CODELIST_CONSERVATION_OFFICE_ID, \
CODELIST_COMPENSATION_ACTION_DETAIL_ID, CODELIST_HANDLER_ID, \
CODELIST_BIOTOPES_EXTRA_CODES_FULL_ID
from compensation.models import CompensationAction, UnitChoices, CompensationState
from intervention.models import Responsibility, Legal, Handler
from konova.models import Deadline, DeadlineType
from konova.utils.message_templates import DATA_UNSHARED
class AbstractModelAPISerializerV1(AbstractModelAPISerializer):
def _model_to_geo_json(self, entry):
""" Adds the basic data, which all elements hold
Args:
entry (): The data entry
Returns:
"""
if entry.geometry.geom is not None:
geom = entry.geometry.geom.geojson
else:
geom = MultiPolygon().geojson
geo_json = json.loads(geom)
self.properties_data = {
"id": entry.id,
"identifier": entry.identifier,
"title": entry.title,
"created_on": self._created_on_to_json(entry),
"modified_on": self._modified_on_to_json(entry),
}
self._extend_properties_data(entry)
geo_json["properties"] = self.properties_data
return geo_json
def _konova_code_to_json(self, konova_code: KonovaCode):
""" Serializes KonovaCode model into json
Args:
konova_code (KonovaCode): The KonovaCode entry
Returns:
serialized_json (dict)
"""
if konova_code is None:
return None
return {
"id": konova_code.id,
"long_name": konova_code.long_name,
"short_name": konova_code.short_name,
}
def _konova_code_from_json(self, json_str, code_list_identifier):
""" Returns a konova code instance
Args:
json_str (str): The value for the code (id)
code_list_identifier (str): From which konova code list this code is supposed to be from
Returns:
"""
if json_str is None:
return None
json_str = str(json_str)
if len(json_str) == 0:
return None
try:
code = KonovaCode.objects.get(
id=json_str,
code_lists__in=[code_list_identifier]
)
except ObjectDoesNotExist as e:
msg = f"{e.args[0]} ({json_str} not found in official list {code_list_identifier})"
raise ObjectDoesNotExist(msg)
return code
def _created_on_to_json(self, entry):
""" Serializes the created_on into json
Args:
entry (BaseObject): The entry
Returns:
created_on (timestamp)
"""
return entry.created.timestamp if entry.created is not None else None
def _modified_on_to_json(self, entry):
""" Serializes the modified_on into json
Args:
entry (BaseObject): The entry
Returns:
modified_on (timestamp)
"""
modified_on = entry.modified or entry.created
modified_on = modified_on.timestamp if modified_on is not None else None
return modified_on
def delete_entry(self, id, user):
""" Marks an entry as deleted
Args:
id (str): The entry's id
user (User): The API user
Returns:
"""
entry = self._get_obj_from_db(id, user)
is_shared = entry.is_shared_with(user)
if not is_shared:
raise PermissionError(DATA_UNSHARED)
# Do not send mails if entry is deleting using API. THere could be hundreds of deletion resulting in hundreds of
# mails at once.
entry.mark_as_deleted(user, send_mail=False)
entry.refresh_from_db()
success = entry.deleted is not None
return success
class DeductableAPISerializerV1Mixin:
class Meta:
abstract = True
def _single_deduction_to_json(self, entry):
""" Serializes a single eco account deduction into json
Args:
entry (EcoAccountDeduction): An EcoAccountDeduction
Returns:
serialized_json (dict)
"""
return {
"id": entry.pk,
"eco_account": {
"id": entry.account.pk,
"identifier": entry.account.identifier,
"title": entry.account.title,
},
"surface": entry.surface,
"intervention": {
"id": entry.intervention.pk,
"identifier": entry.intervention.identifier,
"title": entry.intervention.title,
}
}
def _deductions_to_json(self, qs: QuerySet):
""" Serializes eco account deductions into json
Args:
qs (QuerySet): A queryset of EcoAccountDeduction entries
Returns:
serialized_json (list)
"""
return [
self._single_deduction_to_json(entry)
for entry in qs
]
class ResponsibilityAPISerializerV1Mixin:
class Meta:
abstract = True
def _handler_to_json(self, handler: Handler):
return {
"type": self._konova_code_to_json(handler.type),
"detail": handler.detail
}
def _responsible_to_json(self, responsible: Responsibility):
""" Serializes Responsibility model into json
Args:
responsible (Responsibility): The Responsibility entry
Returns:
serialized_json (dict)
"""
return {
"registration_office": self._konova_code_to_json(responsible.registration_office),
"registration_file_number": responsible.registration_file_number,
"conservation_office": self._konova_code_to_json(responsible.conservation_office),
"conservation_file_number": responsible.conservation_file_number,
"handler": self._handler_to_json(responsible.handler),
}
def _set_responsibility(self, obj, responsibility_data: dict):
""" Sets the responsible data contents to the provided responsibility_data dict
Args:
obj (Intervention): The intervention object
responsibility_data (dict): The new data
Returns:
obj
"""
if responsibility_data is None:
return obj
obj.responsible.registration_office = self._konova_code_from_json(
responsibility_data["registration_office"],
CODELIST_REGISTRATION_OFFICE_ID
)
obj.responsible.registration_file_number = responsibility_data["registration_file_number"]
obj.responsible.conservation_office = self._konova_code_from_json(
responsibility_data["conservation_office"],
CODELIST_CONSERVATION_OFFICE_ID,
)
obj.responsible.conservation_file_number = responsibility_data["conservation_file_number"]
obj.responsible.handler.type = self._konova_code_from_json(
responsibility_data["handler"]["type"],
CODELIST_HANDLER_ID,
)
obj.responsible.handler.detail = responsibility_data["handler"]["detail"]
return obj
class LegalAPISerializerV1Mixin:
class Meta:
abstract = True
def _legal_to_json(self, legal: Legal):
""" Serializes Legal model into json
Args:
legal (Legal): The Legal entry
Returns:
serialized_json (dict)
"""
return {
"registration_date": legal.registration_date,
"binding_date": legal.binding_date,
"process_type": self._konova_code_to_json(legal.process_type),
"laws": [self._konova_code_to_json(law) for law in legal.laws.all()],
}
def _set_legal(self, obj, legal_data):
""" Sets the legal data contents to the provided legal_data dict
Args:
obj (Intervention): The intervention object
legal_data (dict): The new data
Returns:
obj
"""
if legal_data is None:
return obj
obj.legal.registration_date = legal_data.get("registration_date", None)
obj.legal.binding_date = legal_data.get("binding_date", None)
obj.legal.process_type = self._konova_code_from_json(
legal_data.get("process_type", None),
CODELIST_PROCESS_TYPE_ID,
)
laws = [self._konova_code_from_json(law, CODELIST_LAW_ID) for law in legal_data.get("laws", [])]
obj.legal.laws.set(laws)
return obj
class AbstractCompensationAPISerializerV1Mixin:
class Meta:
abstract = True
def _set_deadlines(self, obj, deadline_data):
""" Sets the linked deadline data according to the given deadline_data
Args:
obj (Compensation): The Compensation object
deadline_data (dict): The posted deadline_data
Returns:
obj (Compensation)
"""
deadlines = []
for entry in deadline_data:
try:
deadline_type = entry["type"]
date = entry["date"]
comment = entry["comment"]
except KeyError:
raise ValueError(f"Invalid deadline content. Content was {entry} but should follow the specification")
# Check on validity
if deadline_type not in DeadlineType:
raise ValueError(f"Invalid deadline type. Choices are {DeadlineType.values}")
# If this exact data is already existing, we do not create it new. Instead put it's id in the list of
# entries, we will use to set the new actions
pre_existing_deadline = obj.deadlines.filter(
type=deadline_type,
date=date,
comment=comment,
).exclude(
id__in=deadlines
).first()
if pre_existing_deadline is not None:
deadlines.append(pre_existing_deadline.id)
else:
# Create and add id to list
new_deadline = Deadline.objects.create(
type=deadline_type,
date=date,
comment=comment,
)
deadlines.append(new_deadline.id)
obj.deadlines.set(deadlines)
return obj
def _set_compensation_states(self, obj, states_data, states_manager):
""" Sets the linked compensation state data according to the given states_data
Args:
obj (Compensation): The Compensation object
states_data (dict): The posted states_data
states_manager (Manager): The before_states or after_states manager
Returns:
obj (Compensation)
"""
states = []
for entry in states_data:
try:
biotope_type = entry["biotope"]
biotope_details = [
self._konova_code_from_json(e, CODELIST_BIOTOPES_EXTRA_CODES_FULL_ID) for e in entry["biotope_details"]
]
surface = float(entry["surface"])
except KeyError:
raise ValueError(f"Invalid biotope content. Content was {entry} but should follow the specification ")
# Check on validity
if surface <= 0:
raise ValueError("State surfaces must be > 0")
# If this exact data is already existing, we do not create it new. Instead put it's id in the list of
# entries, we will use to set the new actions
state = states_manager.filter(
biotope_type__id=biotope_type,
surface=surface,
).exclude(
id__in=states
).first()
if state is not None:
states.append(state.id)
else:
# Create and add id to list
state = CompensationState.objects.create(
biotope_type=self._konova_code_from_json(biotope_type, CODELIST_BIOTOPES_ID),
surface=surface
)
states.append(state.id)
state.biotope_type_details.set(biotope_details)
states_manager.set(states)
return obj
def _set_compensation_actions(self, obj, actions_data):
""" Sets the linked compensation action data according to the given actions_data
Args:
obj (Compensation): The Compensation object
actions_data (dict): The posted actions_data
Returns:
obj (Compensation)
"""
actions = []
for entry in actions_data:
try:
action_types = [
self._konova_code_from_json(e, CODELIST_COMPENSATION_ACTION_ID) for e in entry["action_types"]
]
action_details = [
self._konova_code_from_json(e, CODELIST_COMPENSATION_ACTION_DETAIL_ID) for e in entry["action_details"]
]
amount = float(entry["amount"])
# Mapping of old "qm" into "m²"
unit = UnitChoices.m2.value if entry["unit"] == "qm" else entry["unit"]
comment = entry["comment"]
except KeyError:
raise ValueError(f"Invalid action content. Content was {entry} but should follow specification")
# Check on validity
if amount <= 0:
raise ValueError("Action amount must be > 0")
if unit not in UnitChoices:
raise ValueError(f"Invalid unit. Choices are {UnitChoices.values}")
# If this exact data is already existing, we do not create it new. Instead put it's id in the list of
# entries, we will use to set the new actions
action_entry = obj.actions.filter(
action_type__in=action_types,
amount=amount,
unit=unit,
comment=comment,
).exclude(
id__in=actions
).first()
if action_entry is not None:
actions.append(action_entry.id)
else:
# Create and add id to list
action_entry = CompensationAction.objects.create(
amount=amount,
unit=unit,
comment=comment,
)
actions.append(action_entry.id)
action_entry.action_type.set(action_types)
action_entry.action_type_details.set(action_details)
obj.actions.set(actions)
return obj
def _compensation_state_to_json(self, qs: QuerySet):
""" Serializes compensation states into json
Args:
qs (QuerySet): A queryset of CompensationState entries
Returns:
serialized_json (list)
"""
return [
{
"biotope": self._konova_code_to_json(entry.biotope_type),
"biotope_details": [
self._konova_code_to_json(detail) for detail in entry.biotope_type_details.all()
],
"surface": entry.surface,
}
for entry in qs
]
def _compensation_actions_to_json(self, qs: QuerySet):
""" Serializes CompensationActions into json
Args:
qs (QuerySet): A queryset of CompensationAction entries
Returns:
serialized_json (list)
"""
return [
{
"action_types": [
self._konova_code_to_json(action) for action in entry.action_type.all()
],
"action_details": [
self._konova_code_to_json(detail) for detail in entry.action_type_details.all()
],
"amount": entry.amount,
"unit": entry.unit,
"comment": entry.comment,
}
for entry in qs
]
def _deadlines_to_json(self, qs: QuerySet):
""" Serializes deadlines into json
Args:
qs (QuerySet): A queryset of Deadline entries
Returns:
serialized_json (list)
"""
return list(qs.values(
"type",
"date",
"comment",
))