Repository: Project-MONAI/MONAI
File: monai/transforms/io/array.py (Python)

Code before:
            "    Please install the writer libraries, see also the installation instructions:\n"
            "    https://docs.monai.io/en/latest/installation.html#installing-the-recommended-dependencies.\n"
            f"   The current registered writers for {self.output_ext}: {self.writers}.\n{msg}"
        )


class MappingJson(Transform):
    """
    Writes a JSON file that logs the mapping between input image paths and their corresponding output paths.

    Args:
        mapping_json_path (Path or str): Path to the JSON file where the mappings will be saved.
    """

    def __init__(self, mapping_json_path: Path | str = "mapping.json"):
        self.mapping_json_path = Path(mapping_json_path)

    def write_json(self, input_path: str, output_path: str):
        """
        Args:
            input_path (str): The path of the input image file.
            output_path (str): The path of the output image file.
        """
        log_data = {"input": input_path, "output": output_path}
        try:
            with self.mapping_json_path.open("r") as f:
                existing_log_data = json.load(f)
        except FileNotFoundError:
            existing_log_data = []

        existing_log_data.append(log_data)

        with self.mapping_json_path.open("w") as f:
            json.dump(existing_log_data, f, indent=4)

    def __call__(self, img: MetaTensor):
        """
        Args:
            img (MetaTensor): The input image with metadata.
        """
        if "saved_to" not in img.meta:
            raise KeyError("Missing 'saved_to' key in metadata. Check SaveImage savepath_in_metadict.")

        input_path = img.meta["filename_or_obj"]
        output_path = img.meta["saved_to"]
        self.write_json(input_path, output_path)
        return img


Review comment (bug):
I think this part is unsafe for the multiprocess environment. Could you please add a multiprocess test case reading/writing the same file here? If the test turns out failure, I think we should consider to use the [filelock](https://py-filelock.readthedocs.io/en/latest/index.html) to make this one safe. @ericspod  Please correct me if I am wrong about this one.

Thanks  
Bin

Code after:
            else:
                self._data_index += 1
                if self.savepath_in_metadict and meta_data is not None:
                    meta_data[MetaKeys.SAVED_TO] = filename
                return img
        msg = "\n".join([f"{e}" for e in err])
        raise RuntimeError(
            f"{self.__class__.__name__} cannot find a suitable writer for {filename}.\n"
            "    Please install the writer libraries, see also the installation instructions:\n"
            "    https://docs.monai.io/en/latest/installation.html#installing-the-recommended-dependencies.\n"
            f"   The current registered writers for {self.output_ext}: {self.writers}.\n{msg}"
        )


class WriteFileMapping(Transform):
    """
    Writes a JSON file that logs the mapping between input image paths and their corresponding output paths.
    This class uses FileLock to ensure safe writing to the JSON file in a multiprocess environment.

    Args:
        mapping_file_path (Path or str): Path to the JSON file where the mappings will be saved.
    """

    def __init__(self, mapping_file_path: Path | str = "mapping.json"):
        self.mapping_file_path = Path(mapping_file_path)

    def __call__(self, img: NdarrayOrTensor):
        """
        Args:
            img: The input image with metadata.
        """
        if isinstance(img, MetaTensor):
            meta_data = img.meta

        if MetaKeys.SAVED_TO not in meta_data:
            raise KeyError(
                "Missing 'saved_to' key in metadata. Check SaveImage argument 'savepath_in_metadict' is True."
            )

        input_path = meta_data[Key.FILENAME_OR_OBJ]
        output_path = meta_data[MetaKeys.SAVED_TO]
        log_data = {"input": input_path, "output": output_path}

        if has_filelock:
            with FileLock(str(self.mapping_file_path) + ".lock"):
                self._write_to_file(log_data)
        else:
            self._write_to_file(log_data)
        return img

    def _write_to_file(self, log_data):