ronantakizawa/codereview-bench huggingface dataset:
Repository: microsoft/graphrag
File: graphrag/index/verbs/genid.py (Python)

Code before:
    ### yaml
    ```yaml
    verb: genid
    args:
        to: id_output_column_name
        method: md5_hash
        hash:
            - list
            - of
            - column
            - names
        seed: 034324
    ```
    """
    data = cast(pd.DataFrame, input.source.table)

    output = genid_df(data, to, method, hash)

    return TableContainer(table=output)


def genid_df(
    input: pd.DataFrame,
    to: str,
    method: str = "md5_hash",
    hash: list[str] = [],  # noqa A002
):
    """Generate a unique id for each row in the tabular data."""
    data = input
    if method == "md5_hash":
        if len(hash) == 0:
            msg = 'Must specify the "hash" columns to use md5_hash method'
            raise ValueError(msg)

        data[to] = data.apply(lambda row: gen_md5_hash(row, hash), axis=1)
    elif method == "increment":
        data[to] = data.index + 1
    else:
        msg = f"Unknown method {method}"
        raise ValueError(msg)
    return data


Review comment (suggestion):
this is problematic in python, you might want to do `hash: list[str] | None = None`

and in the function body do:
```py
if not hash:
    hash = []
```

Code after:
    ### yaml
    ```yaml
    verb: genid
    args:
        to: id_output_column_name
        method: md5_hash
        hash:
            - list
            - of
            - column
            - names
        seed: 034324
    ```
    """
    data = cast(pd.DataFrame, input.source.table)

    output = genid_df(data, to, method, hash)

    return TableContainer(table=output)


def genid_df(
    input: pd.DataFrame,
    to: str,
    method: str = "md5_hash",
    hash: list[str] | None = None,  # noqa A002
):
    """Generate a unique id for each row in the tabular data."""
    data = input
    match method:
        case "md5_hash":
            if not hash:
                msg = 'Must specify the "hash" columns to use md5_hash method'
                raise ValueError(msg)
            data[to] = data.apply(lambda row: gen_md5_hash(row, hash), axis=1)
        case "increment":
            data[to] = data.index + 1
        case _:
            msg = f"Unknown method {method}"
            raise ValueError(msg)

    return data
Repository: microsoft/nni
File: nni/utils.py (Python)

Code before:

    def range(self, key, keyType, start, end):
        """
        Utility method to create a schema to check whether the `key` is in the range of [start, end].

        Parameters:
        ----------
        key: str
            key name of the data to be validated
        keyType: type
            python data type, such as int, float
        start: type is specified by keyType
            start of the range
        end: type is specified by keyType
            end of the range

        Returns: Schema
        --------
            A scheme to check whether the `key` is in the range of [start, end].
        """
        return And(
            And(keyType, error='%s should be %s type!' % (key, keyType.__name__)),
            And(lambda n: start <= n <= end, error='%s should be in range of (%s, %s)!' % (key, start, end))
        )

    def keyType(self, key, keyType):
        return And(keyType, error='%s should be %s type!' % (key, keyType.__name__))


Review comment (suggestion):
```python
def path(self, key, keyType):
   # check whether this path is valid and exists.
```

Code after:
        return And(lambda n: n in args, error='%s should be in [%s]!' % (key, str(args)))

    def range(self, key, keyType, start, end):
        """
        Utility method to create a schema to check whether the `key` is in the range of [start, end].

        Parameters:
        ----------
        key: str
            key name of the data to be validated
        keyType: type
            python data type, such as int, float
        start: type is specified by keyType
            start of the range
        end: type is specified by keyType
            end of the range

        Returns: Schema
        --------
            A scheme to check whether the `key` is in the range of [start, end].
        """
        return And(
            And(keyType, error='%s should be %s type!' % (key, keyType.__name__)),
            And(lambda n: start <= n <= end, error='%s should be in range of (%s, %s)!' % (key, start, end))
        )

    def path(self, key):
        return And(
            And(str, error='%s should be a string!' % key),
            And(lambda p: Path(p).exists(), error='%s path does not exist!' % (key))
        )
Repository: microsoft/nni
File: nni/retiarii/strategy/base.py (Python)

Code before:
import abc
from typing import List

from .. import report_search_space
from ..graph import Model
from ..mutator import Mutator


class BaseStrategy(abc.ABC):

    @classmethod
    def report_model_space(self, base_model: Model, applied_mutators: List[Mutator]) -> None:
        """
        NOTE: nested search space is not supported (currently also not supported on webui),
        currently, only support type `choice`.
        """
        search_space = {}
        auto_label_seq = 0
        new_model = base_model
        for mutator in applied_mutators:
            recorded_candidates, new_model = mutator.dry_run(new_model)
            # TODO: check label is never None
            if mutator.label is not None:
                label = mutator.label
            else:
                label = f'auto_label_{auto_label_seq}'
                auto_label_seq += 1
            if len(recorded_candidates) == 1:
                search_space[label] = {'_type': 'choice', '_value': recorded_candidates[0]}
            else:
                for i, candidate in enumerate(recorded_candidates):
                    search_space[f'{label}_{i}'] = {'_type': 'choice', '_value': candidate}

        report_search_space(search_space)

    @abc.abstractmethod
    def run(self, base_model: Model, applied_mutators: List[Mutator]) -> None:
        pass


Review comment (refactor):
Where does `auto_label` come from? I think this logic is already done in `dry_run_for_search_space`, and `self` is never used in this function. Suggest refact it into a util.

Code after:
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

import abc
from typing import List

from ..graph import Model
from ..mutator import Mutator


class BaseStrategy(abc.ABC):

    @abc.abstractmethod
    def run(self, base_model: Model, applied_mutators: List[Mutator]) -> None:
        pass
Repository: microsoft/nni
File: nni/retiarii/nn/pytorch/utils.py (Python)

Code before:
from abc import ABC, abstractclassmethod
from typing import Any, Optional, Tuple, Union

import torch.nn as nn
from nni.retiarii.utils import NoContextError, ModelNamespace, get_current_context


class Mutable(ABC, nn.Module):
    """
    This should be the base class for all PyTorch mutables including layer choice, input choice, etc.
    This is not considered as an interface, but rather as a base class consisting of commonly used class/instance methods.
    """

    def __new__(cls, *args, **kwargs):
        if not args and not kwargs:
            # this can be the case of copy/deepcopy
            # attributes are assigned afterwards in __dict__
            return super().__new__(cls)

        try:
            return cls.create_fixed_module(*args, **kwargs)
        except NoContextError:
            return super().__new__(cls)

    @abstractclassmethod
    def create_fixed_module(cls, *args, **kwargs) -> Union[nn.Module, Any]:
        """
        Try to create a fixed module from fixed dict.
        If the code is running in a trial, this method would succeed, and a concrete module instead of a mutable will be created.
        Raises no context error if the creation failed.
        """
        ...


Review comment (suggestion):
I'm not sure but to my understanding `ABC` is not designed for this scenario. The PEP states it's mainly for overriding `isinstance`, and I believe we don't want to treat all objects with `create_fixed_module` method as mutable. To force subclasses overriding a virtual method, `NotImplementedError` should be enough.

Anyway, I guess inheriting `ABC` will not cause problems.

Code after:
from typing import Any, Optional, Tuple, Union

import torch.nn as nn
from nni.retiarii.utils import NoContextError, ModelNamespace, get_current_context


class Mutable(nn.Module):
    """
    This is just an implementation trick for now.

    In future, this could be the base class for all PyTorch mutables including layer choice, input choice, etc.
    This is not considered as an interface, but rather as a base class consisting of commonly used class/instance methods.
    For API developers, it's not recommended to use ``isinstance(module, Mutable)`` to check for mutable modules either,
    before the design is finalized.
    """

    def __new__(cls, *args, **kwargs):
        if not args and not kwargs:
            # this can be the case of copy/deepcopy
            # attributes are assigned afterwards in __dict__
            return super().__new__(cls)

        try:
            return cls.create_fixed_module(*args, **kwargs)
        except NoContextError:
            return super().__new__(cls)

    @classmethod
    def create_fixed_module(cls, *args, **kwargs) -> Union[nn.Module, Any]:
        """
        Try to create a fixed module from fixed dict.
        If the code is running in a trial, this method would succeed, and a concrete module instead of a mutable will be created.
        Raises no context error if the creation failed.
Repository: microsoft/nni
File: nni/experiment/config/training_services/kubeflow.py (Python)

Code before:

from ..base import ConfigBase
from ..training_service import TrainingServiceConfig
from .k8s_storage import K8sStorageConfig

@dataclass(init=False)
class KubeflowRoleConfig(ConfigBase):
    replicas: int
    command: str
    gpu_number: Optional[int] = 0
    cpu_number: int
    memory_size: Union[str, int]
    docker_image: str = 'msranni/nni:latest'
    code_directory: str

@dataclass(init=False)
class KubeflowConfig(TrainingServiceConfig):
    platform: str = 'kubeflow'
    operator: str
    api_version: str
    storage: K8sStorageConfig
    worker: Optional[KubeflowRoleConfig] = None
    ps: Optional[KubeflowRoleConfig] = None
    master: Optional[KubeflowRoleConfig] = None
    reuse_mode: Optional[bool] = True #set reuse mode as true for v2 config
    namespace: Optional[str] = 'default'

    def _canonicalize(self, parents):
        super()._canonicalize(parents)
        # kubeflow does not need these fields, set empty string for type check
        if self.trial_command is None:
            self.trial_command = ''
        if self.trial_code_directory is None:
            self.trial_code_directory = ''

    def _validate_canonical(self):
        super()._validate_canonical()
        assert self.operator in ['tf-operator', 'pytorch-operator']


Review comment (suggestion):
If `null` has same effect as `"default"`, the type should be `str`.

Code after:

from ..base import ConfigBase
from ..training_service import TrainingServiceConfig
from .k8s_storage import K8sStorageConfig

@dataclass(init=False)
class KubeflowRoleConfig(ConfigBase):
    replicas: int
    command: str
    gpu_number: Optional[int] = 0
    cpu_number: int
    memory_size: Union[str, int]
    docker_image: str = 'msranni/nni:latest'
    code_directory: str

@dataclass(init=False)
class KubeflowConfig(TrainingServiceConfig):
    platform: str = 'kubeflow'
    operator: str
    api_version: str
    storage: K8sStorageConfig
    worker: Optional[KubeflowRoleConfig] = None
    ps: Optional[KubeflowRoleConfig] = None
    master: Optional[KubeflowRoleConfig] = None
    reuse_mode: Optional[bool] = True #set reuse mode as true for v2 config
    namespace: str = 'default'

    def _canonicalize(self, parents):
        super()._canonicalize(parents)
        # kubeflow does not need these fields, set empty string for type check
        if self.trial_command is None:
            self.trial_command = ''
        if self.trial_code_directory is None:
            self.trial_code_directory = ''

    def _validate_canonical(self):
        super()._validate_canonical()
        assert self.operator in ['tf-operator', 'pytorch-operator']
Repository: microsoft/nni
File: ts/nni_manager/common/experimentConfig.ts (TypeScript)

Code before:
    keyVaultName?: string;
    keyVaultKey?: string;
}

export interface KubeflowRoleConfig {
    replicas: number;
    command: string;
    gpuNumber: number;
    cpuNumber: number;
    memorySize: string | number;
    dockerImage: string;
    codeDirectory: string;
    privateRegistryAuthPath?: string;
}

export interface KubeflowConfig extends TrainingServiceConfig {
    platform: 'kubeflow';
    operator: KubeflowOperator;
    apiVersion: OperatorApiVersion;
    storage: KubernetesStorageConfig;
    worker?: KubeflowRoleConfig;
    ps?: KubeflowRoleConfig;
    master?: KubeflowRoleConfig;
    reuseMode: boolean;
    maxTrialNumberPerGpu?: number;
    namespace?: 'default';
}

export interface FrameworkControllerTaskRoleConfig {
    name: string;
    dockerImage: string;
    taskNumber: number;
    command: string;
    gpuNumber: number;
    cpuNumber: number;
    memorySize: string | number;
    frameworkAttemptCompletionPolicy: {
        minFailedTaskCount: number;
        minSucceedTaskCount: number;
    };
    privateRegistryAuthPath?: string;
}

export interface FrameworkControllerConfig extends TrainingServiceConfig {
    platform: 'frameworkcontroller';
    storage: KubernetesStorageConfig;
    serviceAccountName: string;
    taskRoles: FrameworkControllerTaskRoleConfig[];
    reuseMode: boolean;
    maxTrialNumberPerGpu?: number;
    namespace?: 'default';

Review comment (suggestion):
The type is `string`. `'default'` in ts means `Literal['default']` in python.

Code after:
    keyVaultName?: string;
    keyVaultKey?: string;
}

export interface KubeflowRoleConfig {
    replicas: number;
    command: string;
    gpuNumber: number;
    cpuNumber: number;
    memorySize: string | number;
    dockerImage: string;
    codeDirectory: string;
    privateRegistryAuthPath?: string;
}

export interface KubeflowConfig extends TrainingServiceConfig {
    platform: 'kubeflow';
    operator: KubeflowOperator;
    apiVersion: OperatorApiVersion;
    storage: KubernetesStorageConfig;
    worker?: KubeflowRoleConfig;
    ps?: KubeflowRoleConfig;
    master?: KubeflowRoleConfig;
    reuseMode: boolean;
    maxTrialNumberPerGpu?: number;
    namespace?: string;
}

export interface FrameworkControllerTaskRoleConfig {
    name: string;
    dockerImage: string;
    taskNumber: number;
    command: string;
    gpuNumber: number;
    cpuNumber: number;
    memorySize: string | number;
    frameworkAttemptCompletionPolicy: {
        minFailedTaskCount: number;
        minSucceedTaskCount: number;
    };
    privateRegistryAuthPath?: string;
}

export interface FrameworkControllerConfig extends TrainingServiceConfig {
    platform: 'frameworkcontroller';
    storage: KubernetesStorageConfig;
    serviceAccountName: string;
    taskRoles: FrameworkControllerTaskRoleConfig[];
    reuseMode: boolean;
    maxTrialNumberPerGpu?: number;
    namespace?: string;
Repository: microsoft/nni
File: nni/typehint.py (Python)

Code before:
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Types for static checking.
"""

__all__ = [
    'Literal', 'TypedDict',
    'Parameters', 'SearchSpace', 'TrialMetric', 'TrialRecord',
]

import sys
from typing import Any, Dict, List, TYPE_CHECKING

if TYPE_CHECKING or sys.version_info >= (3, 8):
    from typing import Literal, TypedDict
else:
    from typing_extensions import Literal, TypedDict

Parameters = Dict[str, Any]
"""
Return type of :func:`nni.get_next_parameter`.

For built-in tuners, this is a ``dict`` whose content is defined by :doc:`search space </hpo/search_space>`.

Customized tuners do not need to follow the constraint and can use anything serializable.
"""

class _ParameterSearchSpace(TypedDict):
    _type: Literal[
        'choice', 'randint',
        'uniform', 'loguniform', 'quniform', 'qloguniform',
        'normal', 'lognormal', 'qnormal', 'qlognormal',
    ]
    _value: List[Any]

SearchSpace = Dict[str, _ParameterSearchSpace]
"""
Type of ``experiment.config.search_space``.


Review comment (suggestion):
By the way, I think this doesn't work on Python 3.7.

Maybe it should be ``not TYPE_CHECKING and``.

Code after:
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Types for static checking.
"""

__all__ = ['Parameters', 'SearchSpace', 'TrialMetric', 'TrialRecord']

from typing import Any, Dict, List

from typing_extensions import Literal, TypedDict

Parameters = Dict[str, Any]
"""
Return type of :func:`nni.get_next_parameter`.

For built-in tuners, this is a ``dict`` whose content is defined by :doc:`search space </hpo/search_space>`.

Customized tuners do not need to follow the constraint and can use anything serializable.
"""

class _ParameterSearchSpace(TypedDict):
    _type: Literal[
        'choice', 'randint',
        'uniform', 'loguniform', 'quniform', 'qloguniform',
        'normal', 'lognormal', 'qnormal', 'qlognormal',
    ]
    _value: List[Any]

SearchSpace = Dict[str, _ParameterSearchSpace]
"""
Type of ``experiment.config.search_space``.

For built-in tuners, the format is detailed in :doc:`/hpo/search_space`.

Customized tuners do not need to follow the constraint and can use anything serializable, except ``None``.
"""

TrialMetric = float
"""
Repository: microsoft/nni
File: test/ut/retiarii/test_multitrial.py (Python)

Code before:
        assert isinstance(exp.export_top_models()[0], dict)
        exp.stop()

python_script = """
from nni.retiarii import strategy
from nni.retiarii.experiment.pytorch import RetiariiExeConfig, RetiariiExperiment
from test_oneshot import _mnist_net

base_model, evaluator = _mnist_net('simple', {'max_epochs': 1})
search_strategy = strategy.Random()
exp = RetiariiExperiment(base_model, evaluator, strategy=search_strategy)
exp_config = RetiariiExeConfig('local')
exp_config.experiment_name = 'mnist_unittest'
exp_config.trial_concurrency = 2
exp_config.max_trial_number = 2
exp_config.training_service.use_active_gpu = False
exp.run(exp_config, 8080)
assert isinstance(exp.export_top_models()[0], dict)
"""

def test_exp_exit_without_stop():
    script_name = 'tmp_multi_trial.py'
    with open(script_name, 'w') as f:
        f.write(python_script)
    proc = Popen(['python3', script_name])
    proc.wait()
    os.remove(script_name)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--exp', type=str, default='all', metavar='E',
                        help='experiment to run, default = all')
    args = parser.parse_args()

    if args.exp == 'all':
        test_multi_trial()
        test_exp_exit_without_stop()
    else:
        globals()[f'test_{args.exp}']()


Review comment (suggestion):
You can write a function creating this experiment and use `multiprocessing.Process(target=my_testing_experiment).start()` to launch the sub-process.

Code after:
        exp_config.trial_concurrency = 1
        exp_config.max_trial_number = 1
        exp_config.training_service.use_active_gpu = False
        exp.run(exp_config, 8080)
        assert isinstance(exp.export_top_models()[0], dict)
        exp.stop()

python_script = """
from nni.retiarii import strategy
from nni.retiarii.experiment.pytorch import RetiariiExeConfig, RetiariiExperiment
from test_oneshot import _mnist_net

base_model, evaluator = _mnist_net('simple', {'max_epochs': 1})
search_strategy = strategy.Random()
exp = RetiariiExperiment(base_model, evaluator, strategy=search_strategy)
exp_config = RetiariiExeConfig('local')
exp_config.experiment_name = 'mnist_unittest'
exp_config.trial_concurrency = 1
exp_config.max_trial_number = 1
exp_config.training_service.use_active_gpu = False
exp.run(exp_config, 8080)
assert isinstance(exp.export_top_models()[0], dict)
"""

@pytest.mark.timeout(600)
def test_exp_exit_without_stop():
    script_name = 'tmp_multi_trial.py'
    with open(script_name, 'w') as f:
        f.write(python_script)
    proc = Popen([sys.executable, script_name])
    proc.wait()
    os.remove(script_name)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--exp', type=str, default='all', metavar='E',
                        help='experiment to run, default = all')
    args = parser.parse_args()

    if args.exp == 'all':
        test_multi_trial()
        test_exp_exit_without_stop()
    else:
        globals()[f'test_{args.exp}']()
Repository: microsoft/nni
File: nni/experiment/config/utils/internal.py (Python)

Code before:
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Utility functions for experiment config classes, internal part.

If you are implementing a config class for a training service, it's unlikely you will need these.
"""

from __future__ import annotations

__all__ = [
    'get_base_path', 'set_base_path', 'unset_base_path', 'resolve_path',
    'case_insensitive', 'camel_case',
    'fields', 'is_instance', 'validate_type', 'is_path_like',
    'guess_config_type', 'guess_list_config_type',
    'training_service_config_factory', 'load_training_service_config',
    'get_ipv4_address', 'init_experiment_config', 'get_experiment_class_using_config'
]

import copy
import dataclasses
import importlib
import json
import os.path
from pathlib import Path
import socket
import typing

import typeguard

import nni.runtime.config

from .public import is_missing

if typing.TYPE_CHECKING:
    from nni.retiarii.experiment.config.experiment_config import RetiariiExeConfig
    from ..base import ConfigBase
    from ..experiment_config import ExperimentConfig
    from ..training_service import TrainingServiceConfig

## handle relative path ##


Review comment (suggestion):
Suggest to put these two functions below `training_service_config_factory` and `load_training_service_config`.
The lines are grouped.

Code after:
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""
Utility functions for experiment config classes, internal part.

If you are implementing a config class for a training service, it's unlikely you will need these.
"""

from __future__ import annotations

__all__ = [
    'get_base_path', 'set_base_path', 'unset_base_path', 'resolve_path',
    'case_insensitive', 'camel_case',
    'fields', 'is_instance', 'validate_type', 'is_path_like',
    'guess_config_type', 'guess_list_config_type',
    'training_service_config_factory', 'load_training_service_config',
    'load_experiment_config', 'get_experiment_cls_using_config',
    'get_ipv4_address'
]

import copy
import dataclasses
import importlib
import json
import os.path
from pathlib import Path
import socket
from typing import Tuple, TYPE_CHECKING, get_type_hints

import typeguard

import nni.runtime.config

from .public import is_missing

if TYPE_CHECKING:
    from nni.nas.experiment.pytorch import RetiariiExperiment
    from nni.nas.experiment.config import RetiariiExeConfig
    from ...experiment import Experiment
    from ..base import ConfigBase
    from ..experiment_config import ExperimentConfig
    from ..training_service import TrainingServiceConfig