import enum
import os
from requests.exceptions import HTTPError
# Constants used in the LiveDesign API
KEY_ID = 'id'
READ_ONLY = 'READ_ONLY'
FILE = 'FILE'
DEFAULT = 'DEFAULT'
ATTACHMENT = 'ATTACHMENT'
# LiveDesign model command types
[docs]class CommandType(enum.Enum):
normal = 'NORMAL'
realtime = 'REALTIME'
click_to_run = 'CLICK_TO_RUN'
clustering = 'CLUSTERING'
def __str__(self):
return self.value
[docs]def upload_ld_attachment(file_path,
project_id,
ld_client,
remote_file_name=None):
"""
Upload a file to LiveDesign as an attachment and return the corresponding
attachment ID.
:param file_path: the path to the file to be uploaded
:type file_path: str
:param project_id: the ID of the LiveDesign project that the file should be
uploaded to
:type project_id: int
:param ld_client: an instance of the LD client to use for the upload
:type ld_client: ldclient.client.LDClient
:param remote_file_name: the name of the attachment once uploaded;
`None`, use the (base) name of the file path
:type remote_file_name: str or NoneType
:return: the attachment ID
:rtype: str
"""
if remote_file_name is None:
remote_file_name = os.path.basename(file_path)
attach_map = ld_client.get_or_create_attachment(
file_path,
file_type=ATTACHMENT,
project_ids=[project_id],
remote_file_name=remote_file_name)
return attach_map[KEY_ID]
[docs]def get_template_vars(template_data, template_class):
"""
Provided a list containing LiveDesign attachment IDs and names, generate and
return a list of template variables.
:param template_data: a list of 2-`tuple` containing LD attachment IDs and
corresponding names
:type template_data: `list` of 2-`tuple` of (`int`, `str`)
:param template_class: the template class to initialize, from the downloaded
ldclient.models module.
:type template_class: class
:return: a list of template variables
:rtype: `list` of `ldclient.models.ModelTemplateVar`
"""
template_vars = []
for attachment_id, name in template_data:
template_var = template_class(tag=READ_ONLY,
type=FILE,
name=name,
data=attachment_id)
template_vars.append(template_var)
return template_vars
[docs]def get_model_by_name(ld_client, model_name, include_archived=True):
"""
Retrieves a single model. Used in KNIME.
:param ld_client: an instance of the LD client to get models
:type ld_client: `ldclient.client.LDClient`
:param model_name: Name of the model to retrieve
:type model_name: str
:param include_archived: Whether to include archived models or not. If
False, ignore archived models
:type include_archived: bool
:return: Matched model object
:rtype: `models.Model`
"""
def is_model_valid(model):
'''
Checks if the given model's name matches model_name
If the name matches, but if include_archived is False and if the matched
model is archived, then return False
'''
if model.name == model_name:
return True if include_archived else not model.archived
return False
matching_models = list(filter(is_model_valid, ld_client.models()))
if len(matching_models) == 0:
raise RuntimeError("No models found with name: {0}".format(model_name))
if len(matching_models) > 1:
raise RuntimeError(
"Multiple models found with name: {0}".format(model_name))
return matching_models[0]
[docs]def upload_ld_model(name,
user_name,
folder,
project_id,
template_vars,
description,
protocol_id,
ld_client,
model_class,
model_recursive_class,
command_type=CommandType.normal,
predictions=None,
model_id=None,
overwrite=False):
"""
Create a LiveDesign model and upload it to a LiveDesign server.
:param name: the name of the model
:type name: `str`
:param user_name: the name of the user uploading this model
:type user_name: `str`
:param folder: the directory path in which the model should be created on
the LiveDesign server
:type folder: `str`
:param project_id: the ID of the LiveDesign project that the file should be
uploaded to
:type project_id: `int`
:param template_vars: a list of template variables containing attachment
data
:rtype: `list` of `ldclient.models.ModelTemplateVar`
:param description: a description of the model
:type description: `str`
:param protocol_id: the ID of the parent protocol for this model
:type protocol_id: `int`
:param ld_client: an instance of the LD client to use for the upload
:type ld_client: `ldclient.client.LDClient`
:param model_class: the model class, from the downloaded ldclient.models
module
:type model_class: class
:param model_recursive_class: the "model recursive class" that is necessary
to build the model instance
:type: model_recursive_class: :class: `ModelRecursive`
:param command_type: Command type of the model to be uploaded
:type command_type: CommandType
:param predictions: the predictions to be added to the model
:type predictions: `list` of `ldclient.models.ModelReturn`
:param model_id: Id of the model to be updated. If model_id is specified,
'overwrite' option will be ignored.
:type model_id: int
:param overwrite: If true and if exactly one model with 'name' already
exists, then instead of creating a new one, update the model.
If no model or more than one model exist with same name, then
create a new model. If 'model_id' is specified, this option will be
ignored and always the given model_id is updated.
:type overwrite: bool
:return: a tuple containing the model instance uploaded to the LiveDesign
server (if available) and a message about the status of the upload
:rtype: tuple[ldclient.models.Model or None, str]
"""
default_command_type = model_recursive_class(DEFAULT, str(command_type))
command_queue = model_recursive_class(DEFAULT, 'SYNC')
if not predictions:
predictions = []
model = model_class(name=name,
archived=False,
published=True,
command_type=default_command_type,
command_queue=command_queue,
user=user_name,
folder=folder,
returns=predictions,
project_ids=[project_id],
template_vars=template_vars,
description=description,
parent=protocol_id)
try:
if model_id:
model.id = model_id
model = ld_client.update_model(model_id, model)
elif overwrite:
model = ld_client.create_or_update_model(model)
else:
model = ld_client.create_model(model)
return model, ''
except HTTPError as exc:
code = exc.response.status_code
msg = f'Export resulted in HTTP error {code}.'
if 'Permission denied' in str(exc):
msg += (' Please make sure your account has the necessary'
' permissions to perform this operation.')
return None, msg