a
    d'                     @   st   d Z ddlmZmZmZ ddlmZ ddlZddl	m
Z
 ddlmZ ddlmZ G dd	 d	eZG d
d de
ZdS )zG
BasePredictionWriter
====================

Aids in saving predictions
    )AnyOptionalSequence)LiteralN)Callback)LightningEnum)MisconfigurationExceptionc                   @   s<   e Zd ZdZdZdZeedddZeedddZ	d	S )
WriteIntervalbatchepochbatch_and_epoch)returnc                 C   s   | | j | jfv S N)BATCHBATCH_AND_EPOCHself r   v/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/callbacks/prediction_writer.pyon_batch#   s    zWriteInterval.on_batchc                 C   s   | | j | jfv S r   )EPOCHr   r   r   r   r   on_epoch'   s    zWriteInterval.on_epochN)
__name__
__module____qualname__r   r   r   propertyboolr   r   r   r   r   r   r	      s   r	   c                	   @   s   e Zd ZdZded ddddZdd	eeee	  ee	e	dd
ddZ
dd	ee eee  ddddZdd	eee	e	ddddZdd	ee ddddZdS )BasePredictionWritera	  Base class to implement how the predictions should be stored.

    Args:
        write_interval: When to write.

    Example::

        import torch
        from pytorch_lightning.callbacks import BasePredictionWriter

        class CustomWriter(BasePredictionWriter):

            def __init__(self, output_dir, write_interval):
                super().__init__(write_interval)
                self.output_dir = output_dir

            def write_on_batch_end(
                self, trainer, pl_module', prediction, batch_indices, batch, batch_idx, dataloader_idx
            ):
                torch.save(prediction, os.path.join(self.output_dir, dataloader_idx, f"{batch_idx}.pt"))

            def write_on_epoch_end(self, trainer, pl_module, predictions, batch_indices):
                torch.save(predictions, os.path.join(self.output_dir, "predictions.pt"))


        pred_writer = CustomWriter(output_dir="pred_path", write_interval="epoch")
        trainer = Trainer(callbacks=[pred_writer])
        model = BoringModel()
        trainer.predict(model, return_predictions=False)

    Example::

        # multi-device inference example

        import torch
        from pytorch_lightning.callbacks import BasePredictionWriter

        class CustomWriter(BasePredictionWriter):

            def __init__(self, output_dir, write_interval):
                super().__init__(write_interval)
                self.output_dir = output_dir

            def write_on_epoch_end(self, trainer, pl_module, predictions, batch_indices):
                # this will create N (num processes) files in `output_dir` each containing
                # the predictions of it's respective rank
                torch.save(predictions, os.path.join(self.output_dir, f"predictions_{trainer.global_rank}.pt"))

                # optionally, you can also save `batch_indices` to get the information about the data index
                # from your prediction data
                torch.save(batch_indices, os.path.join(self.output_dir, f"batch_indices_{trainer.global_rank}.pt"))


        # or you can set `writer_interval="batch"` and override `write_on_batch_end` to save
        # predictions at batch level
        pred_writer = CustomWriter(output_dir="pred_path", write_interval="epoch")
        trainer = Trainer(accelerator="gpu", strategy="ddp", devices=8, callbacks=[pred_writer])
        model = BoringModel()
        trainer.predict(model, return_predictions=False)
    r
   )r
   r   r   N)write_intervalr   c                 C   s4   |t tvr&tddd tD  dt|| _d S )Nz"`write_interval` should be one of c                 S   s   g | ]
}|j qS r   )value).0ir   r   r   
<listcomp>l       z1BasePredictionWriter.__init__.<locals>.<listcomp>.)listr	   r   interval)r   r   r   r   r   __init__j   s    zBasePredictionWriter.__init__z
pl.Trainerzpl.LightningModule)trainer	pl_module
predictionbatch_indicesr
   	batch_idxdataloader_idxr   c                 C   s
   t  dS )z0Override with the logic to write a single batch.NNotImplementedError)r   r(   r)   r*   r+   r
   r,   r-   r   r   r   write_on_batch_endo   s    z'BasePredictionWriter.write_on_batch_end)r(   r)   predictionsr+   r   c                 C   s
   t  dS )z-Override with the logic to write all batches.Nr.   )r   r(   r)   r1   r+   r   r   r   write_on_epoch_end|   s    z'BasePredictionWriter.write_on_epoch_end)r(   r)   outputsr
   r,   r-   r   c              	   C   s0   | j jsd S |jjj}| ||||||| d S r   )r&   r   predict_loopZ
epoch_loopZcurrent_batch_indicesr0   )r   r(   r)   r3   r
   r,   r-   r+   r   r   r   on_predict_batch_end   s    	
z)BasePredictionWriter.on_predict_batch_end)r(   r)   r3   r   c                 C   s,   | j jsd S |jj}| |||jj| d S r   )r&   r   r4   epoch_batch_indicesr2   r1   )r   r(   r)   r3   r6   r   r   r   on_predict_epoch_end   s    z)BasePredictionWriter.on_predict_epoch_end)r
   )r   r   r   __doc__r   r'   r   r   r   intr0   r2   r5   r7   r   r   r   r   r   ,   s8   =


r   )r8   typingr   r   r   Ztyping_extensionsr   Zpytorch_lightningplZ$pytorch_lightning.callbacks.callbackr   Zpytorch_lightning.utilitiesr   Z&pytorch_lightning.utilities.exceptionsr   r	   r   r   r   r   r   <module>   s   