#9061

Committed 15 Aug 2025 06:59PM UTC coverage: 86.707% (-0.02%) from 86.723%

Build # #9061

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 8f5eca031 into d61ea3f70

Run Details

9 of 10 new or added lines in 2 files covered. (90.0%)

75 existing lines in 4 files now uncovered.

41981 of 48417 relevant lines covered (86.71%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.65

/seed/models/facilities_plan.py

"""
SEED Platform (TM), Copyright (c) Alliance for Sustainable Energy, LLC, and other contributors.
See also https://github.com/SEED-platform/seed/blob/main/LICENSE.md
"""

import logging

from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
from django.db.models import BooleanField, Case, CharField, F, FloatField, Sum, Value, When, Q
from django.db.models.functions import Cast, Coalesce, Replace
from django.utils import timezone as tz

from seed.lib.superperms.orgs.models import AccessLevelInstance, Organization
from seed.models import Column, Cycle, PropertyView

logger = logging.getLogger(__name__)


class FacilitiesPlan(models.Model):
    organization = models.ForeignKey(Organization, on_delete=models.CASCADE)
    name = models.CharField(max_length=255)

    energy_running_sum_percentage = models.FloatField(default=0.75, validators=[MinValueValidator(0.0), MaxValueValidator(1.0)])
    compliance_cycle_year_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")
    include_in_total_denominator_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")
    exclude_from_plan_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")
    require_in_plan_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")

    electric_energy_usage_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")
    gas_energy_usage_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")
    steam_energy_usage_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")

    class Meta:
        ordering = ["name"]
        constraints = [
            models.UniqueConstraint(fields=["organization", "name"], name="unique_name_for_plan"),
        ]


def _get_column_or_zero(column):

    c = _get_column_model_field(column)

    if column.is_extra_data or column.derived_column:
        # For JSONB fields, we need to handle string-to-number conversion properly
        # First cast to CharField to extract the string value, then to FloatField
        return Case(
            When(
                Q(**{f"{c}__isnull": False}) &
                Q(**{f"{c}__regex": r'^"*-?\d+(\.\d+)?"*$'}),
                then=Cast(
                    Replace(
                        Replace(Cast(F(c), CharField()), Value('"'), Value('')),
                        Value('"'), Value('')
                    ),
                    FloatField()
                )
            ),
            default=Value(0.0),
            output_field=FloatField(),
        )
    else:
        # For regular model fields
        return Case(
            When(**{f"{c}__isnull": False}, then=Cast(F(c), FloatField())),
            default=Value(0.0),
            output_field=FloatField(),
        )


def _get_column_model_field(column):
    if column.is_extra_data:
        return "state__extra_data__" + column.column_name
    elif column.derived_column:
        return "state__derived_data__" + column.column_name
    else:
        return "state__" + column.column_name


class FacilitiesPlanRun(models.Model):
    facilities_plan = models.ForeignKey(FacilitiesPlan, on_delete=models.CASCADE, related_name="runs")
    cycle = models.ForeignKey(Cycle, on_delete=models.SET_NULL, null=True)
    ali = models.ForeignKey(AccessLevelInstance, on_delete=models.SET_NULL, null=True)
    run_at = models.DateTimeField(auto_now=False, blank=True, null=True)
    name = models.CharField(max_length=255, blank=True, null=True)
    display_columns = models.ManyToManyField(Column)

    def _calculate_properties_percentage_of_total_energy_usage(self, ali: AccessLevelInstance, cycle: Cycle):
        # check that we have all the required columns
        required_columns = [
            "electric_energy_usage_column",
            "gas_energy_usage_column",
            "steam_energy_usage_column",
            "include_in_total_denominator_column",
            "require_in_plan_column",
            "exclude_from_plan_column",
        ]
        missing_columns = [c for c in required_columns if getattr(self.facilities_plan, c) is None]
        if missing_columns:
            raise ValueError(f"`calculate_properties_selected_by_plan` requires the following null columns: {missing_columns}")

        # get relevant properties
        properties = PropertyView.objects.filter(
            property__access_level_instance__lft__gte=ali.lft, property__access_level_instance__rgt__lte=ali.rgt, cycle=cycle
        ).exclude(**{_get_column_model_field(self.facilities_plan.include_in_total_denominator_column): False})

        # calculate properties total energy usage
        properties = properties.annotate(
            total_energy_usage=_get_column_or_zero(self.facilities_plan.electric_energy_usage_column)
            + _get_column_or_zero(self.facilities_plan.gas_energy_usage_column)
            + _get_column_or_zero(self.facilities_plan.steam_energy_usage_column)
        )

        # get floor area
        properties = properties.annotate(gross_floor_area=Coalesce("state__gross_floor_area", 0, output_field=FloatField()))

        # calculate properties percentage of total energy usage
        denominator = properties.aggregate(Sum("total_energy_usage"))["total_energy_usage__sum"]
        properties = properties.annotate(
            percentage_of_total_energy_usage=Cast(F("total_energy_usage"), FloatField()) / denominator,
        )

        # calculate required_in_plan (We're weeding out the nones, which mess up ordering later)
        properties = properties.annotate(
            required_in_plan=Case(
                When(**{_get_column_model_field(self.facilities_plan.require_in_plan_column): True}, then=Value(True)),
                default=Value(False),
                output_field=BooleanField(),
            )
        )
        properties = properties.annotate(
            exclude_from_plan_column=Case(
                When(**{_get_column_model_field(self.facilities_plan.exclude_from_plan_column): True}, then=Value(True)),
                default=Value(False),
                output_field=BooleanField(),
            )
        )

        return properties

    def run(self):
        FacilitiesPlanRunProperty.objects.filter(run=self).all().delete()
        self.run_at = tz.now()
        self.save()

        all_properties = self._calculate_properties_percentage_of_total_energy_usage(self.ali, self.cycle).order_by(
            "exclude_from_plan_column", "-required_in_plan", "-percentage_of_total_energy_usage"
        )

        energy_running_sum_percentage = 0
        running_square_footage = 0

        for rank, p in enumerate(all_properties):
            energy_running_sum_percentage += p.percentage_of_total_energy_usage
            running_square_footage += p.gross_floor_area

            FacilitiesPlanRunProperty.objects.create(
                run=self,
                view=p,
                total_energy_usage=p.total_energy_usage,
                percentage_of_total_energy_usage=p.percentage_of_total_energy_usage,
                rank=rank,
                running_percentage=energy_running_sum_percentage,
                running_square_footage=running_square_footage,
            )


class FacilitiesPlanRunProperty(models.Model):
    run = models.ForeignKey(FacilitiesPlanRun, on_delete=models.SET_NULL, null=True, related_name="property_rankings")
    view = models.ForeignKey(PropertyView, on_delete=models.SET_NULL, null=True, related_name="facility_plan_runs")

    total_energy_usage = models.FloatField()
    percentage_of_total_energy_usage = models.FloatField()
    rank = models.IntegerField()
    running_percentage = models.FloatField()
    running_square_footage = models.FloatField()

1	"""
2	SEED Platform (TM), Copyright (c) Alliance for Sustainable Energy, LLC, and other contributors.
3	See also https://github.com/SEED-platform/seed/blob/main/LICENSE.md
4	"""
5
6	import logging	1✔
7
8	from django.core.validators import MaxValueValidator, MinValueValidator
9	from django.db import models
10	from django.db.models import BooleanField, Case, CharField, F, FloatField, Sum, Value, When, Q
11	from django.db.models.functions import Cast, Coalesce, Replace
12	from django.utils import timezone as tz
13
14	from seed.lib.superperms.orgs.models import AccessLevelInstance, Organization	1✔
15	from seed.models import Column, Cycle, PropertyView	1✔
16
17	logger = logging.getLogger(__name__)	1✔
18
19
20	class FacilitiesPlan(models.Model):	1✔
21	organization = models.ForeignKey(Organization, on_delete=models.CASCADE)	1✔
22	name = models.CharField(max_length=255)	1✔
23
24	energy_running_sum_percentage = models.FloatField(default=0.75, validators=[MinValueValidator(0.0), MaxValueValidator(1.0)])	1✔
25	compliance_cycle_year_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
26	include_in_total_denominator_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
27	exclude_from_plan_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
28	require_in_plan_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
29
30	electric_energy_usage_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
31	gas_energy_usage_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
32	steam_energy_usage_column = models.ForeignKey(Column, on_delete=models.SET_NULL, blank=True, null=True, related_name="+")	1✔
33
34	class Meta:	1✔
35	ordering = ["name"]	1✔
36	constraints = [	1✔
37	models.UniqueConstraint(fields=["organization", "name"], name="unique_name_for_plan"),
38	]
39
40
41	def _get_column_or_zero(column):	1✔
42
43	c = _get_column_model_field(column)	1✔
44
45	if column.is_extra_data or column.derived_column:	1✔
46	# For JSONB fields, we need to handle string-to-number conversion properly
47	# First cast to CharField to extract the string value, then to FloatField
48	return Case(	1✔
49	When(
50	Q(**{f"{c}__isnull": False}) &
51	Q(*{f"{c}__regex": r'^"-?\d+(\.\d+)?"*$'}),
52	then=Cast(
53	Replace(
54	Replace(Cast(F(c), CharField()), Value('"'), Value('')),
55	Value('"'), Value('')
56	),
57	FloatField()
58	)
59	),
60	default=Value(0.0),
61	output_field=FloatField(),
62	)
63	else:
64	# For regular model fields
NEW 65	return Case(	×
66	When(**{f"{c}__isnull": False}, then=Cast(F(c), FloatField())),
67	default=Value(0.0),
68	output_field=FloatField(),
69	)
70
71
72	def _get_column_model_field(column):	1✔
73	if column.is_extra_data:	1✔
74	return "state__extra_data__" + column.column_name	1✔
UNCOV 75	elif column.derived_column:	×
UNCOV 76	return "state__derived_data__" + column.column_name	×
77	else:
UNCOV 78	return "state__" + column.column_name	×
79
80
81	class FacilitiesPlanRun(models.Model):	1✔
82	facilities_plan = models.ForeignKey(FacilitiesPlan, on_delete=models.CASCADE, related_name="runs")	1✔
83	cycle = models.ForeignKey(Cycle, on_delete=models.SET_NULL, null=True)	1✔
84	ali = models.ForeignKey(AccessLevelInstance, on_delete=models.SET_NULL, null=True)	1✔
85	run_at = models.DateTimeField(auto_now=False, blank=True, null=True)	1✔
86	name = models.CharField(max_length=255, blank=True, null=True)	1✔
87	display_columns = models.ManyToManyField(Column)	1✔
88
89	def _calculate_properties_percentage_of_total_energy_usage(self, ali: AccessLevelInstance, cycle: Cycle):	1✔
90	# check that we have all the required columns
91	required_columns = [	1✔
92	"electric_energy_usage_column",
93	"gas_energy_usage_column",
94	"steam_energy_usage_column",
95	"include_in_total_denominator_column",
96	"require_in_plan_column",
97	"exclude_from_plan_column",
98	]
99	missing_columns = [c for c in required_columns if getattr(self.facilities_plan, c) is None]	1✔
100	if missing_columns:	1✔
UNCOV 101	raise ValueError(f"`calculate_properties_selected_by_plan` requires the following null columns: {missing_columns}")	×
102
103	# get relevant properties
104	properties = PropertyView.objects.filter(	1✔
105	property__access_level_instance__lft__gte=ali.lft, property__access_level_instance__rgt__lte=ali.rgt, cycle=cycle
106	).exclude(**{_get_column_model_field(self.facilities_plan.include_in_total_denominator_column): False})
107
108	# calculate properties total energy usage
109	properties = properties.annotate(	1✔
110	total_energy_usage=_get_column_or_zero(self.facilities_plan.electric_energy_usage_column)
111	+ _get_column_or_zero(self.facilities_plan.gas_energy_usage_column)
112	+ _get_column_or_zero(self.facilities_plan.steam_energy_usage_column)
113	)
114
115	# get floor area
116	properties = properties.annotate(gross_floor_area=Coalesce("state__gross_floor_area", 0, output_field=FloatField()))	1✔
117
118	# calculate properties percentage of total energy usage
119	denominator = properties.aggregate(Sum("total_energy_usage"))["total_energy_usage__sum"]	1✔
120	properties = properties.annotate(	1✔
121	percentage_of_total_energy_usage=Cast(F("total_energy_usage"), FloatField()) / denominator,
122	)
123
124	# calculate required_in_plan (We're weeding out the nones, which mess up ordering later)
125	properties = properties.annotate(	1✔
126	required_in_plan=Case(
127	When(**{_get_column_model_field(self.facilities_plan.require_in_plan_column): True}, then=Value(True)),
128	default=Value(False),
129	output_field=BooleanField(),
130	)
131	)
132	properties = properties.annotate(	1✔
133	exclude_from_plan_column=Case(
134	When(**{_get_column_model_field(self.facilities_plan.exclude_from_plan_column): True}, then=Value(True)),
135	default=Value(False),
136	output_field=BooleanField(),
137	)
138	)
139
140	return properties	1✔
141
142	def run(self):	1✔
143	FacilitiesPlanRunProperty.objects.filter(run=self).all().delete()	1✔
144	self.run_at = tz.now()	1✔
145	self.save()	1✔
146
147	all_properties = self._calculate_properties_percentage_of_total_energy_usage(self.ali, self.cycle).order_by(	1✔
148	"exclude_from_plan_column", "-required_in_plan", "-percentage_of_total_energy_usage"
149	)
150
151	energy_running_sum_percentage = 0	1✔
152	running_square_footage = 0	1✔
153
154	for rank, p in enumerate(all_properties):	1✔
155	energy_running_sum_percentage += p.percentage_of_total_energy_usage	1✔
156	running_square_footage += p.gross_floor_area	1✔
157
158	FacilitiesPlanRunProperty.objects.create(	1✔
159	run=self,
160	view=p,
161	total_energy_usage=p.total_energy_usage,
162	percentage_of_total_energy_usage=p.percentage_of_total_energy_usage,
163	rank=rank,
164	running_percentage=energy_running_sum_percentage,
165	running_square_footage=running_square_footage,
166	)
167
168
169	class FacilitiesPlanRunProperty(models.Model):	1✔
170	run = models.ForeignKey(FacilitiesPlanRun, on_delete=models.SET_NULL, null=True, related_name="property_rankings")	1✔
171	view = models.ForeignKey(PropertyView, on_delete=models.SET_NULL, null=True, related_name="facility_plan_runs")	1✔
172
173	total_energy_usage = models.FloatField()	1✔
174	percentage_of_total_energy_usage = models.FloatField()	1✔
175	rank = models.IntegerField()	1✔
176	running_percentage = models.FloatField()	1✔
177	running_square_footage = models.FloatField()	1✔

SEED-platform / seed / #9061

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous