ronantakizawa/codereview-bench huggingface dataset:
Repository: Project-MONAI/MONAI
File: monai/metrics/cumulative_average.py (Python)

Code before:
    """

    def __init__(self) -> None:
        self.reset()

    def reset(self) -> None:
        """
        Reset all  stats
        """
        self.val: torch.Tensor = None  # type: ignore
        self.sum = torch.tensor(0, dtype=torch.float)
        self.count = torch.tensor(0, dtype=torch.float)
        self.is_distributed = dist.is_available() and dist.is_initialized()

    def get_current(self, to_numpy: bool = True) -> NdarrayOrTensor:
        """
        returns the most recent value (averaged across processes)

        Args:
            to_numpy: whether to convert to numpy array. Defaults to True
        """
        if self.val is None:
            return 0

        val = self.val.clone()
        val[~torch.isfinite(val)] = 0

        if self.is_distributed:
            val = val / dist.get_world_size()
            dist.all_reduce(val)

        if to_numpy:
            val = val.cpu().numpy()

        return val

    def aggregate(self, to_numpy: bool = True) -> NdarrayOrTensor:
        """
        returns the total average value (averaged across processes)

        Args:
            to_numpy: whether to convert to numpy array. Defaults to True
        """
        if self.val is None:
            return 0

        sum = self.sum
        count = self.count

        if self.is_distributed:
            sum = sum.to(self.val, copy=True)

Review comment (suggestion):
```suggestion
        val: NdarrayOrTensor = self.val.clone()
```
Does combining the annotation with the assignment cause an issue? This should work by deducing the type of the return value of `clone`.

Code after:
    """

    def __init__(self) -> None:
        self.reset()

    def reset(self) -> None:
        """
        Reset all  stats
        """
        self.val: torch.Tensor = None  # type: ignore
        self.sum = torch.tensor(0, dtype=torch.float)
        self.count = torch.tensor(0, dtype=torch.float)
        self.is_distributed = dist.is_available() and dist.is_initialized()

    def get_current(self, to_numpy: bool = True) -> NdarrayOrTensor:
        """
        returns the most recent value (averaged across processes)

        Args:
            to_numpy: whether to convert to numpy array. Defaults to True
        """
        if self.val is None:
            return 0

        val: NdarrayOrTensor
        val = self.val.clone()
        val[~torch.isfinite(val)] = 0

        if self.is_distributed:
            val = val / dist.get_world_size()
            dist.all_reduce(val)

        if to_numpy:
            val = val.cpu().numpy()

        return val

    def aggregate(self, to_numpy: bool = True) -> NdarrayOrTensor:
        """
        returns the total average value (averaged across processes)

        Args:
            to_numpy: whether to convert to numpy array. Defaults to True
        """
        if self.val is None:
            return 0

        sum = self.sum
        count = self.count

        if self.is_distributed:
Repository: Project-MONAI/MONAI
File: tests/test_diffusion_model_unet.py (Python)

Code before:
        # test loading weights from a model saved in MONAI Generative, version 0.2.3
        with skip_if_downloading_fails():
            net = DiffusionModelUNet(
                spatial_dims=2,
                in_channels=1,
                out_channels=1,
                num_res_blocks=1,
                channels=(8, 8, 8),
                attention_levels=(False, False, True),
                with_conditioning=True,
                cross_attention_dim=3,
                transformer_num_layers=1,
                norm_num_groups=8,
            )

            tmpdir = tempfile.mkdtemp()
            key = "diffusion_model_unet_monai_generative_weights"
            url = testing_data_config("models", key, "url")
            hash_type = testing_data_config("models", key, "hash_type")
            hash_val = testing_data_config("models", key, "hash_val")
            filename = "diffusion_model_unet_monai_generative_weights.pt"

            weight_path = os.path.join(tmpdir, filename)
            download_url(url=url, filepath=weight_path, hash_val=hash_val, hash_type=hash_type)

            net.load_old_state_dict(torch.load(weight_path), verbose=True)


if __name__ == "__main__":
    unittest.main()


Review comment (suggestion):
```suggestion
            net.load_old_state_dict(torch.load(weight_path), verbose=False)
```

Code after:
        # test loading weights from a model saved in MONAI Generative, version 0.2.3
        with skip_if_downloading_fails():
            net = DiffusionModelUNet(
                spatial_dims=2,
                in_channels=1,
                out_channels=1,
                num_res_blocks=1,
                channels=(8, 8, 8),
                attention_levels=(False, False, True),
                with_conditioning=True,
                cross_attention_dim=3,
                transformer_num_layers=1,
                norm_num_groups=8,
            )

            tmpdir = tempfile.mkdtemp()
            key = "diffusion_model_unet_monai_generative_weights"
            url = testing_data_config("models", key, "url")
            hash_type = testing_data_config("models", key, "hash_type")
            hash_val = testing_data_config("models", key, "hash_val")
            filename = "diffusion_model_unet_monai_generative_weights.pt"

            weight_path = os.path.join(tmpdir, filename)
            download_url(url=url, filepath=weight_path, hash_val=hash_val, hash_type=hash_type)

            net.load_old_state_dict(torch.load(weight_path), verbose=False)


if __name__ == "__main__":
    unittest.main()
Repository: Project-MONAI/MONAI
File: monai/config/deviceconfig.py (Python)

Code before:

def print_debug_info(file: TextIO = sys.stdout) -> None:
    """
    Print config (installed dependencies, etc.) and system info for debugging.

    Args:
        file: `print()` text stream file. Defaults to `sys.stdout`.
    """
    print("================================", file=file, flush=True)
    print("Printing MONAI config...", file=file, flush=True)
    print("================================", file=file, flush=True)
    print_config(file)
    print("\n================================", file=file, flush=True)
    print("Printing system config...")
    print("================================", file=file, flush=True)
    print_system_info(file)
    print("\n================================", file=file, flush=True)
    print("Printing GPU config...")
    print("================================", file=file, flush=True)
    print_gpu_info(file)


@deprecated(since=1.4, removed=1.6, msg_suffix="Please use `monai.utils.enums.IgniteInfo` instead.")
class IgniteInfo:
    """
    IgniteInfo enum is moved to `monai.utils.enums.IgniteInfo`.
    """

    OPT_IMPORT_VERSION = NewIgniteInfo.OPT_IMPORT_VERSION


if __name__ == "__main__":
    print_debug_info()


Review comment (suggestion):
We can condense these three docstring lines into one:

```
"""Deprecated Import of IgniteInfo enum, which was moved to `monai.utils.enums.IgniteInfo`."""
```

Code after:


def print_debug_info(file: TextIO = sys.stdout) -> None:
    """
    Print config (installed dependencies, etc.) and system info for debugging.

    Args:
        file: `print()` text stream file. Defaults to `sys.stdout`.
    """
    print("================================", file=file, flush=True)
    print("Printing MONAI config...", file=file, flush=True)
    print("================================", file=file, flush=True)
    print_config(file)
    print("\n================================", file=file, flush=True)
    print("Printing system config...")
    print("================================", file=file, flush=True)
    print_system_info(file)
    print("\n================================", file=file, flush=True)
    print("Printing GPU config...")
    print("================================", file=file, flush=True)
    print_gpu_info(file)


@deprecated(since="1.4.0", removed="1.6.0", msg_suffix="Please use `monai.utils.enums.IgniteInfo` instead.")
class IgniteInfo:
    """Deprecated Import of IgniteInfo enum, which was moved to `monai.utils.enums.IgniteInfo`."""

    OPT_IMPORT_VERSION = _IgniteInfo.OPT_IMPORT_VERSION


if __name__ == "__main__":
    print_debug_info()
Repository: Project-MONAI/MONAI
File: monai/utils/misc.py (Python)

Code before:
    num = ensure_tuple(num)
    sqrt_num = [int(math.sqrt(_num)) for _num in num]
    ret = [_i * _j for _i, _j in zip(sqrt_num, sqrt_num)]
    return ensure_tuple(ret) == num


def unsqueeze_right(arr: NT, ndim: int) -> NT:
    """Append 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(...,) + (None,) * (ndim - arr.ndim)]


def unsqueeze_left(arr: NT, ndim: int) -> NT:
    """Prepend 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(None,) * (ndim - arr.ndim)]


def flatten_dict(metrics: dict[str, Any]) -> dict[str, Any]:
    """
    Flatten the nested dictionary to a flat dictionary.
    """
    result = {}
    for key, value in metrics.items():
        if isinstance(value, dict):
            result.update(flatten_dict(value))
        else:
            result[f"{key}"] = value
    return result


Review comment (suggestion):
May I ask why not use `result[key]` directly since the input type hint has already indicate all keys are in `str` type.

Thanks,  
Bin

Code after:
    num = ensure_tuple(num)
    sqrt_num = [int(math.sqrt(_num)) for _num in num]
    ret = [_i * _j for _i, _j in zip(sqrt_num, sqrt_num)]
    return ensure_tuple(ret) == num


def unsqueeze_right(arr: NT, ndim: int) -> NT:
    """Append 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(...,) + (None,) * (ndim - arr.ndim)]


def unsqueeze_left(arr: NT, ndim: int) -> NT:
    """Prepend 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(None,) * (ndim - arr.ndim)]


def flatten_dict(metrics: dict[str, Any]) -> dict[str, Any]:
    """
    Flatten the nested dictionary to a flat dictionary.
    """
    result = {}
    for key, value in metrics.items():
        if isinstance(value, dict):
            result.update(flatten_dict(value))
        else:
            result[key] = value
    return result
Repository: Project-MONAI/MONAI
File: monai/utils/misc.py (Python)

Code before:
        output = str(e.stdout.decode(errors="replace"))
        errors = str(e.stderr.decode(errors="replace"))
        raise RuntimeError(f"subprocess call error {e.returncode}: {errors}, {output}.") from e


def is_sqrt(num: Sequence[int] | int) -> bool:
    """
    Determine if the input is a square number or a squence of square numbers.
    """
    num = ensure_tuple(num)
    sqrt_num = [int(math.sqrt(_num)) for _num in num]
    ret = [_i * _j for _i, _j in zip(sqrt_num, sqrt_num)]
    return ensure_tuple(ret) == num


def unsqueeze_right(arr: NT, ndim: int) -> NT:
    """Append 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(...,) + (None,) * (ndim - arr.ndim)]


def unsqueeze_left(arr: NT, ndim: int) -> NT:
    """Prepend 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(None,) * (ndim - arr.ndim)]


def flatten_dict(metrics: dict[str, Any]) -> dict[str, Any]:
    """
    Flatten the nested dictionary to a flat dictionary.
    """
    result = {}
    for key, value in metrics.items():
        if isinstance(value, dict):
            result.update(flatten_dict(value))
        else:
            result[f"{key}"] = value
    return result


Review comment (suggestion):
I think there is a potential risk when the input dict has the same key in different dict level. For example:
```
input_dict = {"a":1, "b":2, "c": {"a":3, "d":4}}
print(flatten_dict(input_dict))
```

The output will be:
```
{'a': 3, 'b': 2, 'd': 4}
```
I would suggest to reserve the key as a prefix when its value is a dict.

Thanks,  
Bin

Code after:
        output = str(e.stdout.decode(errors="replace"))
        errors = str(e.stderr.decode(errors="replace"))
        raise RuntimeError(f"subprocess call error {e.returncode}: {errors}, {output}.") from e


def is_sqrt(num: Sequence[int] | int) -> bool:
    """
    Determine if the input is a square number or a squence of square numbers.
    """
    num = ensure_tuple(num)
    sqrt_num = [int(math.sqrt(_num)) for _num in num]
    ret = [_i * _j for _i, _j in zip(sqrt_num, sqrt_num)]
    return ensure_tuple(ret) == num


def unsqueeze_right(arr: NT, ndim: int) -> NT:
    """Append 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(...,) + (None,) * (ndim - arr.ndim)]


def unsqueeze_left(arr: NT, ndim: int) -> NT:
    """Prepend 1-sized dimensions to `arr` to create a result with `ndim` dimensions."""
    return arr[(None,) * (ndim - arr.ndim)]


def flatten_dict(metrics: dict[str, Any]) -> dict[str, Any]:
    """
    Flatten the nested dictionary to a flat dictionary.
    """
    result = {}
    for key, value in metrics.items():
        if isinstance(value, dict):
            result.update(flatten_dict(value))
        else:
            result[key] = value
    return result
Repository: Project-MONAI/MONAI
File: tests/transforms/croppad/test_pad_nd_dtypes.py (Python)

Code before:
            ("constant", torch.float32),
            ("reflect", torch.bool),
            ("reflect", torch.int8),
            ("reflect", torch.float32),
            ("replicate", torch.bool),
            ("replicate", torch.int8),
            ("replicate", torch.float32),
        ]
    )
    def test_pad_multiple_modes_dtype_preserved(self, mode, dtype):
        """Test that pad_nd preserves dtype across multiple padding modes."""
        img = torch.ones((1, 4, 4), dtype=dtype)
        to_pad = [(0, 0), (1, 1), (2, 2)]

        kwargs = {"value": 0} if mode == "constant" else {}
        out = pad_nd(img, to_pad, mode=mode, **kwargs)

        self.assertEqual(out.shape, (1, 6, 8))
        self.assertEqual(out.dtype, img.dtype)

    def test_value_with_non_constant_mode_raises(self):
        """Test that pad_nd raises ValueError when 'value' is provided with non-constant mode."""
        img = torch.ones((1, 4, 4))
        to_pad = [(0, 0), (1, 1), (2, 2)]
        with self.assertRaises(ValueError):
            pad_nd(img, to_pad, mode="reflect", **{"value": 0})


if __name__ == "__main__":
    unittest.main()


Review comment (suggestion):
```suggestion
            pad_nd(img, to_pad, mode="reflect", value=0)
```

Code after:
    def test_pad_dtype_no_error_and_dtype_preserved(self, dtype):
        """Test that pad_nd handles various dtypes without error and preserves dtype.
        Args:
            dtype: Input dtype under test.
        """
        img = torch.ones((1, 4, 4), dtype=dtype)
        to_pad = [(0, 0), (1, 1), (2, 2)]
        out = pad_nd(img, to_pad, mode="constant", value=0)

        self.assertEqual(out.shape, (1, 6, 8))
        self.assertEqual(out.dtype, img.dtype)

    @parameterized.expand(MODES_DTYPES)
    def test_pad_multiple_modes_dtype_preserved(self, mode, dtype):
        """Test that pad_nd preserves dtype across multiple padding modes.
        Args:
            mode: Padding mode under test.
            dtype: Input dtype under test.
        """
        img = torch.ones((1, 4, 4), dtype=dtype)
        to_pad = [(0, 0), (1, 1), (2, 2)]

        kwargs = {"value": 0} if mode == "constant" else {}
        out = pad_nd(img, to_pad, mode=mode, **kwargs)

        self.assertEqual(out.shape, (1, 6, 8))
        self.assertEqual(out.dtype, img.dtype)

    def test_value_with_non_constant_mode_raises(self):
        """Test that pad_nd raises ValueError when 'value' is provided with non-constant mode."""
        img = torch.ones((1, 4, 4))
        to_pad = [(0, 0), (1, 1), (2, 2)]
        with self.assertRaises(ValueError):
            pad_nd(img, to_pad, mode="reflect", value=0)


if __name__ == "__main__":
    unittest.main()
Repository: Pythagora-io/gpt-pilot
File: pilot/main.py (Python)

Code before:
            return ipc_client_instance.listen()
        
    def local_print(*args, **kwargs):
        message = " ".join(map(str, args))
        if 'type' in kwargs:
            if kwargs['type'] == MESSAGE_TYPE['info']:
                return
            del kwargs['type']

        built_in_print(message, **kwargs)

    ipc_client_instance = None
    if '--external-log-process-port' in args:
        ipc_client_instance = IPCClient(args['--external-log-process-port'])
        return print_to_external_process, ipc_client_instance
    else:
        return local_print, ipc_client_instance

if __name__ == "__main__":
    args = init()

    builtins.print, ipc_client_instance = get_custom_print(args)
    if '--api-key' in args:
        const.config.API_KEY = args['--api-key']

    if '--root-folder' in args:
        const.config.ROOT_FOLDER = args['--root-folder']

    if '--get-created-apps-with-steps' in args:
        print({ 'db_data': get_created_apps_with_steps() }, type='info')
    else:
        # TODO get checkpoint from database and fill the project with it
        project = Project(args, ipc_client_instance=ipc_client_instance)
        project.start()

Review comment (suggestion):
`workspace` can now be specified as a CLI arg

https://github.com/Pythagora-io/gpt-pilot/blob/main/pilot/utils/arguments.py#L31

Code after:
            return ipc_client_instance.listen()
        
    def local_print(*args, **kwargs):
        message = " ".join(map(str, args))
        if 'type' in kwargs:
            if kwargs['type'] == MESSAGE_TYPE['info']:
                return
            del kwargs['type']

        built_in_print(message, **kwargs)

    ipc_client_instance = None
    if '--external-log-process-port' in args:
        ipc_client_instance = IPCClient(args['--external-log-process-port'])
        return print_to_external_process, ipc_client_instance
    else:
        return local_print, ipc_client_instance

if __name__ == "__main__":
    args = init()

    builtins.print, ipc_client_instance = get_custom_print(args)
    if '--api-key' in args:
        os.environ["OPENAI_API_KEY"] = args['--api-key']

    if '--root-folder' in args:
        const.config.ROOT_FOLDER = args['--root-folder']

    if '--get-created-apps-with-steps' in args:
        print({ 'db_data': get_created_apps_with_steps() }, type='info')
    else:
        # TODO get checkpoint from database and fill the project with it
        project = Project(args, ipc_client_instance=ipc_client_instance)
        project.start()
Repository: Pythagora-io/gpt-pilot
File: pilot/utils/files.py (Python)

Code before:
    create_directory(project_path, 'tests')
    try:
        save_user_app(args.get('user_id'), args.get('app_id'), project_path)
    except Exception as e:
        print(f'Error saving user app: {str(e)}')

    print(os.path.basename(project_path), type='project_folder_name')
    return project_path


def create_directory(parent_directory, new_directory):
    new_directory_path = os.path.join(parent_directory, new_directory)
    os.makedirs(new_directory_path, exist_ok=True)

    return new_directory_path


def count_lines_of_code(files):
    total_lines = 0

    for file in files:
        # Split the content into lines
        lines = file['content'].split('\n')

        # Count the lines, skipping empty lines and comments if desired
        line_count = sum(1 for line in lines if line.strip() and not line.strip().startswith('//'))
        total_lines += line_count

    return total_lines


Review comment (suggestion):
I don't think there's a need to skip empty lines and comments. These are usually counted in the file lengths, unless you're talking about "real code lines" which is a bit more complex to calculate than this.

This also only removes one-line JS comments, totally ignoring `#` style comments and multi-line `/* */` comments, so it's not terribly accurate.

Suggested solution:

```
return sum(len(file.splitlines()) for file in files)
```

Code after:
import os
from pathlib import Path
from database.database import save_user_app


def get_parent_folder(folder_name):
    current_path = Path(os.path.abspath(__file__))  # get the path of the current script

    while current_path.name != folder_name:  # while the current folder name is not 'folder_name'
        current_path = current_path.parent  # go up one level

    return current_path.parent


def setup_workspace(args) -> str:
    """
    Creates & returns the path to the project workspace.
    Also creates a 'tests' folder inside the workspace.
    :param args: may contain 'root' key
    """
    workspace = args.get('workspace')
    if workspace:
        project_path = workspace
    else:
        root = args.get('root') or get_parent_folder('pilot')
        name = args.get('name', 'default_project_name')
        project_path = create_directory(os.path.join(root, 'workspace'), name)

    create_directory(project_path, 'tests')
    try:
        save_user_app(args.get('user_id'), args.get('app_id'), project_path)
    except Exception as e:
        print(f'Error saving user app: {str(e)}')

    print(os.path.basename(project_path), type='project_folder_name')
    return project_path


def create_directory(parent_directory, new_directory):
    new_directory_path = os.path.join(parent_directory, new_directory)
    os.makedirs(new_directory_path, exist_ok=True)

    return new_directory_path


def count_lines_of_code(files):
    return sum(len(file['content'].splitlines()) for file in files)
Repository: Pythagora-io/gpt-pilot
File: pilot/helpers/files.py (Python)

Code before:
from pathlib import Path
import os
from typing import Optional, Union, Dict

from utils.style import color_green


def update_file(path: str, new_content: Union[str, bytes]):
    """
    Update file with the new content.

    :param path: Full path to the file
    :param new_content: New content to write to the file

    Any intermediate directories will be created if they don't exist.
    If file is text, it will be written using UTF-8 encoding.
    """
    # TODO: we should know where project root is and ensure no
    # files are written outside of it.
    os.makedirs(os.path.dirname(path), exist_ok=True)

    if isinstance(new_content, str):
        file_mode = "w"
        encoding = "utf-8"
    else:
        file_mode = "wb"
        encoding = None


Review comment (suggestion):
Using `Dict`, `List`, etc [is deprecated](https://peps.python.org/pep-0585/#implementation) and from 3.9 (which is the minimal version we support), `dict`, etc should be used

Code after:
from pathlib import Path
import os
from typing import Optional, Union

from utils.style import color_green


def update_file(path: str, new_content: Union[str, bytes]):
    """
    Update file with the new content.

    :param path: Full path to the file
    :param new_content: New content to write to the file

    Any intermediate directories will be created if they don't exist.
    If file is text, it will be written using UTF-8 encoding.
    """
    # TODO: we should know where project root is and ensure no
    # files are written outside of it.
    os.makedirs(os.path.dirname(path), exist_ok=True)

    if isinstance(new_content, str):
        file_mode = "w"
        encoding = "utf-8"
    else:
        file_mode = "wb"
        encoding = None