# coding: utf-8
"""Class to handle recipe inputs and outputs."""
from __future__ import division
import os
import json
import re
import importlib
import shutil
import subprocess
from ladybug.futil import preparedir, nukedir, copy_file_tree
from honeybee.config import folders
from honeybee_radiance.config import folders as rad_folders
from honeybee.model import Model
from .settings import RecipeSettings
from .version import check_radiance_date, check_openstudio_version, \
check_energyplus_version
[docs]class Recipe(object):
"""Recipe class to be used as a base for all Ladybug Tools recipes.
Note that this class is only intended for recipes that have a single output
called "results", which is typically a folder containing all of the
result files of the recipe.
Args:
recipe_name: Text for the name of the recipe folder within this python
package (eg. daylight_factor). This can also be the full path to a
recipe folder (the folder containing the package.json and run.py file).
If the input does not correspond to an installed package or a valid
recipe folder, an exception will be raised.
Properties:
* name
* tag
* path
* default_project_folder
* simulation_id
* inputs
* outputs
"""
MODEL_EXTENSIONS = ('.hbjson', '.hbpkl', '.dfjson', '.dfpkl', '.json', '.pkl')
def __init__(self, recipe_name):
# check to be sure that the requested recipe is installed
install_folder = os.path.dirname(__file__)
recipe_folder = os.path.join(install_folder, recipe_name.replace('-', '_'))
if os.path.isdir(recipe_folder): # it's a recipe in this package
self._name = recipe_name.replace('-', '_')
self._path = recipe_folder
elif os.path.isdir(recipe_name): # it's an externally-installed recipe
self._name = os.path.basename(recipe_name)
self._path = recipe_name
else:
raise ValueError('Recipe "{}" is not installed.'.format(recipe_name))
# load the package.json file to extract the recipe attributes
package_json = os.path.join(self._path, 'package.json')
assert os.path.isfile(package_json), \
'Recipe "{}" lacks a package.json.'.format(self._name)
with open(package_json) as json_file:
package_data = json.load(json_file)
# set the recipe attributes
self._tag = package_data['metadata']['tag']
self._default_project_folder = None
self._simulation_id = None
self._inputs = [RecipeInput(inp) for inp in package_data['inputs']]
self._outputs = [RecipeOutput(outp) for outp in package_data['outputs']]
@property
def name(self):
"""Get text for recipe name."""
return self._name
@property
def tag(self):
"""Get text for recipe tag (aka. its version number)."""
return self._tag
@property
def path(self):
"""Get the path to the recipe's folder.
This folder contains a package.json with metadata about the recipe as well
as a run.py, which is used to execute the recipe.
"""
return self._path
@property
def default_project_folder(self):
"""Get or set the directory in which the recipe's results will be written.
If unset, this will be a folder called unnamed_project within the user's
default simulation folder
"""
if self._default_project_folder is not None:
return self._default_project_folder
else:
def_sim = folders.default_simulation_folder
for inp in self._inputs:
if inp.name == 'model':
if isinstance(inp.value, Model):
clean_name = \
re.sub(r'[^.A-Za-z0-9_-]', '_', inp.value.display_name)
return os.path.join(def_sim, clean_name)
elif isinstance(inp.value, str) and os.path.isfile(str(inp.value)):
model = os.path.basename(inp.value)
for ext in self.MODEL_EXTENSIONS:
model = model.replace(ext, '')
return os.path.join(def_sim, model)
return os.path.join(def_sim, 'unnamed_project')
@default_project_folder.setter
def default_project_folder(self, path):
self._default_project_folder = path
@property
def simulation_id(self):
"""Get or set text for the simulation ID to use within the project folder.
If unset, this will be the same as the name of the recipe.
"""
return self._simulation_id if self._simulation_id is not None else self._name
@simulation_id.setter
def simulation_id(self, value):
self._simulation_id = value
@property
def inputs(self):
"""Get a tuple of RecipeInput objects for the recipe's inputs."""
return tuple(self._inputs)
@property
def outputs(self):
"""Get a tuple of RecipeOutput objects for the recipe's outputs."""
return tuple(self._outputs)
@property
def input_names(self):
"""Get a tuple of text for the recipe's input names."""
return tuple(inp.name for inp in self._inputs)
@property
def output_names(self):
"""Get a tuple of text for the recipe's output names."""
return tuple(otp.name for otp in self._outputs)
[docs] def run(self, settings=None, radiance_check=False, openstudio_check=False,
energyplus_check=False, queenbee_path=None, silent=False,
debug_folder=None):
"""Run the recipe using the queenbee local run command.
Args:
settings: An optional RecipeSettings object or RecipeSettings string
to dictate the settings of the recipe run (eg. the number of
workers or the project folder). If None, default settings will
be assumed. (Default: None).
radiance_check: Boolean to note whether the installed version of
Radiance should be checked before executing the recipe. If there
is no compatible version installed, an exception will be raised
with a clear error message. (Default: False).
openstudio_check: Boolean to note whether the installed version of
OpenStudio should be checked before executing the recipe. If there
is no compatible version installed, an exception will be raised
with a clear error message. (Default: False).
energyplus_check: Boolean to note whether the installed version of
EnergyPlus should be checked before executing the recipe. If there
is no compatible version installed, an exception will be raised
with a clear error message. (Default: False).
queenbee_path: Optional path to the queenbee executable. If None, the
queenbee within the ladybug_tools Python folder will be used.
Setting this to just 'queenbee' will use the system Python.
silent: Boolean to note whether the recipe should be run silently on
Windows (True) or with a command window (False). (Default: False).
debug_folder: An optional path to a debug folder. If debug folder is
provided all the steps of the simulation will be executed inside
the debug folder which can be used for further inspection. Note
that this argument here will override the debug folder specified
in the settings. (Default: None).
Returns:
Path to the project folder containing the recipe results.
"""
# perform any simulation engine checks
if radiance_check:
check_radiance_date()
if openstudio_check:
check_openstudio_version()
if energyplus_check:
check_energyplus_version()
# parse the settings or use default ones
if settings is not None:
settings = RecipeSettings.from_string(settings) \
if isinstance(settings, str) else settings
else:
settings = RecipeSettings()
# get the folder out of which the recipe will be executed
folder = self.default_project_folder if settings.folder is None \
else settings.folder
if not os.path.isdir(folder):
preparedir(folder) # create the directory if it's not there
# delete any existing result files unless reload_old is True
if not settings.reload_old and self.simulation_id is not None:
wf_folder = os.path.join(folder, self.simulation_id)
if os.path.isdir(wf_folder):
nukedir(wf_folder, rmdir=True)
# write the inputs JSON for the recipe and set up the environment variables
inputs_json = self.write_inputs_json(folder, cpu_count=settings.workers)
genv = {}
genv['PATH'] = rad_folders.radbin_path
genv['RAYPATH'] = rad_folders.radlib_path
env_args = ['--env {}="{}"'.format(k, v) for k, v in genv.items()]
# create command
qb_path = os.path.join(folders.python_scripts_path, 'queenbee') \
if queenbee_path is None else queenbee_path
command = '"{qb_path}" local run "{recipe_folder}" ' \
'"{project_folder}" -i "{user_inputs}" --workers {workers} ' \
'{environment} --name {simulation_name}'.format(
qb_path=qb_path, recipe_folder=self.path, project_folder=folder,
user_inputs=inputs_json, workers=settings.workers,
environment=' '.join(env_args),
simulation_name=self.simulation_id
)
if debug_folder is not None:
command += ' --debug "{}"'.format(debug_folder)
elif settings.debug_folder is not None:
command += ' --debug "{}"'.format(settings.debug_folder)
# execute command
shell = False if os.name == 'nt' and not silent else True
custom_env = os.environ.copy()
custom_env['PYTHONHOME'] = ''
if settings.report_out:
process = subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=shell, env=custom_env
)
result = process.communicate()
print(result[0])
print(result[1])
else:
process = subprocess.Popen(command, shell=shell, env=custom_env)
result = process.communicate() # freeze the canvas while running
return folder
[docs] def output_value_by_name(self, output_name, project_folder=None):
"""Set the value of an input given the input name.
Args:
output_name: Text for the name of the output to be obtained.
project_folder: The full path to the project folder containing
completed recipe results. If None, the default_project_folder on
this recipe will be assumed. (Default: None).
"""
proj = self.default_project_folder if project_folder is None else project_folder
sim_path = os.path.join(proj, self.simulation_id)
for outp in self._outputs:
if outp.name == output_name:
return outp.value(sim_path)
else:
raise ValueError(
'Output "{}" was not found for recipe "{}".'.format(
output_name, self.name))
[docs] def luigi_execution_summary(self, project_folder=None):
"""Get a string of the luigi execution summary after the recipe has run.
Args:
project_folder: The full path to the project folder containing
completed recipe logs. If None, the default_project_folder on
this recipe will be assumed. (Default: None).
"""
# determine the log file path from the project folder
proj = self.default_project_folder if project_folder is None else project_folder
sim_path = os.path.join(proj, self.simulation_id)
log_file = os.path.join(sim_path, '__logs__', 'logs.log')
# open the file and load the summary
read_lines, summary_lines = False, []
with open(log_file) as lf:
for line in lf:
if line.strip() == '===== Luigi Execution Summary =====':
read_lines = not read_lines
continue
elif read_lines:
summary_lines.append(line)
return ''.join(summary_lines)
[docs] def error_summary(self, project_folder=None):
"""Get a string of the error summary after the recipe has run.
Args:
project_folder: The full path to the project folder containing
completed recipe logs. If None, the default_project_folder on
this recipe will be assumed. (Default: None).
"""
# determine the log file path from the project folder
proj = self.default_project_folder if project_folder is None else project_folder
sim_path = os.path.join(proj, self.simulation_id)
log_file = os.path.join(sim_path, '__logs__', 'err.log')
if not os.path.isfile(log_file):
return ''
# open the file and load the summary
error_lines = []
with open(log_file) as lf:
for line in lf:
if line == '\n':
continue
error_lines.append(line)
return ''.join(error_lines)
[docs] def failure_message(self, project_folder=None):
"""Get a string of a recipe failure message that gives a summary of failed tasks.
Args:
project_folder: The full path to the project folder containing
completed recipe logs. If None, the default_project_folder on
this recipe will be assumed. (Default: None).
"""
st_msg = '\nThe recipe failed to run with the following error(s):\n\n'
return ''.join([
st_msg, self.error_summary(project_folder),
'\nExecution Summary', self.luigi_execution_summary(project_folder)
])
[docs] def ToString(self):
return self.__repr__()
def __repr__(self):
"""Represent recipe."""
return '{}:\n {}'.format(
self.name, '\n '.join([str(inp) for inp in self.inputs]))
class _RecipeParameter(object):
"""Base class for managing recipe inputs and outputs.
Args:
spec_dict: Dictionary representation of an input or output, taken from the
package.json and following the RecipeInterface schema.
Properties:
* name
* description
* handlers
"""
def __init__(self, spec_dict):
# set the properties based on the output specification
self._name = spec_dict['name']
self._description = spec_dict['description']
# load any of the handlers if they are specified
handlers = []
if 'alias' in spec_dict:
for alias in spec_dict['alias']:
if 'grasshopper' in alias['platform'] and 'handler' in alias:
for hand_dict in alias['handler']:
if hand_dict['language'] == 'python':
module = importlib.import_module(hand_dict['module'])
hand_func = getattr(module, hand_dict['function'])
handlers.append(hand_func)
self._handlers = None if len(handlers) == 0 else tuple(handlers)
@property
def name(self):
"""Get text for the name."""
return self._name
@property
def description(self):
"""Get text for the description."""
return self._description
@property
def handlers(self):
"""Get an array of handler functions for this object.
This will be None if the object has no alias or Grasshopper handler.
"""
return self._handlers
def __repr__(self):
return 'RecipeParameter: {}'.format(self.name)
[docs]class RecipeOutput(_RecipeParameter):
"""Object to represent and manage recipe outputs.
Args:
output_dict: Dictionary representation of an output, taken from the
package.json and following the RecipeInterface schema.
Properties:
* name
* description
* handlers
"""
def __init__(self, output_dict):
_RecipeParameter.__init__(self, output_dict)
# process properties related to the output type
self._type = output_dict['from']['type']
try:
self._path = output_dict['from']['path']
except Exception: # not a file or a folder; type of output not yet supported
self._path = None
[docs] def value(self, simulation_folder):
"""Get the value of this output given the path to a simulation folder.
Args:
simulation_folder: The path to a simulation folder that has finished
running. This is the path of a project folder joined with
the simulation ID.
"""
assert self._type in ('FileReference', 'FolderReference'), \
'Parsing output type "{}" is not yet supported.'.format(self._type)
result = os.path.join(simulation_folder, self._path)
if self._handlers is not None:
for handler in self._handlers:
result = handler(result)
return result
def __repr__(self):
"""Represent recipe output."""
return 'Output: {}'.format(self.name)