#6556

pending completion

Build # #6556

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Bump future from 0.18.2 to 0.18.3 in /requirements (#3792)

Bumps [future](https://github.com/PythonCharmers/python-future) from 0.18.2 to 0.18.3.
- [Release notes](https://github.com/PythonCharmers/python-future/releases)
- [Changelog](https://github.com/PythonCharmers/python-future/blob/master/docs/changelog.rst)
- [Commits](https://github.com/PythonCharmers/python-future/compare/v0.18.2...v0.18.3)

---
updated-dependencies:
- dependency-name: future
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

15695 of 22613 relevant lines covered (69.41%)

0.69 hits per line

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

27.83

/seed/models/compliance_metrics.py

# encoding: utf-8
"""
:copyright (c) 2014 - 2022, The Regents of the University of California, through Lawrence Berkeley National Laboratory (subject to receipt of any required approvals from the U.S. Department of Energy) and contributors. All rights reserved.
"""


from typing import Union

from django.db import models
from django.http import QueryDict

from seed.lib.superperms.orgs.models import Organization
from seed.models.columns import Column
from seed.models.cycles import Cycle
from seed.models.filter_group import FilterGroup
from seed.utils.properties import properties_across_cycles_with_filters


class ComplianceMetric(models.Model):

    TARGET_NONE = 0
    TARGET_GT_ACTUAL = 1  # example: GHG, Site EUI
    TARGET_LT_ACTUAL = 2  # example: EnergyStar Score
    METRIC_TYPES = (
        (TARGET_NONE, ''),
        (TARGET_GT_ACTUAL, 'Target > Actual for Compliance'),
        (TARGET_LT_ACTUAL, 'Target < Actual for Compliance'),
    )

    organization = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name='compliance_metrics', blank=True, null=True)
    name = models.CharField(max_length=255)
    start = models.DateTimeField(null=True, blank=True)  # only care about year, but adding as a DateTime
    end = models.DateTimeField(null=True, blank=True)  # only care about year, but adding as a DateTime
    created = models.DateTimeField(auto_now_add=True)
    # TODO: could these be derived columns?
    actual_energy_column = models.ForeignKey(Column, related_name="actual_energy_column", null=True, on_delete=models.CASCADE)
    target_energy_column = models.ForeignKey(Column, related_name="target_energy_column", null=True, on_delete=models.CASCADE)
    energy_metric_type = models.IntegerField(choices=METRIC_TYPES, blank=True, null=True)
    actual_emission_column = models.ForeignKey(Column, related_name="actual_emission_column", null=True, on_delete=models.CASCADE)
    target_emission_column = models.ForeignKey(Column, related_name="target_emission_column", null=True, on_delete=models.CASCADE)
    emission_metric_type = models.IntegerField(choices=METRIC_TYPES, blank=True, null=True)
    filter_group = models.ForeignKey(FilterGroup, related_name="filter_group", null=True, on_delete=models.CASCADE)
    x_axis_columns = models.ManyToManyField(Column, related_name="x_axis_columns", blank=True)

    def __str__(self):
        return 'Program Metric - %s' % self.name

    def evaluate(self):
        response = {
            'meta': {
                'organization': self.organization.id,
                'compliance_metric': self.id,
            },
            'name': self.name,
            'graph_data': {
                'labels': [],
                'datasets': []
            },
            'cycles': []
        }

        query_dict = QueryDict(mutable=True)
        if self.filter_group and self.filter_group.query_dict:
            query_dict.update(self.filter_group.query_dict)
        # print(f"query dict: {query_dict}")

        # grab cycles within start and end dates
        cycles = Cycle.objects.filter(organization_id=self.organization.id, start__lte=self.end, end__gte=self.start).order_by('start')
        cycle_ids = cycles.values_list('pk', flat=True)
        response['graph_data']['labels'] = list(cycles.values_list('name', flat=True))
        response['cycles'] = list(cycles.values('id', 'name'))

        # get properties (no filter)
        # property_response = properties_across_cycles(self.organization_id, -1, cycle_ids)
        # get properties (applies filter group)
        display_field_id = Column.objects.get(table_name="PropertyState", column_name=self.organization.property_display_field, organization=self.organization).id
        # array of columns to return
        column_ids = [
            display_field_id
        ]

        if self.actual_energy_column is not None:
            column_ids.append(self.actual_energy_column.id)
            if self.target_energy_column is not None:
                column_ids.append(self.target_energy_column.id)

        if self.actual_emission_column is not None:
            column_ids.append(self.actual_emission_column.id)
            if self.target_emission_column is not None:
                column_ids.append(self.target_emission_column.id)

        for col in self.x_axis_columns.all():
            column_ids.append(col.id)

        property_response = properties_across_cycles_with_filters(
            self.organization_id,
            cycle_ids,
            query_dict,
            column_ids
        )

        datasets = {'y': {'data': [], 'label': 'compliant'}, 'n': {'data': [], 'label': 'non-compliant'}, 'u': {'data': [], 'label': 'unknown'}}
        results_by_cycles = {}
#        property_datasets = {}
        # figure out what kind of metric it is (energy? emission? combo? bool?)
        metric = {
            'energy_metric': False,
            'emission_metric': False,
            'energy_bool': False,
            'emission_bool': False,
            'actual_energy_column': None,
            'actual_energy_column_name': None,
            'target_energy_column': None,
            'energy_metric_type': self.energy_metric_type,
            'actual_emission_column': None,
            'actual_emission_column_name': None,
            'target_emission_column': None,
            'emission_metric_type': self.emission_metric_type,
            'fliter_group': None,
            'x_axis_columns': list(self.x_axis_columns.all().values('id', 'display_name'))}

        if self.actual_energy_column is not None:
            metric['actual_energy_column'] = self.actual_energy_column.id
            metric['actual_energy_column_name'] = self.actual_energy_column.display_name
            metric['energy_metric'] = True
            if self.target_energy_column is None:
                metric['energy_bool'] = True
            else:
                metric['target_energy_column'] = self.target_energy_column.id

        if self.actual_emission_column is not None:
            metric['emission_metric'] = True
            metric['actual_emission_column'] = self.actual_emission_column.id
            metric['actual_emission_column_name'] = self.actual_emission_column.display_name
            if self.target_emission_column is None:
                metric['emission_bool'] = True
            else:
                metric['target_emission_column'] = self.target_emission_column.id

        for cyc in property_response:
            properties = {}
            cnts = {'y': 0, 'n': 0, 'u': 0}

            for p in property_response[cyc]:
                # initialize
                properties[p['property_view_id']] = None
                # energy metric
                if metric['energy_metric']:
                    properties[p['property_view_id']] = self._calculate_compliance(p, metric['energy_bool'], 'energy')
                # emission metric
                if metric['emission_metric'] and properties[p['property_view_id']] != 'u':
                    temp_val = self._calculate_compliance(p, metric['emission_bool'], 'emission')

                    # reconcile
                    if temp_val == 'u':
                        # unknown stays unknown (missing data)
                        properties[p['property_view_id']] = 'u'
                    elif properties[p['property_view_id']] is None:
                        # only emission metric (not energy metric)
                        properties[p['property_view_id']] = temp_val
                    else:
                        # compliant if both are compliant
                        properties[p['property_view_id']] = temp_val if temp_val == 'n' else properties[p['property_view_id']]

            # count compliant, non-compliant, unknown for each property with data
            for key in cnts:
                cnts[key] = sum(map((key).__eq__, properties.values()))
                # add to dataset
                datasets[key]['data'].append(cnts[key])

            # reshape and save
            results_by_cycles[cyc] = {}
            for key in cnts:
                results_by_cycles[cyc][key] = [i for i in properties if properties[i] == key]

        # save to response
        response['results_by_cycles'] = results_by_cycles
        response['properties_by_cycles'] = property_response
        response['metric'] = metric

        for key in datasets:
            response['graph_data']['datasets'].append(datasets[key])

        return response

    def _calculate_compliance(self, the_property, bool_metric, metric_type):
        """Return the compliant, non-compliant, and unknown counts

        Args:
            the_property (PropertyState): The property state object to check compliance
            bool_metric (Boolean): If the metric is a boolean metric, otherwise, it is a target metric
            metric_type (Int): Target > Actual or Target < Actual

        Returns:
            string: single character string representing compliance status, u - unknown, y - compliant, n - non-compliant
        """
        actual_col = self.actual_energy_column if metric_type == 'energy' else self.actual_emission_column
        target_col = self.target_energy_column if metric_type == 'energy' else self.target_emission_column
        actual_val = self._get_column_data(the_property, actual_col)
        if actual_val is None:
            return 'u'

        if bool_metric:
            return 'y' if actual_val > 0 else 'n'

        target_val = self._get_column_data(the_property, target_col)
        if target_val is None:
            return 'u'

        # test metric type
        the_type = self.energy_metric_type if metric_type == 'energy' else self.emission_metric_type
        # 1 = target is less than actual, 2 = target is greater than actual
        # TODO: convert int to enum types for readability
        if the_type == 1:
            differential = target_val - actual_val
        else:
            differential = actual_val - target_val

        return 'y' if differential >= 0 else 'n'

    def _get_column_data(self, data: dict, column: Column) -> Union[float, bool]:
        """Get the column datat from the dictionary version of the property state.
        Also, cast the datatype based on the column data_type as needed.

        Args:
            data (dict): property state dictionary
            column (Column): column object

        Returns:
            Union[float, bool]: the resulting value
        """
        # retrieves column data from the property state. The lookup is
        # based on the <column_name>_<column_id> format because the data
        # is the flat dictionary representation of the property state.
        column_lookup = f"{column.column_name}_{column.id}"
        value = data[column_lookup]

        # Now cast it based on the column data_type, this uses
        # the same method that is covered in the search.py file for
        # consistency
        return column.cast(value)

    class Meta:
        ordering = ['-created']
        get_latest_by = 'created'

1	# encoding: utf-8
2	"""	1✔
3	:copyright (c) 2014 - 2022, The Regents of the University of California, through Lawrence Berkeley National Laboratory (subject to receipt of any required approvals from the U.S. Department of Energy) and contributors. All rights reserved.
4	"""
5
6
7	from typing import Union	1✔
8
9	from django.db import models
10	from django.http import QueryDict
11
12	from seed.lib.superperms.orgs.models import Organization	1✔
13	from seed.models.columns import Column	1✔
14	from seed.models.cycles import Cycle	1✔
15	from seed.models.filter_group import FilterGroup	1✔
16	from seed.utils.properties import properties_across_cycles_with_filters	1✔
17
18
19	class ComplianceMetric(models.Model):	1✔
20
21	TARGET_NONE = 0	1✔
22	TARGET_GT_ACTUAL = 1 # example: GHG, Site EUI	1✔
23	TARGET_LT_ACTUAL = 2 # example: EnergyStar Score	1✔
24	METRIC_TYPES = (	1✔
25	(TARGET_NONE, ''),
26	(TARGET_GT_ACTUAL, 'Target > Actual for Compliance'),
27	(TARGET_LT_ACTUAL, 'Target < Actual for Compliance'),
28	)
29
30	organization = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name='compliance_metrics', blank=True, null=True)	1✔
31	name = models.CharField(max_length=255)	1✔
32	start = models.DateTimeField(null=True, blank=True) # only care about year, but adding as a DateTime	1✔
33	end = models.DateTimeField(null=True, blank=True) # only care about year, but adding as a DateTime	1✔
34	created = models.DateTimeField(auto_now_add=True)	1✔
35	# TODO: could these be derived columns?
36	actual_energy_column = models.ForeignKey(Column, related_name="actual_energy_column", null=True, on_delete=models.CASCADE)	1✔
37	target_energy_column = models.ForeignKey(Column, related_name="target_energy_column", null=True, on_delete=models.CASCADE)	1✔
38	energy_metric_type = models.IntegerField(choices=METRIC_TYPES, blank=True, null=True)	1✔
39	actual_emission_column = models.ForeignKey(Column, related_name="actual_emission_column", null=True, on_delete=models.CASCADE)	1✔
40	target_emission_column = models.ForeignKey(Column, related_name="target_emission_column", null=True, on_delete=models.CASCADE)	1✔
41	emission_metric_type = models.IntegerField(choices=METRIC_TYPES, blank=True, null=True)	1✔
42	filter_group = models.ForeignKey(FilterGroup, related_name="filter_group", null=True, on_delete=models.CASCADE)	1✔
43	x_axis_columns = models.ManyToManyField(Column, related_name="x_axis_columns", blank=True)	1✔
44
45	def __str__(self):	1✔
46	return 'Program Metric - %s' % self.name	×
47
48	def evaluate(self):	1✔
49	response = {	×
50	'meta': {
51	'organization': self.organization.id,
52	'compliance_metric': self.id,
53	},
54	'name': self.name,
55	'graph_data': {
56	'labels': [],
57	'datasets': []
58	},
59	'cycles': []
60	}
61
62	query_dict = QueryDict(mutable=True)	×
63	if self.filter_group and self.filter_group.query_dict:	×
64	query_dict.update(self.filter_group.query_dict)	×
65	# print(f"query dict: {query_dict}")
66
67	# grab cycles within start and end dates
68	cycles = Cycle.objects.filter(organization_id=self.organization.id, start__lte=self.end, end__gte=self.start).order_by('start')	×
69	cycle_ids = cycles.values_list('pk', flat=True)	×
70	response['graph_data']['labels'] = list(cycles.values_list('name', flat=True))	×
71	response['cycles'] = list(cycles.values('id', 'name'))	×
72
73	# get properties (no filter)
74	# property_response = properties_across_cycles(self.organization_id, -1, cycle_ids)
75	# get properties (applies filter group)
76	display_field_id = Column.objects.get(table_name="PropertyState", column_name=self.organization.property_display_field, organization=self.organization).id	×
77	# array of columns to return
78	column_ids = [	×
79	display_field_id
80	]
81
82	if self.actual_energy_column is not None:	×
83	column_ids.append(self.actual_energy_column.id)	×
84	if self.target_energy_column is not None:	×
85	column_ids.append(self.target_energy_column.id)	×
86
87	if self.actual_emission_column is not None:	×
88	column_ids.append(self.actual_emission_column.id)	×
89	if self.target_emission_column is not None:	×
90	column_ids.append(self.target_emission_column.id)	×
91
92	for col in self.x_axis_columns.all():	×
93	column_ids.append(col.id)	×
94
95	property_response = properties_across_cycles_with_filters(	×
96	self.organization_id,
97	cycle_ids,
98	query_dict,
99	column_ids
100	)
101
102	datasets = {'y': {'data': [], 'label': 'compliant'}, 'n': {'data': [], 'label': 'non-compliant'}, 'u': {'data': [], 'label': 'unknown'}}	×
103	results_by_cycles = {}	×
104	# property_datasets = {}
105	# figure out what kind of metric it is (energy? emission? combo? bool?)
106	metric = {	×
107	'energy_metric': False,
108	'emission_metric': False,
109	'energy_bool': False,
110	'emission_bool': False,
111	'actual_energy_column': None,
112	'actual_energy_column_name': None,
113	'target_energy_column': None,
114	'energy_metric_type': self.energy_metric_type,
115	'actual_emission_column': None,
116	'actual_emission_column_name': None,
117	'target_emission_column': None,
118	'emission_metric_type': self.emission_metric_type,
119	'fliter_group': None,
120	'x_axis_columns': list(self.x_axis_columns.all().values('id', 'display_name'))}
121
122	if self.actual_energy_column is not None:	×
123	metric['actual_energy_column'] = self.actual_energy_column.id	×
124	metric['actual_energy_column_name'] = self.actual_energy_column.display_name	×
125	metric['energy_metric'] = True	×
126	if self.target_energy_column is None:	×
127	metric['energy_bool'] = True	×
128	else:
129	metric['target_energy_column'] = self.target_energy_column.id	×
130
131	if self.actual_emission_column is not None:	×
132	metric['emission_metric'] = True	×
133	metric['actual_emission_column'] = self.actual_emission_column.id	×
134	metric['actual_emission_column_name'] = self.actual_emission_column.display_name	×
135	if self.target_emission_column is None:	×
136	metric['emission_bool'] = True	×
137	else:
138	metric['target_emission_column'] = self.target_emission_column.id	×
139
140	for cyc in property_response:	×
141	properties = {}	×
142	cnts = {'y': 0, 'n': 0, 'u': 0}	×
143
144	for p in property_response[cyc]:	×
145	# initialize
146	properties[p['property_view_id']] = None	×
147	# energy metric
148	if metric['energy_metric']:	×
149	properties[p['property_view_id']] = self._calculate_compliance(p, metric['energy_bool'], 'energy')	×
150	# emission metric
151	if metric['emission_metric'] and properties[p['property_view_id']] != 'u':	×
152	temp_val = self._calculate_compliance(p, metric['emission_bool'], 'emission')	×
153
154	# reconcile
155	if temp_val == 'u':	×
156	# unknown stays unknown (missing data)
157	properties[p['property_view_id']] = 'u'	×
158	elif properties[p['property_view_id']] is None:	×
159	# only emission metric (not energy metric)
160	properties[p['property_view_id']] = temp_val	×
161	else:
162	# compliant if both are compliant
163	properties[p['property_view_id']] = temp_val if temp_val == 'n' else properties[p['property_view_id']]	×
164
165	# count compliant, non-compliant, unknown for each property with data
166	for key in cnts:	×
167	cnts[key] = sum(map((key).__eq__, properties.values()))	×
168	# add to dataset
169	datasets[key]['data'].append(cnts[key])	×
170
171	# reshape and save
172	results_by_cycles[cyc] = {}	×
173	for key in cnts:	×
174	results_by_cycles[cyc][key] = [i for i in properties if properties[i] == key]	×
175
176	# save to response
177	response['results_by_cycles'] = results_by_cycles	×
178	response['properties_by_cycles'] = property_response	×
179	response['metric'] = metric	×
180
181	for key in datasets:	×
182	response['graph_data']['datasets'].append(datasets[key])	×
183
184	return response	×
185
186	def _calculate_compliance(self, the_property, bool_metric, metric_type):	1✔
187	"""Return the compliant, non-compliant, and unknown counts
188
189	Args:
190	the_property (PropertyState): The property state object to check compliance
191	bool_metric (Boolean): If the metric is a boolean metric, otherwise, it is a target metric
192	metric_type (Int): Target > Actual or Target < Actual
193
194	Returns:
195	string: single character string representing compliance status, u - unknown, y - compliant, n - non-compliant
196	"""
197	actual_col = self.actual_energy_column if metric_type == 'energy' else self.actual_emission_column	×
198	target_col = self.target_energy_column if metric_type == 'energy' else self.target_emission_column	×
199	actual_val = self._get_column_data(the_property, actual_col)	×
200	if actual_val is None:	×
201	return 'u'	×
202
203	if bool_metric:	×
204	return 'y' if actual_val > 0 else 'n'	×
205
206	target_val = self._get_column_data(the_property, target_col)	×
207	if target_val is None:	×
208	return 'u'	×
209
210	# test metric type
211	the_type = self.energy_metric_type if metric_type == 'energy' else self.emission_metric_type	×
212	# 1 = target is less than actual, 2 = target is greater than actual
213	# TODO: convert int to enum types for readability
214	if the_type == 1:	×
215	differential = target_val - actual_val	×
216	else:
217	differential = actual_val - target_val	×
218
219	return 'y' if differential >= 0 else 'n'	×
220
221	def _get_column_data(self, data: dict, column: Column) -> Union[float, bool]:	1✔
222	"""Get the column datat from the dictionary version of the property state.
223	Also, cast the datatype based on the column data_type as needed.
224
225	Args:
226	data (dict): property state dictionary
227	column (Column): column object
228
229	Returns:
230	Union[float, bool]: the resulting value
231	"""
232	# retrieves column data from the property state. The lookup is
233	# based on the <column_name>_<column_id> format because the data
234	# is the flat dictionary representation of the property state.
235	column_lookup = f"{column.column_name}_{column.id}"	×
236	value = data[column_lookup]	×
237
238	# Now cast it based on the column data_type, this uses
239	# the same method that is covered in the search.py file for
240	# consistency
241	return column.cast(value)	×
242
243	class Meta:	1✔
244	ordering = ['-created']	1✔
245	get_latest_by = 'created'	1✔

SEED-platform / seed / #6556

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