Class WriteFileMapping

monai/transforms/io/array.py:527–571

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 mappin

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[ImageMetaKey.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):
        try:
            with self.mapping_file_path.open("r") as f:
                existing_log_data = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            existing_log_data = []
        existing_log_data.append(log_data)
        with self.mapping_file_path.open("w") as f:
            json.dump(existing_log_data, f, indent=4)