Skip to content

filters

Attributes

Classes

TableFiltersModule

Bases: FilterModule

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/modules/filters.py 
15
16
17
18
19
20
21
22
23
24
25
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
class TableFiltersModule(FilterModule):

    _module_type_name = "network_data.filters"

    @classmethod
    def retrieve_supported_type(cls) -> Union[Dict[str, Any], str]:

        return "network_data"

    def create_filter_inputs(self, filter_name: str) -> Union[None, ValueMapSchema]:

        if filter_name == "component":
            return {
                "component_id": {
                    "type": "string",
                    "doc": "The id of the component to extract.",
                    "default": "0",
                },
                "component_column": {
                    "type": "string",
                    "doc": "The name of the colum that contains the component id.",
                    "default": COMPONENT_ID_COLUMN_NAME,
                },
            }

        return None

    def filter__component(self, value: Value, filter_inputs: Mapping[str, Any]):
        """Retrieve a single sub-component from a network data object."""

        component_id = filter_inputs["component_id"]
        component_column = filter_inputs["component_column"]

        network_data: NetworkData = value.data

        if component_column not in network_data.nodes.column_names:
            msg = f"Component column `{component_column}` not valid for this network_data instance.\n\nAvailable column names:\n\n"

            for attr in network_data.nodes.column_names:
                msg += f"  - {attr}\n"

            if component_column == COMPONENT_ID_COLUMN_NAME:
                msg = f"{msg}\n\nTry to run the `network_data.extract_components` module on your network_data before using this module."

            raise KiaraProcessingException(msg)

        network_data.nodes.arrow_table.column(component_column).type
        # filter_item = pa.scalar(component_id, type=pa.int32())

        query = f"select {NODE_ID_COLUMN_NAME} from nodes where {component_column} = {component_id}"
        node_result = network_data.query_nodes(query)

        network_data = NetworkData.from_filtered_nodes(
            network_data=network_data,
            nodes_list=node_result.column(NODE_ID_COLUMN_NAME).to_pylist(),
        )

        return network_data

Functions

retrieve_supported_type() -> Union[Dict[str, Any], str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/modules/filters.py 
19
20
21
22
@classmethod
def retrieve_supported_type(cls) -> Union[Dict[str, Any], str]:

    return "network_data"
create_filter_inputs(filter_name: str) -> Union[None, ValueMapSchema]
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/modules/filters.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def create_filter_inputs(self, filter_name: str) -> Union[None, ValueMapSchema]:

    if filter_name == "component":
        return {
            "component_id": {
                "type": "string",
                "doc": "The id of the component to extract.",
                "default": "0",
            },
            "component_column": {
                "type": "string",
                "doc": "The name of the colum that contains the component id.",
                "default": COMPONENT_ID_COLUMN_NAME,
            },
        }

    return None
filter__component(value: Value, filter_inputs: Mapping[str, Any])

Retrieve a single sub-component from a network data object.

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/network_analysis/modules/filters.py 
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
def filter__component(self, value: Value, filter_inputs: Mapping[str, Any]):
    """Retrieve a single sub-component from a network data object."""

    component_id = filter_inputs["component_id"]
    component_column = filter_inputs["component_column"]

    network_data: NetworkData = value.data

    if component_column not in network_data.nodes.column_names:
        msg = f"Component column `{component_column}` not valid for this network_data instance.\n\nAvailable column names:\n\n"

        for attr in network_data.nodes.column_names:
            msg += f"  - {attr}\n"

        if component_column == COMPONENT_ID_COLUMN_NAME:
            msg = f"{msg}\n\nTry to run the `network_data.extract_components` module on your network_data before using this module."

        raise KiaraProcessingException(msg)

    network_data.nodes.arrow_table.column(component_column).type
    # filter_item = pa.scalar(component_id, type=pa.int32())

    query = f"select {NODE_ID_COLUMN_NAME} from nodes where {component_column} = {component_id}"
    node_result = network_data.query_nodes(query)

    network_data = NetworkData.from_filtered_nodes(
        network_data=network_data,
        nodes_list=node_result.column(NODE_ID_COLUMN_NAME).to_pylist(),
    )

    return network_data