in metaflow/plugins/cards/card_modules/basic.py [0:0]
def render(self):
"""
Returns:
a dictionary of form:
dict(metadata = {},components= [])
"""
task_data_dict = TaskToDict(only_repr=self._only_repr)(
self._task, graph=self._graph
)
# ignore the name as an artifact
if "name" in task_data_dict["data"]:
del task_data_dict["data"]["name"]
_metadata = dict(version=1, template="defaultCardTemplate")
# try to parse out metaflow version from tags, but let it go if unset
# e.g. if a run came from a local, un-versioned metaflow codebase
try:
_metadata["metaflow_version"] = [
t for t in self._task.parent.parent.tags if "metaflow_version" in t
][0].split("metaflow_version:")[1]
except Exception:
pass
final_component_dict = dict(
metadata=_metadata,
components=[],
)
metadata = [
"stderr",
"stdout",
"created_at",
"finished_at",
"pathspec",
]
tags = self._task.parent.parent.tags
user_info = [t for t in tags if t.startswith("user:")]
task_metadata_dict = {
"Task Created On": task_data_dict["created_at"],
"Task Finished On": task_data_dict["finished_at"],
# Remove Microseconds from timedelta
"Tags": ", ".join(tags),
}
if not self.runtime:
task_metadata_dict["Task Duration"] = str(
self._task.finished_at - self._task.created_at
).split(".")[0]
if len(user_info) > 0:
task_metadata_dict["User"] = user_info[0].split("user:")[1]
for m in metadata:
final_component_dict["metadata"][m] = task_data_dict[m]
metadata_table = SectionComponent(
title="Task Metadata",
contents=[
TableComponent(
headers=list(task_metadata_dict.keys()),
data=[list(task_metadata_dict.values())],
vertical=True,
)
],
)
img_components = []
for img_name in task_data_dict["images"]:
img_components.append(
ImageComponent(
src=task_data_dict["images"][img_name], label=img_name
).render()
)
table_comps = []
for tabname in task_data_dict["tables"]:
tab_dict = task_data_dict["tables"][tabname]
tab_title = "Artifact Name: %s" % tabname
sec_tab_comp = [
TableComponent(headers=tab_dict["headers"], data=tab_dict["data"])
]
post_table_md = None
if tab_dict["truncated"]:
tab_title = "Artifact Name: %s (%d columns and %d rows)" % (
tabname,
tab_dict["full_size"][1],
tab_dict["full_size"][0],
)
post_table_md = MarkdownComponent(
"_Truncated - %d rows not shown_"
% ((tab_dict["full_size"][0] - len(tab_dict["data"])))
)
if post_table_md:
sec_tab_comp.append(post_table_md)
table_comps.append(
SectionComponent(
title=tab_title,
contents=sec_tab_comp,
)
)
# ignore the name as a parameter
param_ids = [
p.id for p in self._task.parent.parent["_parameters"].task if p.id != "name"
]
if len(param_ids) > 0:
# Extract parameter from the Parameter Task. That is less brittle.
parameter_data = TaskToDict(
only_repr=self._only_repr, runtime=self.runtime
)(self._task.parent.parent["_parameters"].task, graph=self._graph)
param_component = ArtifactsComponent(
data=[parameter_data["data"][pid] for pid in param_ids]
)
else:
param_component = TitleComponent(text="No Parameters")
parameter_table = SectionComponent(
title="Flow Parameters",
contents=[param_component],
).render()
step_func = getattr(self.flow, self._task.parent.id)
code_table = SectionComponent(
title="Task Code",
contents=[
TableComponent(
data=[[PythonCodeComponent(inspect.getsource(step_func)).render()]]
)
],
).render()
# Don't include parameter ids + "name" in the task artifacts
artifactlist = [
task_data_dict["data"][k]
for k in task_data_dict["data"]
if k not in param_ids
]
if len(artifactlist) > 0:
artifact_component = ArtifactsComponent(data=artifactlist).render()
else:
artifact_component = TitleComponent(text="No Artifacts")
artifact_section = SectionComponent(
title="Artifacts", contents=[artifact_component]
).render()
dag_component = SectionComponent(
title="DAG", contents=[DagComponent(data=task_data_dict["graph"]).render()]
).render()
page_contents = []
if len(self._components) > 0:
page_contents.extend(self._components)
page_contents.extend(
[
metadata_table,
code_table,
parameter_table,
artifact_section,
]
)
if len(table_comps) > 0:
table_section = SectionComponent(
title="Tabular Data", contents=table_comps
).render()
page_contents.append(table_section)
if len(img_components) > 0:
img_section = SectionComponent(
title="Image Data",
columns=len(img_components),
contents=img_components,
).render()
page_contents.append(img_section)
page_contents.append(dag_component)
page_component = PageComponent(
title=self._page_title,
contents=page_contents,
).render()
final_component_dict["components"].append(
TitleComponent(text=task_data_dict["pathspec"]).render()
)
final_component_dict["components"].append(page_component)
# These Properties will provide a way to access these components
# once render is finished
# this will Make this object reusable for run level cards.
self.final_component = final_component_dict
self.page_component = page_component
return final_component_dict