• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

SEED-platform / seed / #6426

pending completion
#6426

push

coveralls-python

web-flow
Merge pull request #3682 from SEED-platform/Fix-legend

Fix legend

15655 of 22544 relevant lines covered (69.44%)

0.69 hits per line

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

12.99
/seed/analysis_pipelines/better/client.py
1
import json
1✔
2
import logging
1✔
3
from tempfile import NamedTemporaryFile, TemporaryDirectory
1✔
4

5
import polling
1✔
6
import requests
1✔
7
from django.conf import settings
8

9
from seed.analysis_pipelines.pipeline import AnalysisPipelineException
1✔
10

11
logger = logging.getLogger(__name__)
1✔
12

13

14
class BETTERClient:
1✔
15

16
    HOST = settings.BETTER_HOST
1✔
17
    API_URL = f'{HOST}/api/v1'
1✔
18

19
    def __init__(self, token):
1✔
20
        self._token = f'Token {token}'
×
21

22
    def token_is_valid(self):
1✔
23
        """Returns true if token is valid
24

25
        :return: bool
26
        """
27
        url = f'{self.API_URL}/verify_token'
×
28
        headers = {
×
29
            'accept': 'application/json',
30
            'Authorization': self._token,
31
        }
32

33
        try:
×
34
            response = requests.request("GET", url, headers=headers)
×
35
            return response.status_code == 200
×
36
        except Exception:
×
37
            return False
×
38

39
    def get_buildings(self):
1✔
40
        """Get list of all buildings
41

42
        :return: tuple(list[dict], list[str]), list of buildings followed by list of errors
43
        """
44
        url = f'{self.API_URL}/buildings/'
×
45
        headers = {
×
46
            'accept': 'application/json',
47
            'Authorization': self._token,
48
        }
49

50
        try:
×
51
            response = requests.request("GET", url, headers=headers)
×
52
            if response.status_code != 200:
×
53
                return None, [f'Expected 200 response from BETTER but got {response.status_code}: {response.content}']
×
54
        except Exception as e:
×
55
            return None, [f'Unexpected error creating BETTER portfolio: {e}']
×
56

57
        return response.json(), []
×
58

59
    def create_portfolio(self, name):
1✔
60
        """Create a new BETTER portfolio
61

62
        :param name: str, portfolio name
63
        :returns: tuple(int, list[str]), portfolio id followed by list of errors
64
        """
65
        url = f'{self.API_URL}/portfolios/'
×
66
        data = {
×
67
            'name': name,
68
            'portfolio_currency': 'USD'
69
        }
70
        headers = {
×
71
            'accept': 'application/json',
72
            'Authorization': self._token,
73
        }
74

75
        try:
×
76
            response = requests.request("POST", url, headers=headers, data=data)
×
77
            if response.status_code == 201:
×
78
                data = response.json()
×
79
                portfolio_id = data['id']
×
80
            else:
81
                return None, [f'Expected 201 response from BETTER but got {response.status_code}: {response.content}']
×
82
        except Exception as e:
×
83
            return None, [f'Unexpected error creating BETTER portfolio: {e}']
×
84

85
        return portfolio_id, []
×
86

87
    def create_portfolio_analysis(self, better_portfolio_id, analysis_config):
1✔
88
        """Create an analysis for the portfolio.
89

90
        :param better_portfolio_id: int
91
        :param analysis_config: dict, Used as analysis configuration, should be structured
92
            according to API requirements
93
        :return: tuple(int, list[str]), ID of analysis followed by list of error messages
94
        """
95
        url = f'{self.API_URL}/portfolios/{better_portfolio_id}/analytics/'
×
96
        data = dict(analysis_config)
×
97
        data.update({'building_ids': 'ALL'})
×
98
        headers = {
×
99
            'accept': 'application/json',
100
            'Authorization': self._token,
101
        }
102

103
        try:
×
104
            response = requests.request("POST", url, headers=headers, data=data)
×
105
            if response.status_code == 201:
×
106
                data = response.json()
×
107
                logger.info(f'CREATED Analysis: {data}')
×
108
                analysis_id = data['id']
×
109
            else:
110
                return None, [f'Expected 201 response from BETTER but got {response.status_code}: {response.content}']
×
111
        except Exception as e:
×
112
            return None, [f'Unexpected error creating BETTER portfolio analysis: {e}']
×
113

114
        return analysis_id, []
×
115

116
    def get_portfolio_analysis(self, better_portfolio_id, better_analysis_id):
1✔
117
        """Get portfolio analysis as dict
118

119
        :param better_portfolio_id: int
120
        :param better_analysis_id: int, ID of analysis created for the portfolio
121
        :return: tuple(dict, list[str]), JSON response as dict followed by list of error messages
122
        """
123
        url = f'{self.API_URL}/portfolios/{better_portfolio_id}/analytics/{better_analysis_id}/'
×
124
        headers = {
×
125
            'accept': 'application/json',
126
            'Authorization': self._token,
127
        }
128

129
        try:
×
130
            response = requests.request("GET", url, headers=headers)
×
131
            if response.status_code != 200:
×
132
                return None, [f'Expected 200 response from BETTER but got {response.status_code}: {response.content}']
×
133
        except Exception as e:
×
134
            return None, [f'Unexpected error getting BETTER portfolio analysis: {e}']
×
135

136
        return response.json(), []
×
137

138
    def run_portfolio_analysis(self, better_portfolio_id, better_analysis_id):
1✔
139
        """Start portfolio analysis and wait for it to finish.
140

141
        :param better_portfolio_id: int
142
        :param better_analysis_id: int, ID of analysis created for the portfolio
143
        :return: list[str], list of error messages
144
        """
145
        url = f'{self.API_URL}/portfolios/{better_portfolio_id}/analytics/{better_analysis_id}/generate/'
×
146
        headers = {
×
147
            'accept': 'application/json',
148
            'Authorization': self._token,
149
        }
150

151
        try:
×
152
            response = requests.request("GET", url, headers=headers)
×
153
            if response.status_code != 200:
×
154
                return [f'Expected 200 response from BETTER but got {response.status_code}: {response.content}']
×
155
        except Exception as e:
×
156
            return [f'Unexpected error generating BETTER portfolio analysis: {e}']
×
157

158
        # Gotta make sure the analysis is done
159
        def is_ready(res):
×
160
            """
161
            :param res: response tuple from get_portfolio_analysis
162
            :return: bool
163
            """
164
            response, errors = res[0], res[1]
×
165
            if errors:
×
166
                raise Exception('; '.join(errors))
×
167

168
            if response['generation_result'] == 'COMPLETE':
×
169
                return True
×
170
            elif response['generation_result'] == 'FAILED':
×
171
                raise Exception(f'BETTER failed to generate the portfolio analysis: {response}')
×
172
            else:
173
                return False
×
174

175
        POLLING_TIMEOUT_SECS = 300
×
176
        try:
×
177
            polling.poll(
×
178
                lambda: self.get_portfolio_analysis(better_portfolio_id, better_analysis_id),
179
                check_success=is_ready,
180
                timeout=POLLING_TIMEOUT_SECS,
181
                step=10,  # wait 10 seconds between polls
182
            )
183
        except polling.TimeoutException as te:
×
184
            return [f'BETTER analysis timed out after {POLLING_TIMEOUT_SECS} seconds: {te}']
×
185
        except Exception as e:
×
186
            return [
×
187
                f'Unexpected error checking status of BETTER portfolio analysis:'
188
                f' better_portfolio_id: "{better_portfolio_id}"; better_analysis_id: "{better_analysis_id}"'
189
                f': {e}'
190
            ]
191

192
        return []
×
193

194
    def get_portfolio_analysis_standalone_html(self, better_analysis_id):
1✔
195
        """Get portfolio analysis HTML results.
196

197
        :param better_analysis_id: int, ID of an analysis for a portfolio
198
        :return: tuple(tempfile.TemporaryDirectory, list[str]), temporary directory
199
            containing result files and list of error messages
200
        """
201
        url = f'{self.API_URL}/standalone_html/portfolio_analytics/{better_analysis_id}/'
×
202
        headers = {
×
203
            'accept': 'text/html',
204
            'Authorization': self._token,
205
        }
206
        params = {'unit': 'IP'}
×
207

208
        try:
×
209
            response = requests.request("GET", url, headers=headers, params=params)
×
210
            if response.status_code != 200:
×
211
                return None, [f'Expected 200 response from BETTER but got {response.status_code}: {response.content}']
×
212

213
            standalone_html = response.text.encode('utf8').decode()
×
214
        except Exception as e:
×
215
            return None, [f'Unexpected error creating BETTER portfolio: {e}']
×
216

217
        # save the file from the response
218
        temporary_results_dir = TemporaryDirectory()
×
219
        with NamedTemporaryFile(mode='w', suffix='.html', dir=temporary_results_dir.name, delete=False) as file:
×
220
            file.write(standalone_html)
×
221

222
        return temporary_results_dir, []
×
223

224
    def create_building(self, bsync_xml, better_portfolio_id=None):
1✔
225
        """Creates BETTER building from bsync_xml
226

227
        :param bsync_xml: str, path to BSync xml file for property
228
        :param better_portfolio_id: int | str, optional, if provided it will add the
229
            building to the portfolio
230
        :returns: tuple(int, list[str]), BETTER Building ID followed by list of errors
231
        """
232
        url = ""
×
233
        if better_portfolio_id is None:
×
234
            url = f"{self.API_URL}/buildings/"
×
235
        else:
236
            url = f"{self.API_URL}/portfolios/{better_portfolio_id}/buildings/"
×
237

238
        with open(bsync_xml, 'r') as file:
×
239
            bsync_content = file.read()
×
240

241
        headers = {
×
242
            'Authorization': self._token,
243
            'Content-Type': 'buildingsync/xml',
244
        }
245
        try:
×
246
            response = requests.request("POST", url, headers=headers, data=bsync_content)
×
247
            if response.status_code == 201:
×
248
                data = response.json()
×
249
                building_id = data['id']
×
250
            else:
251
                return None, [f'Received non 2xx status from BETTER: {response.status_code}: {response.content}']
×
252
        except Exception as e:
×
253
            return None, [f'BETTER service could not create building with the following message: {e}']
×
254

255
        return building_id, []
×
256

257
    def _create_building_analysis(self, building_id, config):
1✔
258
        """Makes request to better analysis endpoint using the provided configuration
259

260
        :param building_id: int
261
        :param config: request body with building_id, savings_target, benchmark_data, min_model_r_squared
262
        :returns: requests.Response
263
        """
264

265
        url = f"{self.API_URL}/buildings/{building_id}/analytics/"
×
266

267
        headers = {
×
268
            'accept': 'application/json',
269
            'Content-Type': 'application/json',
270
            'Authorization': self._token,
271
        }
272

273
        try:
×
274
            response = requests.request("POST", url, headers=headers, data=json.dumps(config))
×
275
        except ConnectionError:
×
276
            message = 'BETTER service could not create analytics for this building'
×
277
            raise AnalysisPipelineException(message)
×
278

279
        return response
×
280

281
    def get_building_analysis_standalone_html(self, analysis_id):
1✔
282
        """Makes request to better html report endpoint using the provided analysis_id
283

284
        :params: analysis id
285
        :returns: tuple(tempfile.TemporaryDirectory, list[str]), temporary directory containing result files and list of error messages
286
        """
287
        url = f"{self.API_URL}/standalone_html/building_analytics/{analysis_id}/"
×
288
        headers = {
×
289
            'accept': '*/*',
290
            'Authorization': self._token,
291
        }
292
        params = {'unit': 'IP'}
×
293

294
        try:
×
295
            response = requests.request("GET", url, headers=headers, params=params)
×
296
            standalone_html = response.text.encode('utf8').decode()
×
297

298
        except ConnectionError:
×
299
            message = 'BETTER service could not find the analysis'
×
300
            raise AnalysisPipelineException(message)
×
301

302
        # save the file from the response
303
        temporary_results_dir = TemporaryDirectory()
×
304
        with NamedTemporaryFile(mode='w', suffix='.html', dir=temporary_results_dir.name, delete=False) as file:
×
305
            file.write(standalone_html)
×
306

307
        return temporary_results_dir, []
×
308

309
    def get_building_analysis(self, better_building_id, better_analysis_id):
1✔
310
        """Get a building analysis
311

312
        :params: better_building_id
313
        :params: better_analysis_id
314
        :returns: tuple(dict, list[str]), analysis response json and error messages
315
        """
316
        url = f'{self.API_URL}/buildings/{better_building_id}/analytics/{better_analysis_id}/?format=json'
×
317

318
        headers = {
×
319
            'accept': '*/*',
320
            'Authorization': self._token,
321
        }
322
        try:
×
323
            response = requests.request("GET", url, headers=headers)
×
324
            if response.status_code != 200:
×
325
                return None, [f'BETTER analysis could not be fetched: {response.text}']
×
326
            response_json = response.json()
×
327
        except ConnectionError as e:
×
328
            message = f'Failed to connect to BETTER service: {e}'
×
329
            raise AnalysisPipelineException(message)
×
330

331
        return response_json, []
×
332

333
    def create_and_run_building_analysis(self, building_id, config):
1✔
334
        """Runs the better analysis by making a request to a better server with the
335
        provided configuration. Returns the analysis id for standalone html
336

337
        :param building_id: BETTER building id analysis configuration
338
        :param config: dict
339
        :returns: better_analysis_pk
340
        """
341
        try:
×
342
            response = self._create_building_analysis(building_id, config)
×
343
        except Exception as e:
×
344
            return None, [f'Failed to create analysis for building: {e}']
×
345

346
        if response.status_code != 201:
×
347
            return None, ['BETTER analysis could not be completed and got the following response: {message}'.format(
×
348
                message=response.text)]
349

350
        # Gotta make sure the analysis is done
351
        url = f"{self.API_URL}/buildings/{building_id}/analytics/"
×
352

353
        headers = {
×
354
            'accept': 'application/json',
355
            'Authorization': self._token,
356
        }
357
        try:
×
358
            response = polling.poll(
×
359
                lambda: requests.request("GET", url, headers=headers),
360
                check_success=lambda response: response.json()[0]['generation_result'] == 'COMPLETE',
361
                timeout=60,
362
                step=1)
363
        except TimeoutError:
×
364
            return None, ['BETTER analysis timed out']
×
365

366
        data = response.json()
×
367
        better_analysis_id = data[0]['id']
×
368
        return better_analysis_id, []
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc