"""
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.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
from compensation.models import CompensationAction, UnitChoices, CompensationState
from intervention.models import Responsibility, Legal
from konova.models import Deadline, DeadlineType
class AbstractModelAPISerializerV1(AbstractModelAPISerializer):
def model_to_geo_json(self, entry):
""" Adds the basic data, which all elements hold
Args:
entry (): The data entry
Returns:
"""
geom = entry.geometry.geom.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 or len(json_str) == 0:
return None
code = KonovaCode.objects.get(
atom_id=json_str,
code_lists__in=[code_list_identifier]
)
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
class DeductableAPISerializerV1Mixin:
class Meta:
abstract = True
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 [
{
"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,
}
}
for entry in qs
]
class ResponsibilityAPISerializerV1Mixin:
class Meta:
abstract = True
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": 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.get("registration_office", None),
CODELIST_REGISTRATION_OFFICE_ID
)
obj.responsible.registration_file_number = responsibility_data.get("registration_file_number", None)
obj.responsible.conservation_office = self.konova_code_from_json(
responsibility_data.get("conservation_office", None),
CODELIST_CONSERVATION_OFFICE_ID,
)
obj.responsible.conservation_file_number = responsibility_data.get("conservation_file_number", None)
obj.responsible.handler = responsibility_data.get("handler", None)
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:
deadline_type = entry["type"]
date = entry["date"]
comment = entry["comment"]
# 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:
biotope_type = entry["biotope"]
surface = float(entry["surface"])
# 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
pre_existing_state = states_manager.filter(
biotope_type__atom_id=biotope_type,
surface=surface,
).exclude(
id__in=states
).first()
if pre_existing_state is not None:
states.append(pre_existing_state.id)
else:
# Create and add id to list
new_state = CompensationState.objects.create(
biotope_type=self.konova_code_from_json(biotope_type, CODELIST_BIOTOPES_ID),
surface=surface
)
states.append(new_state.id)
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:
action = entry["action"]
amount = float(entry["amount"])
unit = entry["unit"]
comment = entry["comment"]
# 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
pre_existing_action = obj.actions.filter(
action_type__atom_id=action,
amount=amount,
unit=unit,
comment=comment,
).exclude(
id__in=actions
).first()
if pre_existing_action is not None:
actions.append(pre_existing_action.id)
else:
# Create and add id to list
new_action = CompensationAction.objects.create(
action_type=self.konova_code_from_json(action, CODELIST_COMPENSATION_ACTION_ID),
amount=amount,
unit=unit,
comment=comment,
)
actions.append(new_action.id)
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),
"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": self.konova_code_to_json(entry.action_type),
"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",
))