module_types
export.network_data
Documentation
Export network data items.
Author(s)
Markus Binsteiner markus@frkl.io
Context
Tags network_analysis
Labels package: kiara_plugin.network_analysis
References source_repo:
https://github.com/DHARPA-Project/kia…
documentation:
https://DHARPA-Project.github.io/kiar…
Module config schema
Field Type Descript… Required Default
─────────────────────────────────────────────────────
constants object Value no
constants
for this
module.
defaults object Value no
defaults
for this
module.
source_t… string The type yes
of the
source
data that
is going
to be
exported.
target_p… string The name yes
of the
target
profile.
Used to
distingu…
different
target
formats
for the
same data
type.
Python class
python_class_name ExportNetworkDataModule
python_module_name kiara_plugin.network_analysis…
full_name kiara_plugin.network_analysis…
Processing source code ─────────────────────────────────────────────────────
def process(self, inputs: ValueMap, outputs: Value…
target_profile: str = self.get_config_value("t…
source_type: str = self.get_config_value("sour…
export_metadata = inputs.get_value_data("expor…
source_obj = inputs.get_value_obj(source_type)
source = source_obj.data
func_name = f"export__{source_type}__as__{targ…
if not hasattr(self, func_name):
raise Exception(
f"Can't export '{source_type}' value: …
)
base_path = inputs.get_value_data("base_path")
if base_path is None:
base_path = os.getcwd()
name = inputs.get_value_data("name")
if not name:
name = str(source_obj.value_id)
func = getattr(self, func_name)
# TODO: check signature?
base_path = os.path.abspath(base_path)
os.makedirs(base_path, exist_ok=True)
result = func(value=source, base_path=base_pat…
if isinstance(result, Mapping):
result = DataExportResult(**result)
elif isinstance(result, str):
result = DataExportResult(files=[result])
if not isinstance(result, DataExportResult):
raise KiaraProcessingException(
f"Can't export value: invalid result t…
)
if export_metadata:
metadata_file = Path(os.path.join(base_pat…
value_info = source_obj.create_info()
value_json = value_info.json()
metadata_file.write_text(value_json)
result.files.append(metadata_file.as_posix…
# schema = ValueSchema(type=self.get_target_va…
# value_lineage = ValueLineage.from_module_and…
# module=self, output_name=output_key, inp…
# )
# value: Value = self._kiara.data_registry.reg…
# value_data=result, value_schema=schema, …
# )
outputs.set_value("export_details", result)
─────────────────────────────────────────────────────
create.network_data.from.tables
Documentation
Create a graph object from one or two tables.
Author(s)
Markus Binsteiner markus@frkl.io
Context
Tags network_analysis
Labels package: kiara_plugin.network_analysis
References source_repo:
https://github.com/DHARPA-Project/kia…
documentation:
https://DHARPA-Project.github.io/kiar…
Module config schema
Field Type Descript… Required Default
─────────────────────────────────────────────────────
constants object Value no
constants
for this
module.
defaults object Value no
defaults
for this
module.
Python class
python_class_name CreateGraphFromTablesModule
python_module_name kiara_plugin.network_analysis…
full_name kiara_plugin.network_analysis…
Processing source code ─────────────────────────────────────────────────────
def process(self, inputs: ValueMap, outputs: Value…
pass
edges = inputs.get_value_obj("edges")
edges_table: KiaraTable = edges.data
edges_source_column_name = inputs.get_value_da…
edges_target_column_name = inputs.get_value_da…
edges_columns = edges_table.column_names
if edges_source_column_name not in edges_colum…
raise KiaraProcessingException(
f"Edges table does not contain source …
)
if edges_target_column_name not in edges_colum…
raise KiaraProcessingException(
f"Edges table does not contain target …
)
nodes = inputs.get_value_obj("nodes")
id_column_name = inputs.get_value_data("id_col…
label_column_name = inputs.get_value_data("lab…
nodes_column_map: Dict[str, str] = inputs.get_…
if nodes_column_map is None:
nodes_column_map = {}
edges_column_map: Dict[str, str] = inputs.get_…
if edges_column_map is None:
edges_column_map = {}
if edges_source_column_name in edges_column_ma…
raise KiaraProcessingException(
"The value of the 'source_column_name'…
)
if edges_target_column_name in edges_column_ma…
raise KiaraProcessingException(
"The value of the 'source_column_name'…
)
edges_column_map[edges_source_column_name] = S…
edges_column_map[edges_target_column_name] = T…
edges_data_schema = create_sqlite_schema_data_…
table=edges_table.arrow_table,
index_columns=[SOURCE_COLUMN_NAME, TARGET_…
column_map=edges_column_map,
)
nodes_table: Optional[KiaraTable] = None
if nodes.is_set:
if (
id_column_name in nodes_column_map.key…
and nodes_column_map[id_column_name] !…
):
raise KiaraProcessingException(
"The value of the 'id_column_name'…
)
nodes_column_map[id_column_name] = ID_COLU…
nodes_table = nodes.data
extra_schema = []
if label_column_name is None:
label_column_name = LABEL_COLUMN_NAME
for cn in nodes_table.column_names:
if cn.lower() == LABEL_COLUMN_NAME.low…
label_column_name = cn
break
if LABEL_COLUMN_NAME in nodes_table.column…
if label_column_name != LABEL_COLUMN_N…
raise KiaraProcessingException(
f"Can't create database for gr…
)
if label_column_name in nodes_table.column…
if label_column_name in nodes_column_m…
raise KiaraProcessingException(
"The value of the 'label_colum…
)
else:
extra_schema.append(" label TEXT…
nodes_column_map[label_column_name] = LABE…
nullable_columns = list(nodes_table.column…
if ID_COLUMN_NAME in nullable_columns:
nullable_columns.remove(ID_COLUMN_NAME)
nodes_data_schema = create_sqlite_schema_d…
table=nodes_table.arrow_table,
index_columns=[ID_COLUMN_NAME],
column_map=nodes_column_map,
nullable_columns=[],
unique_columns=[ID_COLUMN_NAME],
)
else:
nodes_data_schema = None
network_data = NetworkData.create_in_temp_dir(
edges_schema=edges_data_schema,
nodes_schema=nodes_data_schema,
keep_unlocked=True,
)
insert_table_data_into_network_graph(
network_data=network_data,
edges_table=edges_table.arrow_table,
edges_column_map=edges_column_map,
nodes_table=None if nodes_table is None el…
nodes_column_map=nodes_column_map,
chunk_size=DEFAULT_NETWORK_DATA_CHUNK_SIZE,
)
network_data._lock_db()
outputs.set_value("network_data", network_data)
─────────────────────────────────────────────────────