15680097829

Committed 16 Jun 2025 11:55AM UTC coverage: 60.865% (+0.05%) from 60.815%

Build # 15680097829

Build Type

push

github

Committed by

makkus

Commit Message

chore: fix linting & mypy issues

Run Details

63 of 121 branches covered (52.07%)

Branch coverage included in aggregate %.

5 of 18 new or added lines in 3 files covered. (27.78%)

3 existing lines in 2 files now uncovered.

472 of 758 relevant lines covered (62.27%)

3.11 hits per line

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

43.27

/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."""

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

DHARPA-Project / kiara_plugin.network_analysis / 15680097829

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