<commit_before>from typing import Any, Dict, List, Union
from collections import OrderedDict

from ray.experimental.dag import (
    DAGNode,
    ClassNode,
    ClassMethodNode,
    PARENT_CLASS_NODE_KEY,
)
from ray.experimental.dag.function_node import FunctionNode
from ray.experimental.dag.input_node import InputNode
from ray.serve.deployment import Deployment
from ray.serve.pipeline.deployment_method_node import DeploymentMethodNode
from ray.serve.pipeline.deployment_node import DeploymentNode
from ray.serve.pipeline.deployment_function_node import DeploymentFunctionNode


class DeploymentNameGenerator(object):
    """
    Generate unique suffix for each given deployment_name requested for name.
    By default uses deployment_name for the very first time, then append
    monotonic increasing id to it.
    """

    def __init__(self):
        self.name_to_suffix: Dict[str, int] = dict()

    def get_deployment_name(self, dag_node: Union[ClassNode, FunctionNode]):
        assert isinstance(dag_node, (ClassNode, FunctionNode)), (
            "get_deployment_name() should only be called on ClassNode or "
            "FunctionNode instances."
        )

        deployment_name = (
            dag_node.get_options().get("name", None) or dag_node._body.__name__
        )
        if deployment_name not in self.name_to_suffix:
            self.name_to_suffix[deployment_name] = 0
            return deployment_name
        else:
            self.name_to_suffix[deployment_name] += 1
            suffix_num = self.name_to_suffix[deployment_name]

            return f"{deployment_name}_{suffix_num}"

    def reset(self):
        self.name_to_suffix = dict()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.reset()


def _remove_non_default_ray_actor_options(ray_actor_options: Dict[str, Any]):
    """
    In Ray DAG building we pass full ray_actor_options regardless if a field
    was explicitly set. Since some values are invalid, we need to remove them
    from ray_actor_options.
    """
    # TODO: (jiaodong) Revisit when we implement build() when user explicitly
    # pass default value
    ray_actor_options = {k: v for k, v in ray_actor_options.items() if v}
    if ray_actor_options.get("placement_group") == "default":
        del ray_actor_options["placement_group"]
    if ray_actor_options.get("placement_group_bundle_index") == -1:
        del ray_actor_options["placement_group_bundle_index"]
    if ray_actor_options.get("max_pending_calls") == -1:
        del ray_actor_options["max_pending_calls"]

    return ray_actor_options


def transform_ray_dag_to_serve_dag(
    dag_node: DAGNode, deployment_name_generator: DeploymentNameGenerator
):
    """
    Transform a Ray DAG to a Serve DAG. Map ClassNode to DeploymentNode with
    ray decorated body passed in, and ClassMethodNode to DeploymentMethodNode.
    """
    if isinstance(dag_node, ClassNode):
        deployment_name = deployment_name_generator.get_deployment_name(dag_node)
        ray_actor_options = _remove_non_default_ray_actor_options(
            dag_node.get_options()
        )
        return DeploymentNode(
            dag_node._body,
            deployment_name,
            dag_node.get_args(),
            dag_node.get_kwargs(),
            ray_actor_options,
            # TODO: (jiaodong) Support .options(metadata=xxx) for deployment
            other_args_to_resolve=dag_node.get_other_args_to_resolve(),
        )

    elif isinstance(dag_node, ClassMethodNode):
        other_args_to_resolve = dag_node.get_other_args_to_resolve()
        # TODO: (jiaodong) Need to capture DAGNodes in the parent node<commit_msg>[serve] Remove unnecessary code  (#24131)<commit_after>from typing import Dict, List, Union
from collections import OrderedDict

from ray.experimental.dag import (
    DAGNode,
    ClassNode,
    ClassMethodNode,
    PARENT_CLASS_NODE_KEY,
)
from ray.experimental.dag.function_node import FunctionNode
from ray.experimental.dag.input_node import InputNode
from ray.serve.deployment import Deployment
from ray.serve.pipeline.deployment_method_node import DeploymentMethodNode
from ray.serve.pipeline.deployment_node import DeploymentNode
from ray.serve.pipeline.deployment_function_node import DeploymentFunctionNode


class DeploymentNameGenerator(object):
    """
    Generate unique suffix for each given deployment_name requested for name.
    By default uses deployment_name for the very first time, then append
    monotonic increasing id to it.
    """

    def __init__(self):
        self.name_to_suffix: Dict[str, int] = dict()

    def get_deployment_name(self, dag_node: Union[ClassNode, FunctionNode]):
        assert isinstance(dag_node, (ClassNode, FunctionNode)), (
            "get_deployment_name() should only be called on ClassNode or "
            "FunctionNode instances."
        )

        deployment_name = (
            dag_node.get_options().get("name", None) or dag_node._body.__name__
        )
        if deployment_name not in self.name_to_suffix:
            self.name_to_suffix[deployment_name] = 0
            return deployment_name
        else:
            self.name_to_suffix[deployment_name] += 1
            suffix_num = self.name_to_suffix[deployment_name]

            return f"{deployment_name}_{suffix_num}"

    def reset(self):
        self.name_to_suffix = dict()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.reset()


def transform_ray_dag_to_serve_dag(
    dag_node: DAGNode, deployment_name_generator: DeploymentNameGenerator
):
    """
    Transform a Ray DAG to a Serve DAG. Map ClassNode to DeploymentNode with
    ray decorated body passed in, and ClassMethodNode to DeploymentMethodNode.
    """
    if isinstance(dag_node, ClassNode):
        deployment_name = deployment_name_generator.get_deployment_name(dag_node)
        return DeploymentNode(
            dag_node._body,
            deployment_name,
            dag_node.get_args(),
            dag_node.get_kwargs(),
            dag_node.get_options(),
            # TODO: (jiaodong) Support .options(metadata=xxx) for deployment
            other_args_to_resolve=dag_node.get_other_args_to_resolve(),
        )

    elif isinstance(dag_node, ClassMethodNode):
        other_args_to_resolve = dag_node.get_other_args_to_resolve()
        # TODO: (jiaodong) Need to capture DAGNodes in the parent node
{'fraction_non_alphanumeric': 0.07339449541284404, 'fraction_numerical': 0.001710464935468823, 'mean_word_length': 3.276595744680851, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1197370', 'n_tokens_mistral': 1871, 'n_tokens_neox': 1834, 'n_words': 456}