Skip to content

data_types

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

Attributes

Classes

NetworkDataType

Bases: TablesType

Data that can be assembled into a graph.

This data type extends the 'database' type from the kiara_plugin.tabular plugin, restricting the allowed tables to one called 'edges', and one called 'nodes'.

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/data_types.py
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class NetworkDataType(TablesType):
    """Data that can be assembled into a graph.

    This data type extends the 'database' 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 = "network_data"

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

    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,
            )

        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)

Functions

python_class() -> Type classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/data_types.py
35
36
37
@classmethod
def python_class(cls) -> Type:
    return NetworkData
parse_python_obj(data: Any) -> NetworkData
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/data_types.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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,
        )

    return data
pretty_print_as__terminal_renderable(value: Value, render_config: Mapping[str, Any]) -> Any
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/data_types.py
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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)