Source code for plantpredict.prediction

import requests

from plantpredict.plant_predict_entity import PlantPredictEntity
from plantpredict.utilities import convert_json, snake_to_camel
from plantpredict.error_handlers import handle_refused_connection, handle_error_response
from plantpredict.enumerations import PredictionStatusEnum, EntityTypeEnum


[docs]class Prediction(PlantPredictEntity): """ The :py:mod:`plantpredict.Prediction` entity models a single energy prediction within a :py:mod:`plantpredict.Project`. """
[docs] def create(self, use_closest_ashrae_station=True, error_spa_var=2.0, error_model_acc=2.9, error_int_ann_var=3.0, error_sens_acc=5.0, error_mon_acc=2.0, year_repeater=1, status=PredictionStatusEnum.DRAFT_PRIVATE): """ **POST** */Project/ :py:attr:`project_id` /Prediction* Creates a new :py:mod:`plantpredict.Prediction` entity in the PlantPredict database using the attributes assigned to the local object instance. Automatically assigns the resulting :py:attr:`id` to the local object instance. See the minimum required attributes (below) necessary to successfully create a new :py:mod:`plantpredict.Prediction`. Note that the full scope of attributes is not limited to the minimum required set. **Important Note:** the minimum required attributes necessary to create a :py:mod:`plantpredict.Prediction` is not sufficient to successfully call :py:meth:`plantpredict.Prediction.run`. .. container:: toggle .. container:: header **Required Attributes** .. container:: required_attributes .. csv-table:: Minimum required attributes for successful Prediction creation :delim: ; :header: Field, Type, Description :stub-columns: 1 name; str; Name of prediction project_id; int; ID of project within which to contain the prediction year_repeater; int; Must be between :py:data:`1` and :py:data:`50` - unitless. .. container:: toggle .. container:: header **Example Code** .. container:: example_code First, import the plantpredict library and receive an authentication [EDIT THIS] plantpredict.self.api.access_token in your Python session, as shown in Step 3 of :ref:`authentication_oauth2`. Then instantiate a local Prediction. object. .. code-block:: python module_to_create = plantpredict.Prediction() Populate the Prediction's require attributes by either directly assigning them... .. code-block:: python from plantpredict.enumerations import PredictionStatusEnum prediction_to_create.name = "Test Prediction" prediction_to_create.project_id = 1000 prediction_to_create.status = PredictionStatusEnum.DRAFT_SHARED prediction_to_create.year_repeater = 1 ...OR via dictionary assignment. .. code-block:: python prediction_to_create.__dict__ = { "name": "Test Prediction", "model": "Test Module", "status": PredictionStatusEnum.DRAFT_SHARED, "year_repeater": 1, } Create a new prediction in the PlantPredict database, and observe that the Module now has a unique database identifier. .. code-block:: python prediction_to_create.create() print(prediction_to_create.id) :return: A dictionary containing the prediction id. :rtype: dict """ self.create_url_suffix = "/Project/{}/Prediction".format(self.project_id) self.error_spa_var = error_spa_var self.error_model_acc = error_model_acc self.error_int_ann_var = error_int_ann_var self.error_sens_acc = error_sens_acc self.error_mon_acc = error_mon_acc self.year_repeater = year_repeater self.status = status if use_closest_ashrae_station: self._assign_plant_design_temperature_with_closest_ashrae_station() return super(Prediction, self).create()
def _assign_plant_design_temperature_with_closest_ashrae_station(self): """ Assigns the plant design temperatures by using the closest ASHRAE station (based on the associated project's latitude and longitude). """ project = self.api.project(id=self.project_id) project.get() ashrae = self.api.ashrae(latitude=project.latitude, longitude=project.longitude) ashrae.get_closest_station() # set relevant attributes from ASHRAE to Prediction self.ashrae_station = ashrae.station_name self.cool_996 = ashrae.cool_996 self.max_50_year = ashrae.max_50_year self.min_50_year = ashrae.min_50_year
[docs] def delete(self): """HTTP Request: DELETE /Project/{ProjectId}/Prediction/{Id} Deletes an existing Prediction entity in PlantPredict. The local instance of the Project entity must have attribute self.id identical to the prediction id of the Prediction to be deleted. :return: A dictionary {"is_successful": True}. :rtype: dict """ self.delete_url_suffix = "/Project/{}/Prediction/{}".format(self.project_id, self.id) return super(Prediction, self).delete()
[docs] def get(self, id=None, project_id=None): """HTTP Request: GET /Project/{ProjectId}/Prediction/{Id} Retrieves an existing Prediction entity in PlantPredict and automatically assigns all of its attributes to the local Prediction object instance. The local instance of the Prediction entity must have attribute self.id identical to the prediction id of the Prediction to be retrieved. :return: A dictionary containing all of the retrieved Prediction attributes. :rtype: dict """ self.id = id if id is not None else self.id self.project_id = project_id if project_id is not None else self.project_id self.get_url_suffix = "/Project/{}/Prediction/{}".format(self.project_id, self.id) return super(Prediction, self).get()
[docs] def update(self): """HTTP Request: PUT /Project/{ProjectId}/Prediction Updates an existing Prediction entity in PlantPredict using the full attributes of the local Prediction instance. Calling this method is most commonly preceded by instantiating a local instance of Prediction with a specified prediction id, calling the Prediction.get() method, and changing any attributes locally. :return: A dictionary {"is_successful": True}. :rtype: dict """ self.update_url_suffix = "/Project/{}/Prediction".format(self.project_id) return super(Prediction, self).update()
@handle_refused_connection def _wait_for_prediction(self): is_complete = False while not is_complete: self.get() if self.processing_status == 3: is_complete = True @handle_refused_connection @handle_error_response def run(self, export_options=None): """ POST /Project/{ProjectId}/Prediction/{PredictionId}/Run Runs the Prediction and waits for simulation to complete. The input variable "export_options" should take the :param export_options: Contains options for exporting :return: """ response = requests.post( url=self.api.base_url + "/Project/{}/Prediction/{}/Run".format(self.project_id, self.id), headers={"Authorization": "Bearer " + self.api.access_token}, json=convert_json(export_options, snake_to_camel) if export_options else None ) # TODO why didn't this return an error? it only returned when I stopped the script # observes task queue to wait for prediction run to complete self._wait_for_prediction() return response @handle_refused_connection @handle_error_response def get_results_summary(self): """GET /Project/{ProjectId}/Prediction/{Id}/ResultSummary""" return requests.get( url=self.api.base_url + "/Project/{}/Prediction/{}/ResultSummary".format(self.project_id, self.id), headers={"Authorization": "Bearer " + self.api.access_token} ) @handle_refused_connection @handle_error_response def get_results_details(self): """GET /Project/{ProjectId}/Prediction/{Id}/ResultDetails""" return requests.get( url=self.api.base_url + "/Project/{}/Prediction/{}/ResultDetails".format(self.project_id, self.id), headers={"Authorization": "Bearer " + self.api.access_token} ) @handle_refused_connection @handle_error_response def get_nodal_data(self, params=None): """GET /Project/{ProjectId}/Prediction/{Id}/NodalJson""" return requests.get( url=self.api.base_url + "/Project/{}/Prediction/{}/NodalJson".format(self.project_id, self.id), headers={"Authorization": "Bearer " + self.api.access_token}, params=convert_json(params, snake_to_camel) if params else {} ) @handle_refused_connection @handle_error_response def clone(self, new_prediction_name): """ :param new_prediction_name: :return: """ new_prediction = self.api.prediction() self.get() original_prediction_id = self.id new_prediction.__dict__ = self.__dict__ # initialize necessary fields new_prediction.__dict__.pop('id', None) new_prediction.__dict__.pop('created_date', None) new_prediction.__dict__.pop('last_modified', None) new_prediction.__dict__.pop('last_modified_by', None) new_prediction.__dict__.pop('last_modified_by_id', None) new_prediction.__dict__.pop('project', None) new_prediction.__dict__.pop('powerplant_id', None) new_prediction.__dict__.pop('powerplant', None) new_prediction.name = new_prediction_name new_prediction.create() new_prediction_id = new_prediction.id # clone powerplant and attach to new prediction new_powerplant = self.api.powerplant() powerplant = self.api.powerplant(project_id=self.project_id, prediction_id=original_prediction_id) powerplant.get() new_powerplant.__dict__ = powerplant.__dict__ new_powerplant.prediction_id = new_prediction_id new_powerplant.__dict__.pop('id', None) # initialize necessary fields for block in new_powerplant.blocks: block.pop('id', None) for array in block['arrays']: array.pop('id', None) for inverter in array['inverters']: inverter.pop('id', None) for dc_field in inverter['dc_fields']: dc_field.pop('id', None) new_powerplant.create() self.id = original_prediction_id self.get() return new_prediction_id @handle_refused_connection @handle_error_response def change_status(self, new_status, note=""): """ Change the status (and resulting sharing/privacy settings) of a prediction (ex. from py:attr:`DRAFT_PRIVATE` to py:attr:`DRAFT-SHARED`. :param int new_status: Enumeration representing status to change prediction to. See (or import) :py:class:`plantpredict.enumerations.PredictionStatusEnum`. :param str note: Description of reason for change. :return: """ return requests.post( url=self.api.base_url + "/Project/{}/Prediction/Status".format(self.project_id), headers={"Authorization": "Bearer " + self.api.access_token}, json=[{ "name": self.name, "id": self.id, "type": EntityTypeEnum.PREDICTION, "status": new_status, "note": note }] ) def __init__(self, api, id=None, project_id=None, name=None): if id: self.id = id self.project_id = project_id self.name = name self.status = None self.year_repeater = None self.error_spa_var = None self.error_model_acc = None self.error_int_ann_var = None self.error_sens_acc = None self.error_mon_acc = None super(Prediction, self).__init__(api)