Source code for snapchat.research.gbml.subgraph_sampling_strategy_pb2

"""
@generated by mypy-protobuf.  Do not edit manually!
isort:skip_file
"""
import builtins
import collections.abc
import google.protobuf.descriptor
import google.protobuf.internal.containers
import google.protobuf.internal.enum_type_wrapper
import google.protobuf.message
import snapchat.research.gbml.graph_schema_pb2
import sys
import typing

if sys.version_info >= (3, 10):
    import typing as typing_extensions
else:
    import typing_extensions


[docs]
DESCRIPTOR: google.protobuf.descriptor.FileDescriptor


class _SamplingDirection:
    ValueType = typing.NewType("ValueType", builtins.int)
    V: typing_extensions.TypeAlias = ValueType

class _SamplingDirectionEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_SamplingDirection.ValueType], builtins.type):  # noqa: F821
    DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
    INCOMING: _SamplingDirection.ValueType  # 0
    """Sample incoming edges to the dst nodes (default)"""
    OUTGOING: _SamplingDirection.ValueType  # 1
    """Sample outgoing edges from the src nodes"""


[docs]
class SamplingDirection(_SamplingDirection, metaclass=_SamplingDirectionEnumTypeWrapper): ...



[docs]
INCOMING: SamplingDirection.ValueType  # 0

"""Sample incoming edges to the dst nodes (default)"""

[docs]
OUTGOING: SamplingDirection.ValueType  # 1

"""Sample outgoing edges from the src nodes"""

[docs]
global___SamplingDirection = SamplingDirection



[docs]
class RandomUniform(google.protobuf.message.Message):
    """Randomly sample nodes from the neighborhood without replacement."""


[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    NUM_NODES_TO_SAMPLE_FIELD_NUMBER: builtins.int


[docs]
    num_nodes_to_sample: builtins.int

    def __init__(
        self,
        *,
        num_nodes_to_sample: builtins.int = ...,
    ) -> None: ...

[docs]
    def ClearField(self, field_name: typing_extensions.Literal["num_nodes_to_sample", b"num_nodes_to_sample"]) -> None: ...




[docs]
global___RandomUniform = RandomUniform



[docs]
class RandomWeighted(google.protobuf.message.Message):
    """Sample based off some edge weight"""


[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    NUM_NODES_TO_SAMPLE_FIELD_NUMBER: builtins.int


[docs]
    EDGE_FEAT_NAME_FIELD_NUMBER: builtins.int


[docs]
    num_nodes_to_sample: builtins.int


[docs]
    edge_feat_name: builtins.str

    def __init__(
        self,
        *,
        num_nodes_to_sample: builtins.int = ...,
        edge_feat_name: builtins.str = ...,
    ) -> None: ...

[docs]
    def ClearField(self, field_name: typing_extensions.Literal["edge_feat_name", b"edge_feat_name", "num_nodes_to_sample", b"num_nodes_to_sample"]) -> None: ...




[docs]
global___RandomWeighted = RandomWeighted



[docs]
class TopK(google.protobuf.message.Message):
    """Order the edges based off of some edge feature and pick top K"""


[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    NUM_NODES_TO_SAMPLE_FIELD_NUMBER: builtins.int


[docs]
    EDGE_FEAT_NAME_FIELD_NUMBER: builtins.int


[docs]
    num_nodes_to_sample: builtins.int


[docs]
    edge_feat_name: builtins.str

    def __init__(
        self,
        *,
        num_nodes_to_sample: builtins.int = ...,
        edge_feat_name: builtins.str = ...,
    ) -> None: ...

[docs]
    def ClearField(self, field_name: typing_extensions.Literal["edge_feat_name", b"edge_feat_name", "num_nodes_to_sample", b"num_nodes_to_sample"]) -> None: ...




[docs]
global___TopK = TopK



[docs]
class UserDefined(google.protobuf.message.Message):

[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    class ParamsEntry(google.protobuf.message.Message):

[docs]
        DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
        KEY_FIELD_NUMBER: builtins.int


[docs]
        VALUE_FIELD_NUMBER: builtins.int


[docs]
        key: builtins.str


[docs]
        value: builtins.str

        def __init__(
            self,
            *,
            key: builtins.str = ...,
            value: builtins.str = ...,
        ) -> None: ...

[docs]
        def ClearField(self, field_name: typing_extensions.Literal["key", b"key", "value", b"value"]) -> None: ...




[docs]
    PATH_TO_UDF_FIELD_NUMBER: builtins.int


[docs]
    PARAMS_FIELD_NUMBER: builtins.int


[docs]
    path_to_udf: builtins.str

    """Path to class that implements UserDefinedFilter interface (TODO: Define this interface)"""
    @property

[docs]
    def params(self) -> google.protobuf.internal.containers.ScalarMap[builtins.str, builtins.str]:
        """Runtime params to pass into the class"""

    def __init__(
        self,
        *,
        path_to_udf: builtins.str = ...,
        params: collections.abc.Mapping[builtins.str, builtins.str] | None = ...,
    ) -> None: ...

[docs]
    def ClearField(self, field_name: typing_extensions.Literal["params", b"params", "path_to_udf", b"path_to_udf"]) -> None: ...




[docs]
global___UserDefined = UserDefined



[docs]
class SamplingOp(google.protobuf.message.Message):
    """TODO: (svij) Define UserDefinedFilter interface
    Rough idea of UserDefinedFilter interface
    class UserDefinedFilter:
      def filter_next_hop_nodes_and_edges(
        potential_next_hop: Tuple[Set(Edge), Set(Node)],
      ) -> Tuple[Set(Edge), Set(Node)] # Given a set of edges and nodes; the function filters the next hop nodes/edges
    """


[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    OP_NAME_FIELD_NUMBER: builtins.int


[docs]
    EDGE_TYPE_FIELD_NUMBER: builtins.int


[docs]
    INPUT_OP_NAMES_FIELD_NUMBER: builtins.int


[docs]
    RANDOM_UNIFORM_FIELD_NUMBER: builtins.int


[docs]
    RANDOM_WEIGHTED_FIELD_NUMBER: builtins.int


[docs]
    TOP_K_FIELD_NUMBER: builtins.int


[docs]
    USER_DEFINED_FIELD_NUMBER: builtins.int


[docs]
    SAMPLING_DIRECTION_FIELD_NUMBER: builtins.int


[docs]
    op_name: builtins.str

    """1-100 field numbers reserved for expanding SamplingOp
    Can be used as a reference in other operations
    """
    @property

[docs]
    def edge_type(self) -> snapchat.research.gbml.graph_schema_pb2.EdgeType: ...

    @property

[docs]
    def input_op_names(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]:
        """the list of upstream sampling operations in the DAG, empty if this is the root"""

    @property

[docs]
    def random_uniform(self) -> global___RandomUniform: ...

    @property

[docs]
    def random_weighted(self) -> global___RandomWeighted: ...

    @property

[docs]
    def top_k(self) -> global___TopK: ...

    @property

[docs]
    def user_defined(self) -> global___UserDefined: ...


[docs]
    sampling_direction: global___SamplingDirection.ValueType

    """Sampling edge direction, INCOMING or OUTGOING. INCOMING is default since SamplingDirection enum defined INCOMING=0, OUTGOING=1
    INCOMING - Sample incoming edges to the dst nodes (default)
    OUTGOING - Sample outgoing edges from the src nodes
    """
    def __init__(
        self,
        *,
        op_name: builtins.str = ...,
        edge_type: snapchat.research.gbml.graph_schema_pb2.EdgeType | None = ...,
        input_op_names: collections.abc.Iterable[builtins.str] | None = ...,
        random_uniform: global___RandomUniform | None = ...,
        random_weighted: global___RandomWeighted | None = ...,
        top_k: global___TopK | None = ...,
        user_defined: global___UserDefined | None = ...,
        sampling_direction: global___SamplingDirection.ValueType = ...,
    ) -> None: ...

[docs]
    def HasField(self, field_name: typing_extensions.Literal["edge_type", b"edge_type", "random_uniform", b"random_uniform", "random_weighted", b"random_weighted", "sampling_method", b"sampling_method", "top_k", b"top_k", "user_defined", b"user_defined"]) -> builtins.bool: ...


[docs]
    def ClearField(self, field_name: typing_extensions.Literal["edge_type", b"edge_type", "input_op_names", b"input_op_names", "op_name", b"op_name", "random_uniform", b"random_uniform", "random_weighted", b"random_weighted", "sampling_direction", b"sampling_direction", "sampling_method", b"sampling_method", "top_k", b"top_k", "user_defined", b"user_defined"]) -> None: ...


[docs]
    def WhichOneof(self, oneof_group: typing_extensions.Literal["sampling_method", b"sampling_method"]) -> typing_extensions.Literal["random_uniform", "random_weighted", "top_k", "user_defined"] | None: ...




[docs]
global___SamplingOp = SamplingOp



[docs]
class MessagePassingPath(google.protobuf.message.Message):

[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    ROOT_NODE_TYPE_FIELD_NUMBER: builtins.int


[docs]
    SAMPLING_OPS_FIELD_NUMBER: builtins.int


[docs]
    root_node_type: builtins.str

    @property

[docs]
    def sampling_ops(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___SamplingOp]: ...

    def __init__(
        self,
        *,
        root_node_type: builtins.str = ...,
        sampling_ops: collections.abc.Iterable[global___SamplingOp] | None = ...,
    ) -> None: ...

[docs]
    def ClearField(self, field_name: typing_extensions.Literal["root_node_type", b"root_node_type", "sampling_ops", b"sampling_ops"]) -> None: ...




[docs]
global___MessagePassingPath = MessagePassingPath



[docs]
class MessagePassingPathStrategy(google.protobuf.message.Message):

[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    PATHS_FIELD_NUMBER: builtins.int

    @property

[docs]
    def paths(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___MessagePassingPath]:
        """Note: One message passing path needs to be specified per relevant node type
        i.e. for SupervisedNodeBasedTask, one message passing path per supervision node type
        for SupervisedLinkBasedTask, one message passing path per node type in the supervision edges
        We currently do not support multiple message passing paths for the same node type.
        """

    def __init__(
        self,
        *,
        paths: collections.abc.Iterable[global___MessagePassingPath] | None = ...,
    ) -> None: ...

[docs]
    def ClearField(self, field_name: typing_extensions.Literal["paths", b"paths"]) -> None: ...




[docs]
global___MessagePassingPathStrategy = MessagePassingPathStrategy



[docs]
class GlobalRandomUniformStrategy(google.protobuf.message.Message):

[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    NUM_HOPS_FIELD_NUMBER: builtins.int


[docs]
    RANDOM_UNIFORM_SPEC_FIELD_NUMBER: builtins.int


[docs]
    num_hops: builtins.int

    @property

[docs]
    def random_uniform_spec(self) -> global___RandomUniform: ...

    def __init__(
        self,
        *,
        num_hops: builtins.int = ...,
        random_uniform_spec: global___RandomUniform | None = ...,
    ) -> None: ...

[docs]
    def HasField(self, field_name: typing_extensions.Literal["random_uniform_spec", b"random_uniform_spec"]) -> builtins.bool: ...


[docs]
    def ClearField(self, field_name: typing_extensions.Literal["num_hops", b"num_hops", "random_uniform_spec", b"random_uniform_spec"]) -> None: ...




[docs]
global___GlobalRandomUniformStrategy = GlobalRandomUniformStrategy



[docs]
class SubgraphSamplingStrategy(google.protobuf.message.Message):

[docs]
    DESCRIPTOR: google.protobuf.descriptor.Descriptor



[docs]
    MESSAGE_PASSING_PATHS_FIELD_NUMBER: builtins.int


[docs]
    GLOBAL_RANDOM_UNIFORM_FIELD_NUMBER: builtins.int

    @property

[docs]
    def message_passing_paths(self) -> global___MessagePassingPathStrategy: ...

    @property

[docs]
    def global_random_uniform(self) -> global___GlobalRandomUniformStrategy:
        """Randomly sample num nodes specified nodes of each node type without replacement"""

    def __init__(
        self,
        *,
        message_passing_paths: global___MessagePassingPathStrategy | None = ...,
        global_random_uniform: global___GlobalRandomUniformStrategy | None = ...,
    ) -> None: ...

[docs]
    def HasField(self, field_name: typing_extensions.Literal["global_random_uniform", b"global_random_uniform", "message_passing_paths", b"message_passing_paths", "strategy", b"strategy"]) -> builtins.bool: ...


[docs]
    def ClearField(self, field_name: typing_extensions.Literal["global_random_uniform", b"global_random_uniform", "message_passing_paths", b"message_passing_paths", "strategy", b"strategy"]) -> None: ...


[docs]
    def WhichOneof(self, oneof_group: typing_extensions.Literal["strategy", b"strategy"]) -> typing_extensions.Literal["message_passing_paths", "global_random_uniform"] | None: ...




[docs]
global___SubgraphSamplingStrategy = SubgraphSamplingStrategy