IT-Naturschutz/konova
7
0
Fork 0
Code Issues 6 Pull Requests 1 Actions Projects Releases 37 Wiki Activity

Refactoring

Browse Source 
* splits compensation/models.py into subpackage
* renames base objects by adding suffix Mixin
This commit is contained in:
mpeltriaux 2021-11-16 08:29:18 +01:00
parent 6f107ac7a1
commit 78945d648f
11 changed files with 751 additions and 684 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

670
compensation/models.py
View File

@ -1,670 +0,0 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 17.11.20
"""
import shutil
from django.contrib.auth.models import User
from django.contrib.gis.db import models
from django.core.exceptions import ValidationError
from django.core.validators import MinValueValidator
from django.db import transaction
from django.db.models import Sum, QuerySet
from django.utils.translation import gettext_lazy as _
from codelist.models import KonovaCode
from intervention.models import Intervention, Responsibility
from codelist.settings import CODELIST_COMPENSATION_ACTION_ID, CODELIST_BIOTOPES_ID
from compensation.managers import CompensationStateManager, EcoAccountDeductionManager, CompensationActionManager, \
EcoAccountManager, CompensationManager
from compensation.utils.quality import CompensationQualityChecker, EcoAccountQualityChecker
from konova.models import BaseObject, BaseResource, Geometry, UuidModel, AbstractDocument, \
generate_document_file_upload_path, RecordableObject, ShareableObject, Deadline
from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE
from user.models import UserActionLogEntry, UserAction
class Payment(BaseResource):
"""
Holds data on a payment for an intervention (alternative to a classic compensation)
"""
amount = models.FloatField(validators=[MinValueValidator(limit_value=0.00)])
due_on = models.DateField(null=True, blank=True)
comment = models.TextField(
null=True,
blank=True,
help_text="Refers to german money transfer 'Verwendungszweck'",
)
intervention = models.ForeignKey(
Intervention,
null=True,
blank=True,
on_delete=models.CASCADE,
related_name='payments'
)
class Meta:
ordering = [
"-amount",
]
class CompensationState(UuidModel):
"""
Compensations must define the state of an area before and after the compensation.
"""
biotope_type = models.ForeignKey(
KonovaCode,
on_delete=models.SET_NULL,
null=True,
blank=True,
limit_choices_to={
"code_lists__in": [CODELIST_BIOTOPES_ID],
"is_selectable": True,
"is_archived": False,
}
)
surface = models.FloatField()
objects = CompensationStateManager()
def __str__(self):
return "{} | {} m²".format(self.biotope_type, self.surface)
class UnitChoices(models.TextChoices):
"""
Predefines units for selection
"""
cm = "cm", _("cm")
m = "m", _("m")
km = "km", _("km")
qm = "qm", _("m²")
ha = "ha", _("ha")
st = "pcs", _("Pieces") # pieces
class CompensationAction(BaseResource):
"""
Compensations include actions like planting trees, refreshing rivers and so on.
"""
action_type = models.ForeignKey(
KonovaCode,
on_delete=models.SET_NULL,
null=True,
blank=True,
limit_choices_to={
"code_lists__in": [CODELIST_COMPENSATION_ACTION_ID],
"is_selectable": True,
"is_archived": False,
}
)
amount = models.FloatField()
unit = models.CharField(max_length=100, null=True, blank=True, choices=UnitChoices.choices)
comment = models.TextField(blank=True, null=True, help_text="Additional comment")
objects = CompensationActionManager()
def __str__(self):
return "{} | {} {}".format(self.action_type, self.amount, self.unit)
@property
def unit_humanize(self):
""" Returns humanized version of enum
Used for template rendering
Returns:
"""
choices = UnitChoices.choices
for choice in choices:
if choice[0] == self.unit:
return choice[1]
return None
class AbstractCompensation(BaseObject):
"""
Abstract compensation model which holds basic attributes, shared by subclasses like the regular Compensation,
EMA or EcoAccount.
"""
responsible = models.OneToOneField(
Responsibility,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Holds data on responsible organizations ('Zulassungsbehörde', 'Eintragungsstelle') and handler",
)
before_states = models.ManyToManyField(CompensationState, blank=True, related_name='+', help_text="Refers to 'Ausgangszustand Biotop'")
after_states = models.ManyToManyField(CompensationState, blank=True, related_name='+', help_text="Refers to 'Zielzustand Biotop'")
actions = models.ManyToManyField(CompensationAction, blank=True, help_text="Refers to 'Maßnahmen'")
deadlines = models.ManyToManyField("konova.Deadline", blank=True, related_name="+")
geometry = models.ForeignKey(Geometry, null=True, blank=True, on_delete=models.SET_NULL)
class Meta:
abstract = True
def add_new_deadline(self, form) -> Deadline:
""" Adds a new deadline to the abstract compensation
Args:
form (NewDeadlineModalForm): The form holding all relevant data
Returns:
"""
form_data = form.cleaned_data
user = form.user
with transaction.atomic():
created_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.CREATED
)
deadline = Deadline.objects.create(
type=form_data["type"],
date=form_data["date"],
comment=form_data["comment"],
created=created_action,
)
edited_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.EDITED,
comment=_("Added deadline")
)
self.modified = edited_action
self.save()
self.log.add(edited_action)
self.deadlines.add(deadline)
return deadline
def add_new_action(self, form) -> CompensationAction:
""" Adds a new action to the compensation
Args:
form (NewActionModalForm): The form holding all relevant data
Returns:
"""
form_data = form.cleaned_data
user = form.user
with transaction.atomic():
user_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.CREATED,
)
comp_action = CompensationAction.objects.create(
action_type=form_data["action_type"],
amount=form_data["amount"],
unit=form_data["unit"],
comment=form_data["comment"],
created=user_action,
)
edited_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.EDITED,
comment=_("Added action"),
)
self.modified = edited_action
self.save()
self.log.add(edited_action)
self.actions.add(comp_action)
return comp_action
def get_surface_after_states(self) -> float:
""" Calculates the compensation's/account's surface
Returns:
sum_surface (float)
"""
return self._calc_surface(self.after_states.all())
def get_surface_before_states(self) -> float:
""" Calculates the compensation's/account's surface
Returns:
sum_surface (float)
"""
return self._calc_surface(self.before_states.all())
def _calc_surface(self, qs: QuerySet):
""" Calculates the surface sum of a given queryset
Args:
qs (QuerySet): The queryset containing CompensationState entries
Returns:
"""
return qs.aggregate(Sum("surface"))["surface__sum"] or 0
def quality_check(self) -> CompensationQualityChecker:
""" Performs data quality check
Returns:
checker (CompensationQualityChecker): Holds validity data and error messages
"""
checker = CompensationQualityChecker(self)
checker.run_check()
return checker
class CEFMixin(models.Model):
""" Provides CEF flag as Mixin
"""
is_cef = models.BooleanField(
blank=True,
null=True,
default=False,
help_text="Flag if compensation is a 'CEF-Maßnahme'"
)
class Meta:
abstract = True
class CoherenceMixin(models.Model):
""" Provides coherence keeping flag as Mixin
"""
is_coherence_keeping = models.BooleanField(
blank=True,
null=True,
default=False,
help_text="Flag if compensation is a 'Kohärenzsicherung'"
)
class Meta:
abstract = True
class Compensation(AbstractCompensation, CEFMixin, CoherenceMixin):
"""
Regular compensation, linked to an intervention
"""
intervention = models.ForeignKey(
Intervention,
on_delete=models.CASCADE,
null=True,
blank=True,
related_name='compensations'
)
objects = CompensationManager()
def __str__(self):
return "{}".format(self.identifier)
def save(self, *args, **kwargs):
if self.identifier is None or len(self.identifier) == 0:
# Create new identifier is none was given
self.identifier = self.generate_new_identifier()
# Before saving, make sure a given identifier has not been taken already in the meanwhile
while Compensation.objects.filter(identifier=self.identifier).exclude(id=self.id).exists():
self.identifier = self.generate_new_identifier()
super().save(*args, **kwargs)
def is_shared_with(self, user: User):
""" Access check
Checks whether a given user has access to this object
Args:
user (User): The user to be checked
Returns:
"""
# Compensations inherit their shared state from the interventions
return self.intervention.is_shared_with(user)
def get_LANIS_link(self) -> str:
""" Generates a link for LANIS depending on the geometry
Returns:
"""
try:
geom = self.geometry.geom.transform(DEFAULT_SRID_RLP, clone=True)
x = geom.centroid.x
y = geom.centroid.y
zoom_lvl = 16
except AttributeError:
# If no geometry has been added, yet.
x = 1
y = 1
zoom_lvl = 6
return LANIS_LINK_TEMPLATE.format(
zoom_lvl,
x,
y,
)
def get_documents(self) -> QuerySet:
""" Getter for all documents of a compensation
Returns:
docs (QuerySet): The queryset of all documents
"""
docs = CompensationDocument.objects.filter(
instance=self
)
return docs
def add_state(self, form, is_before_state: bool) -> CompensationState:
""" Adds a new compensation state to the compensation
Args:
form (NewStateModalForm): The form, holding all relevant data
is_before_state (bool): Whether this is a new before_state or after_state
Returns:
"""
form_data = form.cleaned_data
user = form.user
with transaction.atomic():
user_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.EDITED,
comment=_("Added state")
)
self.log.add(user_action)
self.modified = user_action
self.save()
state = CompensationState.objects.create(
biotope_type=form_data["biotope_type"],
surface=form_data["surface"],
)
if is_before_state:
self.before_states.add(state)
else:
self.after_states.add(state)
return state
class CompensationDocument(AbstractDocument):
"""
Specializes document upload for revocations with certain path
"""
instance = models.ForeignKey(
Compensation,
on_delete=models.CASCADE,
related_name="documents",
)
file = models.FileField(
upload_to=generate_document_file_upload_path,
max_length=1000,
)
def delete(self, *args, **kwargs):
"""
Custom delete functionality for CompensationDocuments.
Removes the folder from the file system if there are no further documents for this entry.
Args:
*args ():
**kwargs ():
Returns:
"""
comp_docs = self.instance.get_documents()
folder_path = None
if comp_docs.count() == 1:
# The only file left for this compensation is the one which is currently processed and will be deleted
# Make sure that the compensation folder itself is deleted as well, not only the file
# Therefore take the folder path from the file path
folder_path = self.file.path.split("/")[:-1]
folder_path = "/".join(folder_path)
# Remove the file itself
super().delete(*args, **kwargs)
# If a folder path has been set, we need to delete the whole folder!
if folder_path is not None:
try:
shutil.rmtree(folder_path)
except FileNotFoundError:
# Folder seems to be missing already...
pass
class EcoAccount(AbstractCompensation, ShareableObject, RecordableObject):
"""
An eco account is a kind of 'prepaid' compensation. It can be compared to an account that already has been filled
with some kind of currency. From this account one is able to deduct currency for current projects.
"""
from intervention.models import Legal
deductable_surface = models.FloatField(
blank=True,
null=True,
help_text="Amount of deductable surface - can be lower than the total surface due to deduction limitations",
default=0,
)
legal = models.OneToOneField(
Legal,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Holds data on legal dates or law"
)
objects = EcoAccountManager()
def __str__(self):
return "{}".format(self.identifier)
def clean(self):
# Deductable surface can not be larger than added states after surface
after_state_sum = self.get_state_after_surface_sum()
if self.deductable_surface > after_state_sum:
raise ValidationError(_("Deductable surface can not be larger than existing surfaces in after states"))
# Deductable surface can not be lower than amount of already deducted surfaces
# User needs to contact deducting user in case of further problems
deducted_sum = self.get_deductions_surface()
if self.deductable_surface < deducted_sum:
raise ValidationError(
_("Deductable surface can not be smaller than the sum of already existing deductions. Please contact the responsible users for the deductions!")
)
def save(self, *args, **kwargs):
if self.identifier is None or len(self.identifier) == 0:
# Create new identifier if none was given
self.identifier = self.generate_new_identifier()
# Before saving, make sure the given identifier is not used, yet
while EcoAccount.objects.filter(identifier=self.identifier).exclude(id=self.id).exists():
self.identifier = self.generate_new_identifier()
super().save(*args, **kwargs)
@property
def deductions_surface_sum(self) -> float:
""" Shortcut for get_deductions_surface.
Can be used in templates
Returns:
sum_surface (float)
"""
return self.get_deductions_surface()
def get_deductions_surface(self) -> float:
""" Calculates the account's deductions surface sum
Returns:
sum_surface (float)
"""
return self.deductions.all().aggregate(Sum("surface"))["surface__sum"] or 0
def get_state_after_surface_sum(self) -> float:
""" Calculates the account's after state surface sum
Returns:
sum_surface (float)
"""
return self.after_states.all().aggregate(Sum("surface"))["surface__sum"] or 0
def get_available_rest(self) -> (float, float):
""" Calculates available rest surface of the eco account
Args:
Returns:
ret_val_total (float): Total amount
ret_val_relative (float): Amount as percentage (0-100)
"""
deductions = self.deductions.filter(
intervention__deleted=None,
)
deductions_surfaces = deductions.aggregate(Sum("surface"))["surface__sum"] or 0
available_surfaces = self.deductable_surface or deductions_surfaces ## no division by zero
ret_val_total = available_surfaces - deductions_surfaces
if available_surfaces > 0:
ret_val_relative = int((ret_val_total / available_surfaces) * 100)
else:
ret_val_relative = 0
return ret_val_total, ret_val_relative
def get_LANIS_link(self) -> str:
""" Generates a link for LANIS depending on the geometry
Returns:
"""
try:
geom = self.geometry.geom.transform(DEFAULT_SRID_RLP, clone=True)
x = geom.centroid.x
y = geom.centroid.y
zoom_lvl = 16
except AttributeError:
# If no geometry has been added, yet.
x = 1
y = 1
zoom_lvl = 6
return LANIS_LINK_TEMPLATE.format(
zoom_lvl,
x,
y,
)
def quality_check(self) -> EcoAccountQualityChecker:
""" Quality check
Returns:
ret_msgs (EcoAccountQualityChecker): Holds validity and error messages
"""
checker = EcoAccountQualityChecker(self)
checker.run_check()
return checker
def get_documents(self) -> QuerySet:
""" Getter for all documents of an EcoAccount
Returns:
docs (QuerySet): The queryset of all documents
"""
docs = EcoAccountDocument.objects.filter(
instance=self
)
return docs
class EcoAccountDocument(AbstractDocument):
"""
Specializes document upload for revocations with certain path
"""
instance = models.ForeignKey(
EcoAccount,
on_delete=models.CASCADE,
related_name="documents",
)
file = models.FileField(
upload_to=generate_document_file_upload_path,
max_length=1000,
)
def delete(self, *args, **kwargs):
"""
Custom delete functionality for EcoAccountDocuments.
Removes the folder from the file system if there are no further documents for this entry.
Args:
*args ():
**kwargs ():
Returns:
"""
acc_docs = self.instance.get_documents()
folder_path = None
if acc_docs.count() == 1:
# The only file left for this eco account is the one which is currently processed and will be deleted
# Make sure that the compensation folder itself is deleted as well, not only the file
# Therefore take the folder path from the file path
folder_path = self.file.path.split("/")[:-1]
folder_path = "/".join(folder_path)
# Remove the file itself
super().delete(*args, **kwargs)
# If a folder path has been set, we need to delete the whole folder!
if folder_path is not None:
try:
shutil.rmtree(folder_path)
except FileNotFoundError:
# Folder seems to be missing already...
pass
class EcoAccountDeduction(BaseResource):
"""
A deduction object for eco accounts
"""
from intervention.models import Intervention
account = models.ForeignKey(
EcoAccount,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Deducted from",
related_name="deductions",
)
surface = models.FloatField(
null=True,
blank=True,
help_text="Amount deducted (m²)",
validators=[
MinValueValidator(limit_value=0.00),
]
)
intervention = models.ForeignKey(
Intervention,
on_delete=models.CASCADE,
null=True,
blank=True,
help_text="Deducted for",
related_name="deductions",
)
objects = EcoAccountDeductionManager()
def __str__(self):
return "{} of {}".format(self.surface, self.account)

12
compensation/models/__init__.py Normal file
View File

@ -0,0 +1,12 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 16.11.21
"""
from .action import *
from .state import *
from .compensation import *
from .eco_account import *
from .payment import *

66
compensation/models/action.py Normal file
View File

@ -0,0 +1,66 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 16.11.21
"""
from django.db import models
from django.utils.translation import gettext_lazy as _
from codelist.models import KonovaCode
from codelist.settings import CODELIST_COMPENSATION_ACTION_ID
from compensation.managers import CompensationActionManager
from konova.models import BaseResource
class UnitChoices(models.TextChoices):
"""
Predefines units for selection
"""
cm = "cm", _("cm")
m = "m", _("m")
km = "km", _("km")
qm = "qm", _("m²")
ha = "ha", _("ha")
st = "pcs", _("Pieces") # pieces
class CompensationAction(BaseResource):
"""
Compensations include actions like planting trees, refreshing rivers and so on.
"""
action_type = models.ForeignKey(
KonovaCode,
on_delete=models.SET_NULL,
null=True,
blank=True,
limit_choices_to={
"code_lists__in": [CODELIST_COMPENSATION_ACTION_ID],
"is_selectable": True,
"is_archived": False,
}
)
amount = models.FloatField()
unit = models.CharField(max_length=100, null=True, blank=True, choices=UnitChoices.choices)
comment = models.TextField(blank=True, null=True, help_text="Additional comment")
objects = CompensationActionManager()
def __str__(self):
return "{} | {} {}".format(self.action_type, self.amount, self.unit)
@property
def unit_humanize(self):
""" Returns humanized version of enum
Used for template rendering
Returns:
"""
choices = UnitChoices.choices
for choice in choices:
if choice[0] == self.unit:
return choice[1]
return None

336
compensation/models/compensation.py Normal file
View File

@ -0,0 +1,336 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 16.11.21
"""
import shutil
from django.contrib.auth.models import User
from django.db import models, transaction
from django.db.models import QuerySet, Sum
from django.utils.translation import gettext_lazy as _
from compensation.managers import CompensationManager
from compensation.models import CompensationState, CompensationAction
from compensation.utils.quality import CompensationQualityChecker
from intervention.models import Responsibility, Intervention
from konova.models import BaseObject, AbstractDocument, Geometry, Deadline, generate_document_file_upload_path
from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE
from user.models import UserActionLogEntry, UserAction
class AbstractCompensation(BaseObject):
"""
Abstract compensation model which holds basic attributes, shared by subclasses like the regular Compensation,
EMA or EcoAccount.
"""
responsible = models.OneToOneField(
Responsibility,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Holds data on responsible organizations ('Zulassungsbehörde', 'Eintragungsstelle') and handler",
)
before_states = models.ManyToManyField(CompensationState, blank=True, related_name='+', help_text="Refers to 'Ausgangszustand Biotop'")
after_states = models.ManyToManyField(CompensationState, blank=True, related_name='+', help_text="Refers to 'Zielzustand Biotop'")
actions = models.ManyToManyField(CompensationAction, blank=True, help_text="Refers to 'Maßnahmen'")
deadlines = models.ManyToManyField("konova.Deadline", blank=True, related_name="+")
geometry = models.ForeignKey(Geometry, null=True, blank=True, on_delete=models.SET_NULL)
class Meta:
abstract = True
def add_new_deadline(self, form) -> Deadline:
""" Adds a new deadline to the abstract compensation
Args:
form (NewDeadlineModalForm): The form holding all relevant data
Returns:
"""
form_data = form.cleaned_data
user = form.user
with transaction.atomic():
created_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.CREATED
)
deadline = Deadline.objects.create(
type=form_data["type"],
date=form_data["date"],
comment=form_data["comment"],
created=created_action,
)
edited_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.EDITED,
comment=_("Added deadline")
)
self.modified = edited_action
self.save()
self.log.add(edited_action)
self.deadlines.add(deadline)
return deadline
def add_new_action(self, form) -> CompensationAction:
""" Adds a new action to the compensation
Args:
form (NewActionModalForm): The form holding all relevant data
Returns:
"""
form_data = form.cleaned_data
user = form.user
with transaction.atomic():
user_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.CREATED,
)
comp_action = CompensationAction.objects.create(
action_type=form_data["action_type"],
amount=form_data["amount"],
unit=form_data["unit"],
comment=form_data["comment"],
created=user_action,
)
edited_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.EDITED,
comment=_("Added action"),
)
self.modified = edited_action
self.save()
self.log.add(edited_action)
self.actions.add(comp_action)
return comp_action
def get_surface_after_states(self) -> float:
""" Calculates the compensation's/account's surface
Returns:
sum_surface (float)
"""
return self._calc_surface(self.after_states.all())
def get_surface_before_states(self) -> float:
""" Calculates the compensation's/account's surface
Returns:
sum_surface (float)
"""
return self._calc_surface(self.before_states.all())
def _calc_surface(self, qs: QuerySet):
""" Calculates the surface sum of a given queryset
Args:
qs (QuerySet): The queryset containing CompensationState entries
Returns:
"""
return qs.aggregate(Sum("surface"))["surface__sum"] or 0
def quality_check(self) -> CompensationQualityChecker:
""" Performs data quality check
Returns:
checker (CompensationQualityChecker): Holds validity data and error messages
"""
checker = CompensationQualityChecker(self)
checker.run_check()
return checker
class CEFMixin(models.Model):
""" Provides CEF flag as Mixin
"""
is_cef = models.BooleanField(
blank=True,
null=True,
default=False,
help_text="Flag if compensation is a 'CEF-Maßnahme'"
)
class Meta:
abstract = True
class CoherenceMixin(models.Model):
""" Provides coherence keeping flag as Mixin
"""
is_coherence_keeping = models.BooleanField(
blank=True,
null=True,
default=False,
help_text="Flag if compensation is a 'Kohärenzsicherung'"
)
class Meta:
abstract = True
class Compensation(AbstractCompensation, CEFMixin, CoherenceMixin):
"""
Regular compensation, linked to an intervention
"""
intervention = models.ForeignKey(
Intervention,
on_delete=models.CASCADE,
null=True,
blank=True,
related_name='compensations'
)
objects = CompensationManager()
def __str__(self):
return "{}".format(self.identifier)
def save(self, *args, **kwargs):
if self.identifier is None or len(self.identifier) == 0:
# Create new identifier is none was given
self.identifier = self.generate_new_identifier()
# Before saving, make sure a given identifier has not been taken already in the meanwhile
while Compensation.objects.filter(identifier=self.identifier).exclude(id=self.id).exists():
self.identifier = self.generate_new_identifier()
super().save(*args, **kwargs)
def is_shared_with(self, user: User):
""" Access check
Checks whether a given user has access to this object
Args:
user (User): The user to be checked
Returns:
"""
# Compensations inherit their shared state from the interventions
return self.intervention.is_shared_with(user)
def get_LANIS_link(self) -> str:
""" Generates a link for LANIS depending on the geometry
Returns:
"""
try:
geom = self.geometry.geom.transform(DEFAULT_SRID_RLP, clone=True)
x = geom.centroid.x
y = geom.centroid.y
zoom_lvl = 16
except AttributeError:
# If no geometry has been added, yet.
x = 1
y = 1
zoom_lvl = 6
return LANIS_LINK_TEMPLATE.format(
zoom_lvl,
x,
y,
)
def get_documents(self) -> QuerySet:
""" Getter for all documents of a compensation
Returns:
docs (QuerySet): The queryset of all documents
"""
docs = CompensationDocument.objects.filter(
instance=self
)
return docs
def add_state(self, form, is_before_state: bool) -> CompensationState:
""" Adds a new compensation state to the compensation
Args:
form (NewStateModalForm): The form, holding all relevant data
is_before_state (bool): Whether this is a new before_state or after_state
Returns:
"""
form_data = form.cleaned_data
user = form.user
with transaction.atomic():
user_action = UserActionLogEntry.objects.create(
user=user,
action=UserAction.EDITED,
comment=_("Added state")
)
self.log.add(user_action)
self.modified = user_action
self.save()
state = CompensationState.objects.create(
biotope_type=form_data["biotope_type"],
surface=form_data["surface"],
)
if is_before_state:
self.before_states.add(state)
else:
self.after_states.add(state)
return state
class CompensationDocument(AbstractDocument):
"""
Specializes document upload for revocations with certain path
"""
instance = models.ForeignKey(
Compensation,
on_delete=models.CASCADE,
related_name="documents",
)
file = models.FileField(
upload_to=generate_document_file_upload_path,
max_length=1000,
)
def delete(self, *args, **kwargs):
"""
Custom delete functionality for CompensationDocuments.
Removes the folder from the file system if there are no further documents for this entry.
Args:
*args ():
**kwargs ():
Returns:
"""
comp_docs = self.instance.get_documents()
folder_path = None
if comp_docs.count() == 1:
# The only file left for this compensation is the one which is currently processed and will be deleted
# Make sure that the compensation folder itself is deleted as well, not only the file
# Therefore take the folder path from the file path
folder_path = self.file.path.split("/")[:-1]
folder_path = "/".join(folder_path)
# Remove the file itself
super().delete(*args, **kwargs)
# If a folder path has been set, we need to delete the whole folder!
if folder_path is not None:
try:
shutil.rmtree(folder_path)
except FileNotFoundError:
# Folder seems to be missing already...
pass

248
compensation/models/eco_account.py Normal file
View File

@ -0,0 +1,248 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 16.11.21
"""
import shutil
from django.core.exceptions import ValidationError
from django.core.validators import MinValueValidator
from django.db import models
from django.db.models import Sum, QuerySet
from django.utils.translation import gettext_lazy as _
from compensation.managers import EcoAccountManager, EcoAccountDeductionManager
from compensation.models.compensation import AbstractCompensation
from compensation.utils.quality import EcoAccountQualityChecker
from konova.models import ShareableObjectMixin, RecordableObjectMixin, AbstractDocument, BaseResource, \
generate_document_file_upload_path
from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE
from intervention.models import Intervention, Legal
class EcoAccount(AbstractCompensation, ShareableObjectMixin, RecordableObjectMixin):
"""
An eco account is a kind of 'prepaid' compensation. It can be compared to an account that already has been filled
with some kind of currency. From this account one is able to deduct currency for current projects.
"""
deductable_surface = models.FloatField(
blank=True,
null=True,
help_text="Amount of deductable surface - can be lower than the total surface due to deduction limitations",
default=0,
)
legal = models.OneToOneField(
Legal,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Holds data on legal dates or law"
)
objects = EcoAccountManager()
def __str__(self):
return "{}".format(self.identifier)
def clean(self):
# Deductable surface can not be larger than added states after surface
after_state_sum = self.get_state_after_surface_sum()
if self.deductable_surface > after_state_sum:
raise ValidationError(_("Deductable surface can not be larger than existing surfaces in after states"))
# Deductable surface can not be lower than amount of already deducted surfaces
# User needs to contact deducting user in case of further problems
deducted_sum = self.get_deductions_surface()
if self.deductable_surface < deducted_sum:
raise ValidationError(
_("Deductable surface can not be smaller than the sum of already existing deductions. Please contact the responsible users for the deductions!")
)
def save(self, *args, **kwargs):
if self.identifier is None or len(self.identifier) == 0:
# Create new identifier if none was given
self.identifier = self.generate_new_identifier()
# Before saving, make sure the given identifier is not used, yet
while EcoAccount.objects.filter(identifier=self.identifier).exclude(id=self.id).exists():
self.identifier = self.generate_new_identifier()
super().save(*args, **kwargs)
@property
def deductions_surface_sum(self) -> float:
""" Shortcut for get_deductions_surface.
Can be used in templates
Returns:
sum_surface (float)
"""
return self.get_deductions_surface()
def get_deductions_surface(self) -> float:
""" Calculates the account's deductions surface sum
Returns:
sum_surface (float)
"""
return self.deductions.all().aggregate(Sum("surface"))["surface__sum"] or 0
def get_state_after_surface_sum(self) -> float:
""" Calculates the account's after state surface sum
Returns:
sum_surface (float)
"""
return self.after_states.all().aggregate(Sum("surface"))["surface__sum"] or 0
def get_available_rest(self) -> (float, float):
""" Calculates available rest surface of the eco account
Args:
Returns:
ret_val_total (float): Total amount
ret_val_relative (float): Amount as percentage (0-100)
"""
deductions = self.deductions.filter(
intervention__deleted=None,
)
deductions_surfaces = deductions.aggregate(Sum("surface"))["surface__sum"] or 0
available_surfaces = self.deductable_surface or deductions_surfaces ## no division by zero
ret_val_total = available_surfaces - deductions_surfaces
if available_surfaces > 0:
ret_val_relative = int((ret_val_total / available_surfaces) * 100)
else:
ret_val_relative = 0
return ret_val_total, ret_val_relative
def get_LANIS_link(self) -> str:
""" Generates a link for LANIS depending on the geometry
Returns:
"""
try:
geom = self.geometry.geom.transform(DEFAULT_SRID_RLP, clone=True)
x = geom.centroid.x
y = geom.centroid.y
zoom_lvl = 16
except AttributeError:
# If no geometry has been added, yet.
x = 1
y = 1
zoom_lvl = 6
return LANIS_LINK_TEMPLATE.format(
zoom_lvl,
x,
y,
)
def quality_check(self) -> EcoAccountQualityChecker:
""" Quality check
Returns:
ret_msgs (EcoAccountQualityChecker): Holds validity and error messages
"""
checker = EcoAccountQualityChecker(self)
checker.run_check()
return checker
def get_documents(self) -> QuerySet:
""" Getter for all documents of an EcoAccount
Returns:
docs (QuerySet): The queryset of all documents
"""
docs = EcoAccountDocument.objects.filter(
instance=self
)
return docs
class EcoAccountDocument(AbstractDocument):
"""
Specializes document upload for revocations with certain path
"""
instance = models.ForeignKey(
EcoAccount,
on_delete=models.CASCADE,
related_name="documents",
)
file = models.FileField(
upload_to=generate_document_file_upload_path,
max_length=1000,
)
def delete(self, *args, **kwargs):
"""
Custom delete functionality for EcoAccountDocuments.
Removes the folder from the file system if there are no further documents for this entry.
Args:
*args ():
**kwargs ():
Returns:
"""
acc_docs = self.instance.get_documents()
folder_path = None
if acc_docs.count() == 1:
# The only file left for this eco account is the one which is currently processed and will be deleted
# Make sure that the compensation folder itself is deleted as well, not only the file
# Therefore take the folder path from the file path
folder_path = self.file.path.split("/")[:-1]
folder_path = "/".join(folder_path)
# Remove the file itself
super().delete(*args, **kwargs)
# If a folder path has been set, we need to delete the whole folder!
if folder_path is not None:
try:
shutil.rmtree(folder_path)
except FileNotFoundError:
# Folder seems to be missing already...
pass
class EcoAccountDeduction(BaseResource):
"""
A deduction object for eco accounts
"""
account = models.ForeignKey(
EcoAccount,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Deducted from",
related_name="deductions",
)
surface = models.FloatField(
null=True,
blank=True,
help_text="Amount deducted (m²)",
validators=[
MinValueValidator(limit_value=0.00),
]
)
intervention = models.ForeignKey(
Intervention,
on_delete=models.CASCADE,
null=True,
blank=True,
help_text="Deducted for",
related_name="deductions",
)
objects = EcoAccountDeductionManager()
def __str__(self):
return "{} of {}".format(self.surface, self.account)

38
compensation/models/payment.py Normal file
View File

@ -0,0 +1,38 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 16.11.21
"""
from django.core.validators import MinValueValidator
from django.db import models
from intervention.models import Intervention
from konova.models import BaseResource
class Payment(BaseResource):
"""
Holds data on a payment for an intervention (alternative to a classic compensation)
"""
amount = models.FloatField(validators=[MinValueValidator(limit_value=0.00)])
due_on = models.DateField(null=True, blank=True)
comment = models.TextField(
null=True,
blank=True,
help_text="Refers to german money transfer 'Verwendungszweck'",
)
intervention = models.ForeignKey(
Intervention,
null=True,
blank=True,
on_delete=models.CASCADE,
related_name='payments'
)
class Meta:
ordering = [
"-amount",
]

36
compensation/models/state.py Normal file
View File

@ -0,0 +1,36 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 16.11.21
"""
from django.db import models
from codelist.models import KonovaCode
from codelist.settings import CODELIST_BIOTOPES_ID
from compensation.managers import CompensationStateManager
from konova.models import UuidModel
class CompensationState(UuidModel):
"""
Compensations must define the state of an area before and after the compensation.
"""
biotope_type = models.ForeignKey(
KonovaCode,
on_delete=models.SET_NULL,
null=True,
blank=True,
limit_choices_to={
"code_lists__in": [CODELIST_BIOTOPES_ID],
"is_selectable": True,
"is_archived": False,
}
)
surface = models.FloatField()
objects = CompensationStateManager()
def __str__(self):
return "{} | {} m²".format(self.biotope_type, self.surface)

4
ema/models/ema.py
View File

@ -13,11 +13,11 @@ from django.db.models import QuerySet
from compensation.models import AbstractCompensation
from ema.managers import EmaManager
from ema.utils.quality import EmaQualityChecker
from konova.models import AbstractDocument, generate_document_file_upload_path, RecordableObject, ShareableObject
from konova.models import AbstractDocument, generate_document_file_upload_path, RecordableObjectMixin, ShareableObjectMixin
from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE
class Ema(AbstractCompensation, ShareableObject, RecordableObject):
class Ema(AbstractCompensation, ShareableObjectMixin, RecordableObjectMixin):
"""
EMA = Ersatzzahlungsmaßnahme
(compensation actions from payments)

9
intervention/models/intervention.py
View File

@ -5,6 +5,8 @@ Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 15.11.21
"""
import shutil
from django.contrib.auth.models import User
from django.db import models, transaction
from django.db.models import QuerySet
@ -15,13 +17,13 @@ from intervention.models.legal import Legal
from intervention.models.responsibility import Responsibility
from intervention.models.revocation import RevocationDocument
from intervention.utils.quality import InterventionQualityChecker
from konova.models import generate_document_file_upload_path, AbstractDocument, Geometry, BaseObject, ShareableObject, \
RecordableObject, CheckableObject
from konova.models import generate_document_file_upload_path, AbstractDocument, Geometry, BaseObject, ShareableObjectMixin, \
RecordableObjectMixin, CheckableObjectMixin
from konova.settings import LANIS_LINK_TEMPLATE, LANIS_ZOOM_LUT, DEFAULT_SRID_RLP
from user.models import UserAction, UserActionLogEntry
class Intervention(BaseObject, ShareableObject, RecordableObject, CheckableObject):
class Intervention(BaseObject, ShareableObjectMixin, RecordableObjectMixin, CheckableObjectMixin):
"""
Interventions are e.g. construction sites where nature used to be.
"""
@ -251,7 +253,6 @@ class InterventionDocument(AbstractDocument):
if folder_path is not None:
try:
shutil.rmtree(folder_path)
pass
except FileNotFoundError:
# Folder seems to be missing already...
pass

6
konova/forms.py
View File

@ -22,7 +22,7 @@ from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from konova.contexts import BaseContext
from konova.models import BaseObject, Geometry, RecordableObject
from konova.models import BaseObject, Geometry, RecordableObjectMixin
from konova.settings import DEFAULT_SRID
from konova.utils.message_templates import FORM_INVALID
from user.models import UserActionLogEntry, UserAction
@ -290,7 +290,7 @@ class SimpleGeomForm(BaseForm):
geometry.geom = self.cleaned_data.get("geom", MultiPolygon(srid=DEFAULT_SRID))
geometry.modified = action
geometry.save()
except (AttributeError) as e:
except AttributeError:
# No geometry or linked instance holding a geometry exist --> create a new one!
geometry = Geometry.objects.create(
geom=self.cleaned_data.get("geom", MultiPolygon(srid=DEFAULT_SRID)),
@ -446,7 +446,7 @@ class RecordModalForm(BaseModalForm):
self.form_title = _("Unrecord data")
self.form_caption = _("I, {} {}, confirm that this data must be unrecorded.").format(self.user.first_name, self.user.last_name)
if not isinstance(self.instance, RecordableObject):
if not isinstance(self.instance, RecordableObjectMixin):
raise NotImplementedError
def is_valid(self):

10
konova/models/object.py
View File

@ -71,7 +71,7 @@ class BaseResource(UuidModel):
"""
try:
self.created.delete()
except (ObjectDoesNotExist, AttributeError) as e:
except (ObjectDoesNotExist, AttributeError):
# Object does not exist anymore - we can skip this
pass
super().delete()
@ -146,7 +146,7 @@ class BaseObject(BaseResource):
Returns:
"""
if isinstance(self, ShareableObject):
if isinstance(self, ShareableObjectMixin):
return self.users.filter(id=user.id)
else:
return User.objects.none()
@ -220,7 +220,7 @@ class BaseObject(BaseResource):
return definitions[self.__class__]["template"].format(_str)
class RecordableObject(models.Model):
class RecordableObjectMixin(models.Model):
""" Wraps record related fields and functionality
"""
@ -289,7 +289,7 @@ class RecordableObject(models.Model):
return ret_log_entry
class CheckableObject(models.Model):
class CheckableObjectMixin(models.Model):
# Checks - Refers to "Genehmigen" but optional
checked = models.OneToOneField(
UserActionLogEntry,
@ -351,7 +351,7 @@ class CheckableObject(models.Model):
return ret_log_entry
class ShareableObject(models.Model):
class ShareableObjectMixin(models.Model):
# Users having access on this object
users = models.ManyToManyField(User, help_text="Users having access (data shared with)")
access_token = models.CharField(