15679907420

Committed 16 Jun 2025 11:45AM UTC coverage: 60.815% (-0.9%) from 61.702%

Build # 15679907420

Build Type

push

github

Committed by

makkus

Commit Message

build: add tabular plugin dependency

Run Details

61 of 119 branches covered (51.26%)

Branch coverage included in aggregate %.

476 of 764 relevant lines covered (62.3%)

3.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

43.81

/src/kiara_plugin/network_analysis/data_types.py

# -*- coding: utf-8 -*-

"""This module contains the value type classes that are used in the ``kiara_plugin.network_analysis`` package."""

# -*- coding: utf-8 -*-

"""This module contains the value type classes that are used in the ``kiara_plugin.network_analysis`` package.
"""
from typing import Any, ClassVar, List, Mapping, Type, Union

from rich.console import Group

from kiara.defaults import DEFAULT_PRETTY_PRINT_CONFIG
from kiara.exceptions import KiaraException
from kiara.models.values.value import Value
from kiara.utils.output import ArrowTabularWrap
from kiara_plugin.network_analysis.defaults import (
    CONNECTIONS_COLUMN_NAME,
    CONNECTIONS_MULTI_COLUMN_NAME,
    COUNT_DIRECTED_COLUMN_NAME,
    COUNT_IDX_DIRECTED_COLUMN_NAME,
    COUNT_IDX_UNDIRECTED_COLUMN_NAME,
    COUNT_UNDIRECTED_COLUMN_NAME,
    EDGE_ID_COLUMN_NAME,
    EDGES_TABLE_NAME,
    IN_DIRECTED_COLUMN_NAME,
    IN_DIRECTED_MULTI_COLUMN_NAME,
    LABEL_COLUMN_NAME,
    NODE_ID_COLUMN_NAME,
    NODES_TABLE_NAME,
    OUT_DIRECTED_COLUMN_NAME,
    OUT_DIRECTED_MULTI_COLUMN_NAME,
    SOURCE_COLUMN_NAME,
    TARGET_COLUMN_NAME,
)
from kiara_plugin.network_analysis.models import NetworkData
from kiara_plugin.tabular.data_types.tables import TablesType
from kiara_plugin.tabular.models.tables import KiaraTables


class NetworkDataType(TablesType):
    """Data that can be assembled into a graph.

    This data type extends the 'tables' type from the [kiara_plugin.tabular](https://github.com/DHARPA-Project/kiara_plugin.tabular) plugin, restricting the allowed tables to one called 'edges',
    and one called 'nodes'.
    """

    _data_type_name: ClassVar[str] = "network_data"
    _cached_doc: ClassVar[Union[str, None]] = None

    @classmethod
    def python_class(cls) -> Type:
        return NetworkData  # type: ignore

    @classmethod
    def type_doc(cls) -> str:

        if cls._cached_doc:
            return cls._cached_doc

        from kiara_plugin.network_analysis.models.metadata import (
            EDGE_COUNT_DUP_DIRECTED_COLUMN_METADATA,
            EDGE_COUNT_DUP_UNDIRECTED_COLUMN_METADATA,
            EDGE_ID_COLUMN_METADATA,
            EDGE_IDX_DUP_DIRECTED_COLUMN_METADATA,
            EDGE_IDX_DUP_UNDIRECTED_COLUMN_METADATA,
            EDGE_SOURCE_COLUMN_METADATA,
            EDGE_TARGET_COLUMN_METADATA,
            NODE_COUND_EDGES_MULTI_COLUMN_METADATA,
            NODE_COUNT_EDGES_COLUMN_METADATA,
            NODE_COUNT_IN_EDGES_COLUMN_METADATA,
            NODE_COUNT_IN_EDGES_MULTI_COLUMN_METADATA,
            NODE_COUNT_OUT_EDGES_COLUMN_METADATA,
            NODE_COUNT_OUT_EDGES_MULTI_COLUMN_METADATA,
            NODE_ID_COLUMN_METADATA,
            NODE_LABEL_COLUMN_METADATA,
        )

        edge_properties = {}
        edge_properties[EDGE_ID_COLUMN_NAME] = EDGE_ID_COLUMN_METADATA.doc.full_doc
        edge_properties[SOURCE_COLUMN_NAME] = EDGE_SOURCE_COLUMN_METADATA.doc.full_doc
        edge_properties[TARGET_COLUMN_NAME] = EDGE_TARGET_COLUMN_METADATA.doc.full_doc
        edge_properties[
            COUNT_DIRECTED_COLUMN_NAME
        ] = EDGE_COUNT_DUP_DIRECTED_COLUMN_METADATA.doc.full_doc
        edge_properties[
            COUNT_IDX_DIRECTED_COLUMN_NAME
        ] = EDGE_IDX_DUP_DIRECTED_COLUMN_METADATA.doc.full_doc
        edge_properties[
            COUNT_UNDIRECTED_COLUMN_NAME
        ] = EDGE_COUNT_DUP_UNDIRECTED_COLUMN_METADATA.doc.full_doc
        edge_properties[
            COUNT_IDX_UNDIRECTED_COLUMN_NAME
        ] = EDGE_IDX_DUP_UNDIRECTED_COLUMN_METADATA.doc.full_doc

        properties_node = {}
        properties_node[NODE_ID_COLUMN_NAME] = NODE_ID_COLUMN_METADATA.doc.full_doc
        properties_node[LABEL_COLUMN_NAME] = NODE_LABEL_COLUMN_METADATA.doc.full_doc
        properties_node[
            CONNECTIONS_COLUMN_NAME
        ] = NODE_COUNT_EDGES_COLUMN_METADATA.doc.full_doc
        properties_node[
            CONNECTIONS_MULTI_COLUMN_NAME
        ] = NODE_COUND_EDGES_MULTI_COLUMN_METADATA.doc.full_doc
        properties_node[
            IN_DIRECTED_COLUMN_NAME
        ] = NODE_COUNT_IN_EDGES_COLUMN_METADATA.doc.full_doc
        properties_node[
            IN_DIRECTED_MULTI_COLUMN_NAME
        ] = NODE_COUNT_IN_EDGES_MULTI_COLUMN_METADATA.doc.full_doc
        properties_node[
            OUT_DIRECTED_COLUMN_NAME
        ] = NODE_COUNT_OUT_EDGES_COLUMN_METADATA.doc.full_doc
        properties_node[
            OUT_DIRECTED_MULTI_COLUMN_NAME
        ] = NODE_COUNT_OUT_EDGES_MULTI_COLUMN_METADATA.doc.full_doc

        edge_properties_str = "\n\n".join(
            f"***{key}***:\n\n{value}" for key, value in edge_properties.items()
        )
        node_properties_str = "\n\n".join(
            f"***{key}***:\n\n{value}" for key, value in properties_node.items()
        )

        doc = cls.__doc__
        doc_tables = f"""

## Edges
The 'edges' table contains the following columns:

{edge_properties_str}

## Nodes

The 'nodes' table contains the following columns:

{node_properties_str}

"""

        cls._cached_doc = f"{doc}\n\n{doc_tables}"
        return cls._cached_doc

    def parse_python_obj(self, data: Any) -> NetworkData:

        if isinstance(data, KiaraTables):
            if EDGES_TABLE_NAME not in data.tables.keys():
                raise KiaraException(
                    f"Can't import network data: no '{EDGES_TABLE_NAME}' table found"
                )

            if NODES_TABLE_NAME not in data.tables.keys():
                raise KiaraException(
                    f"Can't import network data: no '{NODES_TABLE_NAME}' table found"
                )

            # return NetworkData(
            #     tables={
            #         EDGES_TABLE_NAME: data.tables[EDGES_TABLE_NAME],
            #         NODES_TABLE_NAME: data.tables[NODES_TABLE_NAME],
            #     },
            #
            # )
            return NetworkData.create_network_data(
                edges_table=data.tables[EDGES_TABLE_NAME].arrow_table,
                nodes_table=data.tables[NODES_TABLE_NAME].arrow_table,
                augment_tables=False,
            )

        if not isinstance(data, NetworkData):
            raise KiaraException(
                f"Can't parse object to network data: invalid type '{type(data)}'."
            )

        return data

    def _validate(cls, value: Any) -> None:
        if not isinstance(value, NetworkData):
            raise ValueError(
                f"Invalid type '{type(value)}': must be of 'NetworkData' (or a sub-class)."
            )

        network_data: NetworkData = value

        table_names = network_data.table_names
        if EDGES_TABLE_NAME not in table_names:
            raise Exception(
                f"Invalid 'network_data' value: database does not contain table '{EDGES_TABLE_NAME}'."
            )
        if NODES_TABLE_NAME not in table_names:
            raise Exception(
                f"Invalid 'network_data' value: database does not contain table '{NODES_TABLE_NAME}'."
            )

        edges_columns = network_data.edges.column_names
        if SOURCE_COLUMN_NAME not in edges_columns:
            raise Exception(
                f"Invalid 'network_data' value: 'edges' table does not contain a '{SOURCE_COLUMN_NAME}' column. Available columns: {', '.join(edges_columns)}."
            )
        if TARGET_COLUMN_NAME not in edges_columns:
            raise Exception(
                f"Invalid 'network_data' value: 'edges' table does not contain a '{TARGET_COLUMN_NAME}' column. Available columns: {', '.join(edges_columns)}."
            )

        nodes_columns = network_data.nodes.column_names
        if NODE_ID_COLUMN_NAME not in nodes_columns:
            raise Exception(
                f"Invalid 'network_data' value: 'nodes' table does not contain a '{NODE_ID_COLUMN_NAME}' column. Available columns: {', '.join(nodes_columns)}."
            )
        if LABEL_COLUMN_NAME not in nodes_columns:
            raise Exception(
                f"Invalid 'network_data' value: 'nodes' table does not contain a '{LABEL_COLUMN_NAME}' column. Available columns: {', '.join(nodes_columns)}."
            )

    def pretty_print_as__terminal_renderable(
        self, value: Value, render_config: Mapping[str, Any]
    ) -> Any:

        max_rows = render_config.get(
            "max_no_rows", DEFAULT_PRETTY_PRINT_CONFIG["max_no_rows"]
        )
        max_row_height = render_config.get(
            "max_row_height", DEFAULT_PRETTY_PRINT_CONFIG["max_row_height"]
        )
        max_cell_length = render_config.get(
            "max_cell_length", DEFAULT_PRETTY_PRINT_CONFIG["max_cell_length"]
        )

        half_lines: Union[int, None] = None
        if max_rows:
            half_lines = int(max_rows / 2)

        network_data: NetworkData = value.data

        result: List[Any] = [""]

        nodes_atw = ArrowTabularWrap(network_data.nodes.arrow_table)
        nodes_pretty = nodes_atw.as_terminal_renderable(
            rows_head=half_lines,
            rows_tail=half_lines,
            max_row_height=max_row_height,
            max_cell_length=max_cell_length,
        )
        result.append(f"[b]{NODES_TABLE_NAME}[/b]")
        result.append(nodes_pretty)

        edges_atw = ArrowTabularWrap(network_data.edges.arrow_table)
        edges_pretty = edges_atw.as_terminal_renderable(
            rows_head=half_lines,
            rows_tail=half_lines,
            max_row_height=max_row_height,
            max_cell_length=max_cell_length,
        )
        result.append(f"[b]{EDGES_TABLE_NAME}[/b]")
        result.append(edges_pretty)

        return Group(*result)

1	# -- coding: utf-8 --
2
3	"""This module contains the value type classes that are used in the ``kiara_plugin.network_analysis`` package."""
4
5	# -- coding: utf-8 --
6
7	"""This module contains the value type classes that are used in the ``kiara_plugin.network_analysis`` package.	4✔
8	"""
9	from typing import Any, ClassVar, List, Mapping, Type, Union	5✔
10
11	from rich.console import Group	5✔
12
13	from kiara.defaults import DEFAULT_PRETTY_PRINT_CONFIG	5✔
14	from kiara.exceptions import KiaraException	5✔
15	from kiara.models.values.value import Value	5✔
16	from kiara.utils.output import ArrowTabularWrap	5✔
17	from kiara_plugin.network_analysis.defaults import (	5✔
18	CONNECTIONS_COLUMN_NAME,
19	CONNECTIONS_MULTI_COLUMN_NAME,
20	COUNT_DIRECTED_COLUMN_NAME,
21	COUNT_IDX_DIRECTED_COLUMN_NAME,
22	COUNT_IDX_UNDIRECTED_COLUMN_NAME,
23	COUNT_UNDIRECTED_COLUMN_NAME,
24	EDGE_ID_COLUMN_NAME,
25	EDGES_TABLE_NAME,
26	IN_DIRECTED_COLUMN_NAME,
27	IN_DIRECTED_MULTI_COLUMN_NAME,
28	LABEL_COLUMN_NAME,
29	NODE_ID_COLUMN_NAME,
30	NODES_TABLE_NAME,
31	OUT_DIRECTED_COLUMN_NAME,
32	OUT_DIRECTED_MULTI_COLUMN_NAME,
33	SOURCE_COLUMN_NAME,
34	TARGET_COLUMN_NAME,
35	)
36	from kiara_plugin.network_analysis.models import NetworkData	5✔
37	from kiara_plugin.tabular.data_types.tables import TablesType	5✔
38	from kiara_plugin.tabular.models.tables import KiaraTables	5✔
39
40
41	class NetworkDataType(TablesType):	5✔
42	"""Data that can be assembled into a graph.
43
44	This data type extends the 'tables' type from the [kiara_plugin.tabular](https://github.com/DHARPA-Project/kiara_plugin.tabular) plugin, restricting the allowed tables to one called 'edges',
45	and one called 'nodes'.
46	"""
47
48	_data_type_name: ClassVar[str] = "network_data"	5✔
49	_cached_doc: ClassVar[Union[str, None]] = None	5✔
50
51	@classmethod	5✔
52	def python_class(cls) -> Type:	5✔
53	return NetworkData # type: ignore	×
54
55	@classmethod	5✔
56	def type_doc(cls) -> str:	5✔
57
58	if cls._cached_doc:	×
59	return cls._cached_doc	×
60
61	from kiara_plugin.network_analysis.models.metadata import (	×
62	EDGE_COUNT_DUP_DIRECTED_COLUMN_METADATA,
63	EDGE_COUNT_DUP_UNDIRECTED_COLUMN_METADATA,
64	EDGE_ID_COLUMN_METADATA,
65	EDGE_IDX_DUP_DIRECTED_COLUMN_METADATA,
66	EDGE_IDX_DUP_UNDIRECTED_COLUMN_METADATA,
67	EDGE_SOURCE_COLUMN_METADATA,
68	EDGE_TARGET_COLUMN_METADATA,
69	NODE_COUND_EDGES_MULTI_COLUMN_METADATA,
70	NODE_COUNT_EDGES_COLUMN_METADATA,
71	NODE_COUNT_IN_EDGES_COLUMN_METADATA,
72	NODE_COUNT_IN_EDGES_MULTI_COLUMN_METADATA,
73	NODE_COUNT_OUT_EDGES_COLUMN_METADATA,
74	NODE_COUNT_OUT_EDGES_MULTI_COLUMN_METADATA,
75	NODE_ID_COLUMN_METADATA,
76	NODE_LABEL_COLUMN_METADATA,
77	)
78
79	edge_properties = {}	×
80	edge_properties[EDGE_ID_COLUMN_NAME] = EDGE_ID_COLUMN_METADATA.doc.full_doc	×
81	edge_properties[SOURCE_COLUMN_NAME] = EDGE_SOURCE_COLUMN_METADATA.doc.full_doc	×
82	edge_properties[TARGET_COLUMN_NAME] = EDGE_TARGET_COLUMN_METADATA.doc.full_doc	×
83	edge_properties[	×
84	COUNT_DIRECTED_COLUMN_NAME
85	] = EDGE_COUNT_DUP_DIRECTED_COLUMN_METADATA.doc.full_doc
86	edge_properties[	×
87	COUNT_IDX_DIRECTED_COLUMN_NAME
88	] = EDGE_IDX_DUP_DIRECTED_COLUMN_METADATA.doc.full_doc
89	edge_properties[	×
90	COUNT_UNDIRECTED_COLUMN_NAME
91	] = EDGE_COUNT_DUP_UNDIRECTED_COLUMN_METADATA.doc.full_doc
92	edge_properties[	×
93	COUNT_IDX_UNDIRECTED_COLUMN_NAME
94	] = EDGE_IDX_DUP_UNDIRECTED_COLUMN_METADATA.doc.full_doc
95
96	properties_node = {}	×
97	properties_node[NODE_ID_COLUMN_NAME] = NODE_ID_COLUMN_METADATA.doc.full_doc	×
98	properties_node[LABEL_COLUMN_NAME] = NODE_LABEL_COLUMN_METADATA.doc.full_doc	×
99	properties_node[	×
100	CONNECTIONS_COLUMN_NAME
101	] = NODE_COUNT_EDGES_COLUMN_METADATA.doc.full_doc
102	properties_node[	×
103	CONNECTIONS_MULTI_COLUMN_NAME
104	] = NODE_COUND_EDGES_MULTI_COLUMN_METADATA.doc.full_doc
105	properties_node[	×
106	IN_DIRECTED_COLUMN_NAME
107	] = NODE_COUNT_IN_EDGES_COLUMN_METADATA.doc.full_doc
108	properties_node[	×
109	IN_DIRECTED_MULTI_COLUMN_NAME
110	] = NODE_COUNT_IN_EDGES_MULTI_COLUMN_METADATA.doc.full_doc
111	properties_node[	×
112	OUT_DIRECTED_COLUMN_NAME
113	] = NODE_COUNT_OUT_EDGES_COLUMN_METADATA.doc.full_doc
114	properties_node[	×
115	OUT_DIRECTED_MULTI_COLUMN_NAME
116	] = NODE_COUNT_OUT_EDGES_MULTI_COLUMN_METADATA.doc.full_doc
117
118	edge_properties_str = "\n\n".join(	×
119	f"*{key}*:\n\n{value}" for key, value in edge_properties.items()
120	)
121	node_properties_str = "\n\n".join(	×
122	f"*{key}*:\n\n{value}" for key, value in properties_node.items()
123	)
124
125	doc = cls.__doc__	×
126	doc_tables = f"""	×
127
128	## Edges
129	The 'edges' table contains the following columns:
130
131	{edge_properties_str}
132
133	## Nodes
134
135	The 'nodes' table contains the following columns:
136
137	{node_properties_str}
138
139	"""
140
141	cls._cached_doc = f"{doc}\n\n{doc_tables}"	×
142	return cls._cached_doc	×
143
144	def parse_python_obj(self, data: Any) -> NetworkData:	5✔
145
146	if isinstance(data, KiaraTables):	5✔
147	if EDGES_TABLE_NAME not in data.tables.keys():	5✔
148	raise KiaraException(	×
149	f"Can't import network data: no '{EDGES_TABLE_NAME}' table found"
150	)
151
152	if NODES_TABLE_NAME not in data.tables.keys():	5✔
153	raise KiaraException(	×
154	f"Can't import network data: no '{NODES_TABLE_NAME}' table found"
155	)
156
157	# return NetworkData(
158	# tables={
159	# EDGES_TABLE_NAME: data.tables[EDGES_TABLE_NAME],
160	# NODES_TABLE_NAME: data.tables[NODES_TABLE_NAME],
161	# },
162	#
163	# )
164	return NetworkData.create_network_data(	5✔
165	edges_table=data.tables[EDGES_TABLE_NAME].arrow_table,
166	nodes_table=data.tables[NODES_TABLE_NAME].arrow_table,
167	augment_tables=False,
168	)
169
170	if not isinstance(data, NetworkData):	×
171	raise KiaraException(	×
172	f"Can't parse object to network data: invalid type '{type(data)}'."
173	)
174
175	return data	×
176
177	def _validate(cls, value: Any) -> None:	5✔
178	if not isinstance(value, NetworkData):	5✔
179	raise ValueError(	×
180	f"Invalid type '{type(value)}': must be of 'NetworkData' (or a sub-class)."
181	)
182
183	network_data: NetworkData = value	5✔
184
185	table_names = network_data.table_names	5✔
186	if EDGES_TABLE_NAME not in table_names:	5✔
187	raise Exception(	×
188	f"Invalid 'network_data' value: database does not contain table '{EDGES_TABLE_NAME}'."
189	)
190	if NODES_TABLE_NAME not in table_names:	5✔
191	raise Exception(	×
192	f"Invalid 'network_data' value: database does not contain table '{NODES_TABLE_NAME}'."
193	)
194
195	edges_columns = network_data.edges.column_names	5✔
196	if SOURCE_COLUMN_NAME not in edges_columns:	5✔
197	raise Exception(	×
198	f"Invalid 'network_data' value: 'edges' table does not contain a '{SOURCE_COLUMN_NAME}' column. Available columns: {', '.join(edges_columns)}."
199	)
200	if TARGET_COLUMN_NAME not in edges_columns:	5✔
201	raise Exception(	×
202	f"Invalid 'network_data' value: 'edges' table does not contain a '{TARGET_COLUMN_NAME}' column. Available columns: {', '.join(edges_columns)}."
203	)
204
205	nodes_columns = network_data.nodes.column_names	5✔
206	if NODE_ID_COLUMN_NAME not in nodes_columns:	5✔
207	raise Exception(	×
208	f"Invalid 'network_data' value: 'nodes' table does not contain a '{NODE_ID_COLUMN_NAME}' column. Available columns: {', '.join(nodes_columns)}."
209	)
210	if LABEL_COLUMN_NAME not in nodes_columns:	5✔
211	raise Exception(	×
212	f"Invalid 'network_data' value: 'nodes' table does not contain a '{LABEL_COLUMN_NAME}' column. Available columns: {', '.join(nodes_columns)}."
213	)
214
215	def pretty_print_as__terminal_renderable(	5✔
216	self, value: Value, render_config: Mapping[str, Any]
217	) -> Any:
218
219	max_rows = render_config.get(	×
220	"max_no_rows", DEFAULT_PRETTY_PRINT_CONFIG["max_no_rows"]
221	)
222	max_row_height = render_config.get(	×
223	"max_row_height", DEFAULT_PRETTY_PRINT_CONFIG["max_row_height"]
224	)
225	max_cell_length = render_config.get(	×
226	"max_cell_length", DEFAULT_PRETTY_PRINT_CONFIG["max_cell_length"]
227	)
228
229	half_lines: Union[int, None] = None	×
230	if max_rows:	×
231	half_lines = int(max_rows / 2)	×
232
233	network_data: NetworkData = value.data	×
234
235	result: List[Any] = [""]	×
236
237	nodes_atw = ArrowTabularWrap(network_data.nodes.arrow_table)	×
238	nodes_pretty = nodes_atw.as_terminal_renderable(	×
239	rows_head=half_lines,
240	rows_tail=half_lines,
241	max_row_height=max_row_height,
242	max_cell_length=max_cell_length,
243	)
244	result.append(f"[b]{NODES_TABLE_NAME}[/b]")	×
245	result.append(nodes_pretty)	×
246
247	edges_atw = ArrowTabularWrap(network_data.edges.arrow_table)	×
248	edges_pretty = edges_atw.as_terminal_renderable(	×
249	rows_head=half_lines,
250	rows_tail=half_lines,
251	max_row_height=max_row_height,
252	max_cell_length=max_cell_length,
253	)
254	result.append(f"[b]{EDGES_TABLE_NAME}[/b]")	×
255	result.append(edges_pretty)	×
256
257	return Group(*result)	×

DHARPA-Project / kiara_plugin.network_analysis / 15679907420

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous