summaryrefslogtreecommitdiff
path: root/archaeological_operations/models.py
diff options
context:
space:
mode:
Diffstat (limited to 'archaeological_operations/models.py')
-rw-r--r--archaeological_operations/models.py531
1 files changed, 0 insertions, 531 deletions
diff --git a/archaeological_operations/models.py b/archaeological_operations/models.py
index 9e038adeb..57f538383 100644
--- a/archaeological_operations/models.py
+++ b/archaeological_operations/models.py
@@ -3299,537 +3299,6 @@ class ParcelOwner(LightHistorizedItem):
return self.parcel.associated_file
-class OperationDashboard:
- def __init__(self):
- main_dashboard = Dashboard(Operation)
-
- self.total_number = main_dashboard.total_number
-
- self.filters_keys = [
- "recorded",
- "effective",
- "active",
- "field",
- "documented",
- "closed",
- "documented_closed",
- ]
- filters = {
- "recorded": {},
- "effective": {"scientist__isnull": False},
- "active": {"scientist__isnull": False, "end_date__isnull": True},
- "field": {"excavation_end_date__isnull": True},
- "documented": {"documents__isnull": False},
- "documented_closed": {
- "documents__isnull": False,
- "end_date__isnull": False,
- },
- "closed": {"end_date__isnull": False},
- }
- filters_label = {
- "recorded": _("Recorded"),
- "effective": _("Effective"),
- "active": _("Active"),
- "field": _("Field completed"),
- "documented": _("Associated report"),
- "closed": _("Closed"),
- "documented_closed": _("Documented and closed"),
- }
- self.filters_label = [filters_label[k] for k in self.filters_keys]
- self.total = []
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- nb = Operation.objects.filter(**fltr).count()
- self.total.append((lbl, nb))
-
- self.surface_by_type = (
- Operation.objects.values("operation_type__label")
- .annotate(number=Sum("surface"))
- .order_by("-number", "operation_type__label")
- )
-
- self.by_type = []
- self.types = OperationType.objects.filter(available=True).all()
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- type_res = (
- Operation.objects.filter(**fltr)
- .values("operation_type", "operation_type__label")
- .annotate(number=Count("pk"))
- .order_by("operation_type")
- )
- types_dct = {}
- for typ in type_res.all():
- types_dct[typ["operation_type"]] = typ["number"]
- types = []
- for typ in self.types:
- if typ.pk in types_dct:
- types.append(types_dct[typ.pk])
- else:
- types.append(0)
- self.by_type.append((lbl, types))
-
- self.by_year = []
- self.years = [
- res["year"]
- for res in Operation.objects.values("year").order_by("-year").distinct()
- ]
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- year_res = (
- Operation.objects.filter(**fltr)
- .values("year")
- .annotate(number=Count("pk"))
- .order_by("year")
- )
- years_dct = {}
- for yr in year_res.all():
- years_dct[yr["year"]] = yr["number"]
- years = []
- for yr in self.years:
- if yr in years_dct:
- years.append(years_dct[yr])
- else:
- years.append(0)
- self.by_year.append((lbl, years))
-
- self.by_realisation_year = []
- self.realisation_years = [
- res["date"]
- for res in Operation.objects.extra(
- {"date": "date_trunc('year', start_date)"}
- )
- .values("date")
- .filter(start_date__isnull=False)
- .order_by("-date")
- .distinct()
- ]
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- year_res = (
- Operation.objects.filter(**fltr)
- .extra({"date": "date_trunc('year', start_date)"})
- .values("date")
- .values("date")
- .filter(start_date__isnull=False)
- .annotate(number=Count("pk"))
- .order_by("-date")
- )
- years_dct = {}
- for yr in year_res.all():
- years_dct[yr["date"]] = yr["number"]
- years = []
- for yr in self.realisation_years:
- if yr in years_dct:
- years.append(years_dct[yr])
- else:
- years.append(0)
- self.by_realisation_year.append((lbl, years))
-
- self.effective = []
- for typ in self.types:
- year_res = (
- Operation.objects.filter(
- **{"scientist__isnull": False, "operation_type": typ}
- )
- .values("year")
- .annotate(number=Count("pk"))
- .order_by("-year")
- .distinct()
- )
- years_dct = {}
- for yr in year_res.all():
- years_dct[yr["year"]] = yr["number"]
- years = []
- for yr in self.years:
- if yr in years_dct:
- years.append(years_dct[yr])
- else:
- years.append(0)
- self.effective.append((typ, years))
-
- # TODO: by date
- now = datetime.date.today()
- limit = datetime.date(now.year, now.month, 1) - datetime.timedelta(365)
- by_realisation_month = Operation.objects.filter(
- start_date__gt=limit, start_date__isnull=False
- ).extra({"date": "date_trunc('month', start_date)"})
- self.last_months = []
- date = datetime.datetime(now.year, now.month, 1)
- for mt_idx in range(12):
- self.last_months.append(date)
- if date.month > 1:
- date = datetime.datetime(date.year, date.month - 1, 1)
- else:
- date = datetime.datetime(date.year - 1, 12, 1)
- self.by_realisation_month = []
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- month_res = (
- by_realisation_month.filter(**fltr)
- .annotate(number=Count("pk"))
- .order_by("-date")
- )
- month_dct = {}
- for mt in month_res.all():
- month_dct[mt.date] = mt.number
- date = datetime.date(now.year, now.month, 1)
- months = []
- for date in self.last_months:
- if date in month_dct:
- months.append(month_dct[date])
- else:
- months.append(0)
- self.by_realisation_month.append((lbl, months))
-
- # survey and excavations
- self.survey, self.excavation = {}, {}
- for dct_res, ope_types in (
- (self.survey, ("arch_diagnostic",)),
- (self.excavation, ("prev_excavation", "prog_excavation")),
- ):
- dct_res["total"] = []
- operation_type = {"operation_type__txt_idx__in": ope_types}
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- fltr.update(operation_type)
- nb = Operation.objects.filter(**fltr).count()
- dct_res["total"].append((lbl, nb))
-
- dct_res["by_year"] = []
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- fltr.update(operation_type)
- year_res = (
- Operation.objects.filter(**fltr)
- .values("year")
- .annotate(number=Count("pk"))
- .order_by("year")
- )
- years_dct = {}
- for yr in year_res.all():
- years_dct[yr["year"]] = yr["number"]
- years = []
- for yr in self.years:
- if yr in years_dct:
- years.append(years_dct[yr])
- else:
- years.append(0)
- dct_res["by_year"].append((lbl, years))
-
- dct_res["by_realisation_year"] = []
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- fltr.update(operation_type)
- year_res = (
- Operation.objects.filter(**fltr)
- .extra({"date": "date_trunc('year', start_date)"})
- .values("date")
- .filter(start_date__isnull=False)
- .annotate(number=Count("pk"))
- .order_by("-date")
- )
- years_dct = {}
- for yr in year_res.all():
- years_dct[yr["date"]] = yr["number"]
- years = []
- for yr in self.realisation_years:
- if yr in years_dct:
- years.append(years_dct[yr])
- else:
- years.append(0)
- dct_res["by_realisation_year"].append((lbl, years))
-
- current_year_ope = Operation.objects.filter(**operation_type).filter(
- year=datetime.date.today().year
- )
- current_realisation_year_ope = Operation.objects.filter(
- **operation_type
- ).filter(start_date__year=datetime.date.today().year)
- res_keys = [("area_realised", current_realisation_year_ope)]
- if dct_res == self.survey:
- res_keys.append(("area", current_year_ope))
- for res_key, base_ope in res_keys:
- dct_res[res_key] = []
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- area_res = (
- base_ope.filter(**fltr).annotate(number=Sum("surface")).all()
- )
- val = 0
- if area_res:
- val = (area_res[0].number or 0) / 10000.0
- dct_res[res_key].append(val)
- # TODO...
- res_keys = [("manday_realised", current_realisation_year_ope)]
- if dct_res == self.survey:
- res_keys.append(("manday", current_year_ope))
- for res_key, base_ope in res_keys:
- dct_res[res_key] = []
- for fltr_key in self.filters_keys:
- dct_res[res_key].append("-")
- # TODO...
- res_keys = [("mandayhect_realised", current_realisation_year_ope)]
- if dct_res == self.survey:
- res_keys.append(("mandayhect", current_year_ope))
- for res_key, base_ope in res_keys:
- dct_res[res_key] = []
- for fltr_key in self.filters_keys:
- dct_res[res_key].append("-")
- # TODO...
- dct_res["mandayhect_real_effective"] = "-"
- if dct_res == self.survey:
- dct_res["mandayhect_effective"] = "-"
-
- res_keys = [("org_realised", current_realisation_year_ope)]
- if dct_res == self.survey:
- res_keys.append(("org", current_year_ope))
- for res_key, base_ope in res_keys:
- org_res = (
- base_ope.filter(scientist__attached_to__isnull=False)
- .values("scientist__attached_to", "scientist__attached_to__name")
- .annotate(area=Sum("surface"))
- .order_by("scientist__attached_to__name")
- .all()
- )
- # TODO: man-days, man-days/hectare
-
- dct_res[res_key] = []
- for vals in org_res:
- vals["area"] = (vals["area"] or 0) / 10000.0
- dct_res[res_key].append(vals)
-
- year_ope = Operation.objects.filter(**operation_type)
- res_keys = ["org_by_year"]
- if dct_res == self.survey:
- res_keys.append("org_by_year_realised")
- q = (
- year_ope.values(
- "scientist__attached_to", "scientist__attached_to__name"
- )
- .filter(scientist__attached_to__isnull=False)
- .order_by("scientist__attached_to__name")
- .distinct()
- )
- org_list = [
- (org["scientist__attached_to"], org["scientist__attached_to__name"])
- for org in q
- ]
- # org_list_dct = dict(org_list)
- for res_key in res_keys:
- dct_res[res_key] = []
- years = self.years
- if res_key == "org_by_year_realised":
- years = self.realisation_years
- for org_id, org_label in org_list:
- org_res = year_ope.filter(scientist__attached_to__pk=org_id)
- key_date = ""
- if res_key == "org_by_year":
- org_res = org_res.values("year")
- key_date = "year"
- else:
- org_res = (
- org_res.extra({"date": "date_trunc('year', start_date)"})
- .values("date")
- .filter(start_date__isnull=False)
- )
- key_date = "date"
- org_res = org_res.annotate(area=Sum("surface"), cost=Sum("cost"))
- years_dct = {}
- for yr in org_res.all():
- area = (yr["area"] if yr["area"] else 0) / 10000.0
- cost = yr["cost"] if yr["cost"] else 0
- years_dct[yr[key_date]] = (area, cost)
- r_years = []
- for yr in years:
- if yr in years_dct:
- r_years.append(years_dct[yr])
- else:
- r_years.append((0, 0))
- dct_res[res_key].append((org_label, r_years))
- area_means, area_sums = [], []
- cost_means, cost_sums = [], []
- for idx, year in enumerate(years):
- vals = [r_yars[idx] for lb, r_yars in dct_res[res_key]]
- if not vals:
- continue
- sum_area = sum([a for a, c in vals])
- sum_cost = sum([c for a, c in vals])
- area_means.append(sum_area / len(vals))
- area_sums.append(sum_area)
- cost_means.append(sum_cost / len(vals))
- cost_sums.append(sum_cost)
- dct_res[res_key + "_area_mean"] = area_means
- dct_res[res_key + "_area_sum"] = area_sums
- dct_res[res_key + "_cost_mean"] = cost_means
- dct_res[res_key + "_cost_mean"] = cost_sums
-
- if dct_res == self.survey:
- self.survey["effective"] = []
- for yr in self.years:
- year_res = Operation.objects.filter(
- scientist__isnull=False,
- year=yr,
- operation_type__txt_idx__in=ope_types,
- ).annotate(number=Sum("surface"), mean=Avg("surface"))
- nb = year_res[0].number if year_res.count() else 0
- nb = nb if nb else 0
- mean = year_res[0].mean if year_res.count() else 0
- mean = mean if mean else 0
- self.survey["effective"].append((nb, mean))
-
- # TODO:Man-Days/hectare by Year
-
- # CHECK: month of realisation or month?
- dct_res["by_month"] = []
- for fltr_key in self.filters_keys:
- fltr, lbl = filters[fltr_key], filters_label[fltr_key]
- fltr.update(operation_type)
- month_res = (
- by_realisation_month.filter(**fltr)
- .annotate(number=Count("pk"))
- .order_by("-date")
- )
- month_dct = {}
- for mt in month_res.all():
- month_dct[mt.date] = mt.number
- date = datetime.date(now.year, now.month, 1)
- months = []
- for date in self.last_months:
- if date in month_dct:
- months.append(month_dct[date])
- else:
- months.append(0)
- dct_res["by_month"].append((lbl, months))
-
- operation_type = {"operation_type__txt_idx__in": ope_types}
- self.departments = [
- (fd["department__pk"], fd["department__label"])
- for fd in OperationByDepartment.objects.filter(department__isnull=False)
- .values("department__label", "department__pk")
- .order_by("department__label")
- .distinct()
- ]
- dct_res["by_dpt"] = []
- for dpt_id, dpt_label in self.departments:
- vals = (
- OperationByDepartment.objects.filter(
- department__pk=dpt_id,
- operation__operation_type__txt_idx__in=ope_types,
- )
- .values("department__pk", "operation__year")
- .annotate(number=Count("operation"))
- .order_by("operation__year")
- )
- dct_years = {}
- for v in vals:
- dct_years[v["operation__year"]] = v["number"]
- years = []
- for y in self.years:
- if y in dct_years:
- years.append(dct_years[y])
- else:
- years.append(0)
- years.append(sum(years))
- dct_res["by_dpt"].append((dpt_label, years))
- dct_res["effective_by_dpt"] = []
- for dpt_id, dpt_label in self.departments:
- vals = (
- OperationByDepartment.objects.filter(
- department__pk=dpt_id,
- operation__scientist__isnull=False,
- operation__operation_type__txt_idx__in=ope_types,
- )
- .values("department__pk", "operation__year")
- .annotate(
- number=Count("operation"),
- area=Sum("operation__surface"),
- fnap=Sum("operation__fnap_cost"),
- cost=Sum("operation__cost"),
- )
- .order_by("operation__year")
- )
- dct_years = {}
- for v in vals:
- values = []
- for k in ("number", "area", "cost", "fnap"):
- value = v[k] or 0
- if k == "area":
- value /= 10000.0
- values.append(value)
-
- dct_years[v["operation__year"]] = values
- years = []
- for y in self.years:
- if y in dct_years:
- years.append(dct_years[y])
- else:
- years.append((0, 0, 0, 0))
- nbs, areas, costs, fnaps = zip(*years)
- years.append((sum(nbs), sum(areas), sum(costs), sum(fnaps)))
- dct_res["effective_by_dpt"].append((dpt_label, years))
-
- OperationTown = Operation.towns.through
- query = (
- OperationTown.objects.filter(
- operation__scientist__isnull=False,
- operation__operation_type__txt_idx__in=ope_types,
- )
- .values("town__name", "town__departement__number")
- .annotate(nb=Count("operation"))
- .order_by("-nb", "town__name")[:10]
- )
- dct_res["towns"] = []
- for r in query:
- dct_res["towns"].append(
- (
- "%s (%s)" % (r["town__name"], r["town__departement__number"]),
- r["nb"],
- )
- )
-
- if dct_res == self.survey:
- query = (
- OperationTown.objects.filter(
- operation__scientist__isnull=False,
- operation__operation_type__txt_idx__in=ope_types,
- operation__surface__isnull=False,
- )
- .values("town__name", "town__departement__number")
- .annotate(nb=Sum("operation__surface"))
- .order_by("-nb", "town__name")[:10]
- )
- dct_res["towns_surface"] = []
- for r in query:
- dct_res["towns_surface"].append(
- (
- "%s (%s)"
- % (r["town__name"], r["town__departement__number"]),
- r["nb"],
- )
- )
- else:
- query = (
- OperationTown.objects.filter(
- operation__scientist__isnull=False,
- operation__operation_type__txt_idx__in=ope_types,
- operation__cost__isnull=False,
- )
- .values("town__name", "town__departement__number")
- .annotate(nb=Sum("operation__cost"))
- .order_by("-nb", "town__name")[:10]
- )
- dct_res["towns_cost"] = []
- for r in query:
- dct_res["towns_cost"].append(
- (
- "%s (%s)"
- % (r["town__name"], r["town__departement__number"]),
- r["nb"],
- )
- )
-
-
class OperationTypeOld(GeneralType):
order = models.IntegerField(_("Order"), default=1)
preventive = models.BooleanField(_("Is preventive"), default=True)
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-16 17:05:20 +0000