497 lines
17 KiB
Python
497 lines
17 KiB
Python
"""
|
|
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_BIOTOPES_EXTRA_CODES_ID, CODELIST_HANDLER_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 {
|
|
"atom_id": konova_code.atom_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 (atom 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(
|
|
atom_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_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__atom_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",
|
|
)) |