"""
Constructs for grouping tool parameters
"""
import io
import logging
import os
import unicodedata
from typing import (
Any,
Callable,
Dict,
List,
Mapping,
Optional,
TYPE_CHECKING,
)
from galaxy.datatypes import data
from galaxy.exceptions import (
AdminRequiredException,
ConfigDoesNotAllowException,
)
from galaxy.files.uris import stream_to_file
from galaxy.util import (
asbool,
inflector,
relpath,
sanitize_for_filename,
)
from galaxy.util.bunch import Bunch
from galaxy.util.dictifiable import Dictifiable
from galaxy.util.expressions import ExpressionContext
if TYPE_CHECKING:
from galaxy.tools import Tool
from galaxy.tools.parameter.basic import ToolParameter
log = logging.getLogger(__name__)
URI_PREFIXES = [f"{x}://" for x in ["http", "https", "ftp", "file", "gxfiles", "gximport", "gxuserimport", "gxftp"]]
[docs]class Group(Dictifiable):
dict_collection_visible_keys = ["name", "type"]
[docs] def __init__(self):
self.name = None
@property
def visible(self):
return True
[docs] def value_to_basic(self, value, app, use_security=False):
"""
Convert value to a (possibly nested) representation using only basic
types (dict, list, tuple, string_types, int, long, float, bool, None)
"""
return value
[docs] def value_from_basic(self, value, app, ignore_errors=False):
"""
Convert a basic representation as produced by `value_to_basic` back
into the preferred value form.
"""
return value
[docs] def get_initial_value(self, trans, context):
"""
Return the initial state/value for this group
"""
raise TypeError("Not implemented")
[docs] def to_dict(self, trans):
group_dict = super().to_dict()
return group_dict
[docs]class Repeat(Group):
dict_collection_visible_keys = ["name", "type", "title", "help", "default", "min", "max"]
type = "repeat"
[docs] def __init__(self):
Group.__init__(self)
self._title = None
self.inputs = None
self.help = None
self.default = 0
self.min = None
self.max = None
@property
def title(self):
return self._title or self.name
@title.setter
def title(self, value):
self._title = value
@property
def title_plural(self):
return inflector.pluralize(self.title)
[docs] def label(self):
return f"Repeat ({self.title})"
[docs] def value_to_basic(self, value, app, use_security=False):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = []
for d in value:
rval_dict = {}
# Propogate __index__
if "__index__" in d:
rval_dict["__index__"] = d["__index__"]
for input in self.inputs.values():
if input.name in d:
rval_dict[input.name] = input.value_to_basic(d[input.name], app, use_security)
rval.append(rval_dict)
return rval
[docs] def value_from_basic(self, value, app, ignore_errors=False):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = []
try:
for i, d in enumerate(value):
rval_dict = {}
# If the special __index__ key is not set, create it (for backward
# compatibility)
rval_dict["__index__"] = d.get("__index__", i)
# Restore child inputs
for input in self.inputs.values():
if ignore_errors and input.name not in d:
# If we do not have a value, and are ignoring errors, we simply
# do nothing. There will be no value for the parameter in the
# conditional's values dictionary.
pass
else:
rval_dict[input.name] = input.value_from_basic(d[input.name], app, ignore_errors)
rval.append(rval_dict)
except Exception as e:
if not ignore_errors:
raise e
return rval
[docs] def get_initial_value(self, trans, context):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = []
for i in range(self.default):
rval_dict = {"__index__": i}
for input in self.inputs.values():
rval_dict[input.name] = input.get_initial_value(trans, context)
rval.append(rval_dict)
return rval
[docs] def to_dict(self, trans):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
repeat_dict = super().to_dict(trans)
def input_to_dict(input):
return input.to_dict(trans)
repeat_dict["inputs"] = list(map(input_to_dict, self.inputs.values()))
return repeat_dict
[docs]class Section(Group):
dict_collection_visible_keys = ["name", "type", "title", "help", "expanded"]
type = "section"
[docs] def __init__(self):
Group.__init__(self)
self.title = None
self.inputs = None
self.help = None
self.expanded = False
@property
def title_plural(self):
return inflector.pluralize(self.title)
[docs] def label(self):
return f"Section ({self.title})"
[docs] def value_to_basic(self, value, app, use_security=False):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = {}
for input in self.inputs.values():
if input.name in value: # parameter might be absent in unverified workflow
rval[input.name] = input.value_to_basic(value[input.name], app, use_security)
return rval
[docs] def value_from_basic(self, value, app, ignore_errors=False):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = {}
try:
for input in self.inputs.values():
if not ignore_errors or input.name in value:
rval[input.name] = input.value_from_basic(value[input.name], app, ignore_errors)
except Exception as e:
if not ignore_errors:
raise e
return rval
[docs] def get_initial_value(self, trans, context):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval: Dict[str, Any] = {}
child_context = ExpressionContext(rval, context)
for child_input in self.inputs.values():
rval[child_input.name] = child_input.get_initial_value(trans, child_context)
return rval
[docs] def to_dict(self, trans):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
section_dict = super().to_dict(trans)
def input_to_dict(input):
return input.to_dict(trans)
section_dict["inputs"] = list(map(input_to_dict, self.inputs.values()))
return section_dict
[docs]class Dataset(Bunch):
type: str
file_type: str
dbkey: str
datatype: data.Data
warnings: List[str]
metadata: Dict[str, str]
composite_files: Dict[str, Optional[str]]
uuid: Optional[str]
tag_using_filenames: Optional[str]
tags: Optional[str]
name: str
primary_file: str
to_posix_lines: bool
auto_decompress: bool
ext: str
space_to_tab: bool
[docs]class UploadDataset(Group):
type = "upload_dataset"
[docs] def __init__(self):
Group.__init__(self)
self.title = None
self.inputs = None
self.file_type_name = "file_type"
self.default_file_type = "txt"
self.file_type_to_ext = {"auto": self.default_file_type}
self.metadata_ref = "files_metadata"
[docs] def get_composite_dataset_name(self, context):
# FIXME: HACK
# Special case of using 'base_name' metadata for use as Dataset name needs to be done in a General Fashion, as defined within a particular Datatype.
# We get two different types of contexts here, one straight from submitted parameters, the other after being parsed into tool inputs
dataset_name = context.get("files_metadata|base_name", None)
if dataset_name is None:
dataset_name = context.get("files_metadata", {}).get("base_name", None)
if dataset_name is None:
filenames = list()
for composite_file in context.get("files", []):
if not composite_file.get("ftp_files", ""):
filenames.append((composite_file.get("file_data") or {}).get("filename", ""))
else:
filenames.append(composite_file.get("ftp_files", [])[0])
dataset_name = os.path.commonprefix(filenames).rstrip(".") or None
if dataset_name is None:
dataset_name = f"Uploaded Composite Dataset ({self.get_file_type(context)})"
return dataset_name
[docs] def get_file_base_name(self, context):
fd = context.get("files_metadata|base_name", "Galaxy_Composite_file")
return fd
[docs] def get_file_type(self, context, parent_context=None):
file_type = context.get(self.file_type_name, None)
if file_type == "":
if parent_context:
file_type = parent_context.get(self.file_type_name, self.default_file_type)
else:
file_type = self.default_file_type
return file_type
[docs] def get_dbkey(self, context, parent_context=None):
dbkey = context.get("dbkey", None)
if dbkey == "":
if parent_context:
dbkey = parent_context.get("dbkey", dbkey)
return dbkey
[docs] def get_datatype_ext(self, trans, context, parent_context=None):
ext = self.get_file_type(context, parent_context=parent_context)
if ext in self.file_type_to_ext:
ext = self.file_type_to_ext[
ext
] # when using autodetect, we will use composite info from 'text', i.e. only the main file
return ext
[docs] def get_datatype(self, trans, context, parent_context=None):
ext = self.get_datatype_ext(trans, context, parent_context=parent_context)
return trans.app.datatypes_registry.get_datatype_by_extension(ext)
@property
def title_plural(self):
return inflector.pluralize(self.title)
[docs] def group_title(self, context):
return f"{self.title} ({context.get(self.file_type_name, self.default_file_type)})"
[docs] def title_by_index(self, trans, index, context):
d_type = self.get_datatype(trans, context)
for i, (composite_name, composite_file) in enumerate(d_type.writable_files.items()):
if i == index:
rval = composite_name
if composite_file.description:
rval = f"{rval} ({composite_file.description})"
if composite_file.optional:
rval = f"{rval} [optional]"
return rval
if index < self.get_file_count(trans, context):
return "Extra primary file"
return None
[docs] def value_to_basic(self, value, app, use_security=False):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = []
for d in value:
rval_dict = {}
# Propogate __index__
if "__index__" in d:
rval_dict["__index__"] = d["__index__"]
for input in self.inputs.values():
if input.name in d:
rval_dict[input.name] = input.value_to_basic(d[input.name], app, use_security)
rval.append(rval_dict)
return rval
[docs] def value_from_basic(self, value, app, ignore_errors=False):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
rval = []
for i, d in enumerate(value):
try:
rval_dict = {}
# If the special __index__ key is not set, create it (for backward
# compatibility)
rval_dict["__index__"] = d.get("__index__", i)
# Restore child inputs
for input in self.inputs.values():
if ignore_errors and input.name not in d: # this wasn't tested
rval_dict[input.name] = input.get_initial_value(None, d)
else:
rval_dict[input.name] = input.value_from_basic(d[input.name], app, ignore_errors)
rval.append(rval_dict)
except Exception as e:
if not ignore_errors:
raise e
return rval
[docs] def get_file_count(self, trans, context):
file_count = context.get("file_count", "auto")
if file_count == "auto":
d_type = self.get_datatype(trans, context)
return len(d_type.writable_files) if d_type else 1
else:
return int(file_count)
[docs] def get_initial_value(self, trans, context):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
file_count = self.get_file_count(trans, context)
rval = []
for i in range(file_count):
rval_dict = {}
rval_dict["__index__"] = i # create __index__
for input in self.inputs.values():
rval_dict[input.name] = input.get_initial_value(trans, context)
rval.append(rval_dict)
return rval
[docs] def get_uploaded_datasets(self, trans, context, override_name=None, override_info=None):
def get_data_file_filename(data_file, override_name=None, override_info=None, purge=True):
dataset_name = override_name
def get_file_name(file_name):
file_name = file_name.split("\\")[-1]
file_name = file_name.split("/")[-1]
return file_name
try:
# Use the existing file
if not dataset_name and "filename" in data_file:
dataset_name = get_file_name(data_file["filename"])
return Bunch(type="file", path=data_file["local_filename"], name=dataset_name, purge_source=purge)
except Exception:
# The uploaded file should've been persisted by the upload tool action
return Bunch(type=None, path=None, name=None)
def get_url_paste_urls_or_filename(group_incoming, override_name=None, override_info=None):
url_paste_file = group_incoming.get("url_paste", None)
if url_paste_file is not None:
url_paste = open(url_paste_file).read()
def start_of_url(content):
start_of_url_paste = content.lstrip()[0:10].lower()
looks_like_url = False
for url_prefix in URI_PREFIXES:
if start_of_url_paste.startswith(url_prefix):
looks_like_url = True
break
return looks_like_url
if start_of_url(url_paste):
for line in url_paste.replace("\r", "").split("\n"):
line = line.strip()
if line:
if not start_of_url(line):
continue # non-url line, ignore
if "file://" in line:
if not trans.user_is_admin:
raise AdminRequiredException()
elif not trans.app.config.allow_path_paste:
raise ConfigDoesNotAllowException()
upload_path = line[len("file://") :]
dataset_name = os.path.basename(upload_path)
else:
dataset_name = line
if override_name:
dataset_name = override_name
yield Bunch(type="url", path=line, name=dataset_name)
else:
dataset_name = "Pasted Entry" # we need to differentiate between various url pastes here
if override_name:
dataset_name = override_name
yield Bunch(type="file", path=url_paste_file, name=dataset_name)
def get_one_filename(context):
data_file = context["file_data"]
url_paste = context["url_paste"]
ftp_files = context["ftp_files"]
name = context.get("NAME", None)
info = context.get("INFO", None)
uuid = context.get("uuid", None) or None # Turn '' to None
file_type = context.get("file_type", None)
dbkey = self.get_dbkey(context)
warnings = []
to_posix_lines = False
if context.get("to_posix_lines", None) not in ["None", None, False]:
to_posix_lines = True
auto_decompress = False
if context.get("auto_decompress", None) not in ["None", None, False]:
auto_decompress = True
space_to_tab = False
if context.get("space_to_tab", None) not in ["None", None, False]:
space_to_tab = True
file_bunch = get_data_file_filename(data_file, override_name=name, override_info=info)
if file_bunch.path:
if url_paste is not None and url_paste.strip():
warnings.append("All file contents specified in the paste box were ignored.")
if ftp_files:
warnings.append("All FTP uploaded file selections were ignored.")
elif url_paste is not None and url_paste.strip(): # we need to use url_paste
for file_bunch in get_url_paste_urls_or_filename(context, override_name=name, override_info=info):
if file_bunch.path:
break
if file_bunch.path and ftp_files is not None:
warnings.append("All FTP uploaded file selections were ignored.")
elif ftp_files is not None and trans.user is not None: # look for files uploaded via FTP
user_ftp_dir = trans.user_ftp_dir
assert not os.path.islink(user_ftp_dir), "User FTP directory cannot be a symbolic link"
for dirpath, _dirnames, filenames in os.walk(user_ftp_dir):
for filename in filenames:
for ftp_filename in ftp_files:
if ftp_filename == filename:
path = relpath(os.path.join(dirpath, filename), user_ftp_dir)
if not os.path.islink(os.path.join(dirpath, filename)):
ftp_data_file = {
"local_filename": os.path.abspath(os.path.join(user_ftp_dir, path)),
"filename": os.path.basename(path),
}
purge = getattr(trans.app.config, "ftp_upload_purge", True)
file_bunch = get_data_file_filename(
ftp_data_file,
override_name=name,
override_info=info,
purge=purge,
)
if file_bunch.path:
break
if file_bunch.path:
break
if file_bunch.path:
break
file_bunch.to_posix_lines = to_posix_lines
file_bunch.auto_decompress = auto_decompress
file_bunch.space_to_tab = space_to_tab
file_bunch.uuid = uuid
if file_type is not None:
file_bunch.file_type = file_type
if dbkey is not None:
file_bunch.dbkey = dbkey
return file_bunch, warnings
def get_filenames(context):
rval = []
data_file = context["file_data"]
ftp_files = context["ftp_files"]
uuid = context.get("uuid", None) or None # Turn '' to None
name = context.get("NAME", None)
info = context.get("INFO", None)
file_type = context.get("file_type", None)
dbkey = self.get_dbkey(context)
to_posix_lines = False
if context.get("to_posix_lines", None) not in ["None", None, False]:
to_posix_lines = True
auto_decompress = False
if context.get("auto_decompress", None) not in ["None", None, False]:
auto_decompress = True
space_to_tab = False
if context.get("space_to_tab", None) not in ["None", None, False]:
space_to_tab = True
file_bunch = get_data_file_filename(data_file, override_name=name, override_info=info)
file_bunch.uuid = uuid
if file_bunch.path:
file_bunch.to_posix_lines = to_posix_lines
file_bunch.auto_decompress = auto_decompress
file_bunch.space_to_tab = space_to_tab
if file_type is not None:
file_bunch.file_type = file_type
if dbkey is not None:
file_bunch.dbkey = dbkey
rval.append(file_bunch)
for file_bunch in get_url_paste_urls_or_filename(context, override_name=name, override_info=info):
if file_bunch.path:
file_bunch.uuid = uuid
file_bunch.to_posix_lines = to_posix_lines
file_bunch.auto_decompress = auto_decompress
file_bunch.space_to_tab = space_to_tab
if file_type is not None:
file_bunch.file_type = file_type
if dbkey is not None:
file_bunch.dbkey = dbkey
rval.append(file_bunch)
# look for files uploaded via FTP
valid_files = []
if ftp_files is not None:
# Normalize input paths to ensure utf-8 encoding is normal form c.
# This allows for comparison when the filesystem uses a different encoding than the browser.
ftp_files = [unicodedata.normalize("NFC", f) for f in ftp_files if isinstance(f, str)]
if trans.user is None:
log.warning(f"Anonymous user passed values in ftp_files: {ftp_files}")
ftp_files = []
# TODO: warning to the user (could happen if session has become invalid)
else:
user_ftp_dir = trans.user_ftp_dir
assert not os.path.islink(user_ftp_dir), "User FTP directory cannot be a symbolic link"
for dirpath, _dirnames, filenames in os.walk(user_ftp_dir):
for filename in filenames:
path = relpath(os.path.join(dirpath, filename), user_ftp_dir)
if not os.path.islink(os.path.join(dirpath, filename)):
# Normalize filesystem paths
if isinstance(path, str):
valid_files.append(unicodedata.normalize("NFC", path))
else:
valid_files.append(path)
else:
ftp_files = []
for ftp_file in ftp_files:
if ftp_file not in valid_files:
log.warning(f"User passed an invalid file path in ftp_files: {ftp_file}")
continue
# TODO: warning to the user (could happen if file is already imported)
ftp_data_file = {
"local_filename": os.path.abspath(os.path.join(user_ftp_dir, ftp_file)),
"filename": os.path.basename(ftp_file),
}
purge = getattr(trans.app.config, "ftp_upload_purge", True)
file_bunch = get_data_file_filename(ftp_data_file, override_name=name, override_info=info, purge=purge)
if file_bunch.path:
file_bunch.to_posix_lines = to_posix_lines
file_bunch.auto_decompress = auto_decompress
file_bunch.space_to_tab = space_to_tab
if file_type is not None:
file_bunch.file_type = file_type
if dbkey is not None:
file_bunch.dbkey = dbkey
rval.append(file_bunch)
return rval
file_type = self.get_file_type(context)
file_count = self.get_file_count(trans, context)
d_type = self.get_datatype(trans, context)
dbkey = self.get_dbkey(context)
tag_using_filenames = context.get("tag_using_filenames", False)
tags = context.get("tags", False)
force_composite = asbool(context.get("force_composite", "False"))
writable_files = d_type.writable_files
writable_files_offset = 0
groups_incoming = [None for _ in range(file_count)]
for group_incoming in context.get(self.name, []):
i = int(group_incoming["__index__"])
groups_incoming[i] = group_incoming
if d_type.composite_type is not None or force_composite:
# handle uploading of composite datatypes
# Only one Dataset can be created
dataset = Dataset()
dataset.type = "composite"
dataset.file_type = file_type
dataset.dbkey = dbkey
dataset.datatype = d_type
dataset.warnings = []
dataset.metadata = {}
dataset.composite_files = {}
dataset.uuid = None
dataset.tag_using_filenames = None
dataset.tags = None
# load metadata
files_metadata = context.get(self.metadata_ref, {})
metadata_name_substition_default_dict = {
composite_file.substitute_name_with_metadata: d_type.metadata_spec[
composite_file.substitute_name_with_metadata
].default
for composite_file in d_type.composite_files.values()
if composite_file.substitute_name_with_metadata
}
for meta_name, meta_spec in d_type.metadata_spec.items():
if meta_spec.set_in_upload:
if meta_name in files_metadata:
meta_value = files_metadata[meta_name]
if meta_name in metadata_name_substition_default_dict:
meta_value = sanitize_for_filename(
meta_value, default=metadata_name_substition_default_dict[meta_name]
)
dataset.metadata[meta_name] = meta_value
dataset.name = self.get_composite_dataset_name(context)
if dataset.datatype.composite_type == "auto_primary_file":
# replace sniff here with just creating an empty file
temp_name = stream_to_file(
io.StringIO(d_type.generate_primary_file(dataset)), prefix="upload_auto_primary_file"
)
dataset.primary_file = temp_name
dataset.to_posix_lines = True
dataset.auto_decompress = True
dataset.space_to_tab = False
else:
file_bunch, warnings = get_one_filename(groups_incoming[0])
writable_files_offset = 1
dataset.primary_file = file_bunch.path
dataset.to_posix_lines = file_bunch.to_posix_lines
dataset.auto_decompress = file_bunch.auto_decompress
dataset.space_to_tab = file_bunch.space_to_tab
if file_bunch.file_type:
dataset.file_type = file_type
if file_bunch.dbkey:
dataset.dbkey = dbkey
dataset.warnings.extend(warnings)
if dataset.primary_file is None: # remove this before finish, this should create an empty dataset
raise Exception("No primary dataset file was available for composite upload")
if not force_composite:
keys = [value.name for value in writable_files.values()]
else:
keys = [str(index) for index in range(file_count)]
for i, group_incoming in enumerate(groups_incoming[writable_files_offset:]):
key = keys[i + writable_files_offset]
if (
not force_composite
and group_incoming is None
and not writable_files[list(writable_files.keys())[keys.index(key)]].optional
):
dataset.warnings.append(f"A required composite file ({key}) was not specified.")
dataset.composite_files[key] = None
else:
file_bunch, warnings = get_one_filename(group_incoming)
dataset.warnings.extend(warnings)
if file_bunch.path:
if force_composite:
key = group_incoming.get("NAME") or i
dataset.composite_files[key] = file_bunch.__dict__
elif not force_composite:
dataset.composite_files[key] = None
if not writable_files[list(writable_files.keys())[keys.index(key)]].optional:
dataset.warnings.append(f"A required composite file ({key}) was not specified.")
return [dataset]
else:
rval = []
for i, file_contexts in enumerate(context[self.name]):
datasets = get_filenames(file_contexts)
for dataset in datasets:
override_file_type = self.get_file_type(context[self.name][i], parent_context=context)
d_type = self.get_datatype(trans, context[self.name][i], parent_context=context)
dataset.file_type = override_file_type
dataset.datatype = d_type
dataset.ext = self.get_datatype_ext(trans, context[self.name][i], parent_context=context)
dataset.dbkey = self.get_dbkey(context[self.name][i], parent_context=context)
dataset.tag_using_filenames = tag_using_filenames
dataset.tags = tags
rval.append(dataset)
return rval
[docs]class Conditional(Group):
type = "conditional"
value_from: Callable[["Conditional", ExpressionContext, "Conditional", "Tool"], Mapping[str, str]]
[docs] def __init__(self):
Group.__init__(self)
self.test_param: Optional["ToolParameter"] = None
self.cases = []
self.value_ref = None
self.value_ref_in_group = True # When our test_param is not part of the conditional Group, this is False
@property
def label(self):
return f"Conditional ({self.name})"
[docs] def get_current_case(self, value):
if self.test_param is None:
raise Exception("Must set 'test_param' attribute to use.")
# Convert value to user representation
str_value = self.test_param.to_param_dict_string(value)
# Find the matching case
for index, case in enumerate(self.cases):
if str_value == case.value:
return index
raise ValueError("No case matched value:", self.name, str_value)
[docs] def value_to_basic(self, value, app, use_security=False):
if self.test_param is None:
raise Exception("Must set 'test_param' attribute to use.")
rval = dict()
rval[self.test_param.name] = self.test_param.value_to_basic(value[self.test_param.name], app)
current_case = rval["__current_case__"] = self.get_current_case(value[self.test_param.name])
for input in self.cases[current_case].inputs.values():
if input.name in value: # parameter might be absent in unverified workflow
rval[input.name] = input.value_to_basic(value[input.name], app, use_security=use_security)
return rval
[docs] def value_from_basic(self, value, app, ignore_errors=False):
if self.test_param is None:
raise Exception("Must set 'test_param' attribute to use.")
rval = dict()
try:
rval[self.test_param.name] = self.test_param.value_from_basic(
value.get(self.test_param.name), app, ignore_errors
)
current_case = rval["__current_case__"] = self.get_current_case(rval[self.test_param.name])
# Inputs associated with current case
for input in self.cases[current_case].inputs.values():
# If we do not have a value, and are ignoring errors, we simply
# do nothing. There will be no value for the parameter in the
# conditional's values dictionary.
if not ignore_errors or input.name in value:
rval[input.name] = input.value_from_basic(value[input.name], app, ignore_errors)
except Exception as e:
if not ignore_errors:
raise e
return rval
[docs] def get_initial_value(self, trans, context):
if self.test_param is None:
raise Exception("Must set 'test_param' attribute to use.")
# State for a conditional is a plain dictionary.
rval = {}
# Get the default value for the 'test element' and use it
# to determine the current case
test_value = self.test_param.get_initial_value(trans, context)
current_case = self.get_current_case(test_value)
# Store the current case in a special value
rval["__current_case__"] = current_case
# Store the value of the test element
rval[self.test_param.name] = test_value
# Fill in state for selected case
child_context = ExpressionContext(rval, context)
for child_input in self.cases[current_case].inputs.values():
rval[child_input.name] = child_input.get_initial_value(trans, child_context)
return rval
[docs] def to_dict(self, trans):
if self.test_param is None:
raise Exception("Must set 'test_param' attribute to use.")
cond_dict = super().to_dict(trans)
def nested_to_dict(input):
return input.to_dict(trans)
cond_dict["cases"] = list(map(nested_to_dict, self.cases))
cond_dict["test_param"] = nested_to_dict(self.test_param)
return cond_dict
[docs]class ConditionalWhen(Dictifiable):
dict_collection_visible_keys = ["value"]
[docs] def __init__(self):
self.value = None
self.inputs = None
[docs] def to_dict(self, trans):
if self.inputs is None:
raise Exception("Must set 'inputs' attribute to use.")
when_dict = super().to_dict()
def input_to_dict(input):
return input.to_dict(trans)
when_dict["inputs"] = list(map(input_to_dict, self.inputs.values()))
return when_dict