code: def __init__(
      self,
      network_config,
      denoiser_config,
      first_stage_config,
      conditioner_config: Union[
          None, Dict, ListConfig, OmegaConf
      ] = None,
      sampler_config: Union[None, Dict, ListConfig, OmegaConf] = None,
      optimizer_config: Union[None, Dict, ListConfig, OmegaConf] = None,
      scheduler_config: Union[None, Dict, ListConfig, OmegaConf] = None,
      loss_fn_config: Union[None, Dict, ListConfig, OmegaConf] = None,
      network_wrapper: Union[None, str] = None,
      ckpt_path: Union[None, str] = None,
      use_ema: bool = False,
      ema_decay_rate: float = 0.9999,
      scale_factor: float = 1.0,
      disable_first_stage_autocast=False,
      input_key: str = "jpg",
      log_keys: Union[List, None] = None,
      no_cond_log: bool = False,
      compile_model: bool = False,
      en_and_decode_n_samples_a_time: Optional[int] = None,
  ):
      super().__init__()
      self.log_keys = log_keys
      self.input_key = input_key
      self.optimizer_config = default(
          optimizer_config, {"target": "torch.optim.AdamW"}
      )
      model = instantiate_from_config(network_config)
      self.model = get_obj_from_str(
          default(network_wrapper, OPENAIUNETWRAPPER)
      )(model, compile_model=compile_model)
  
      self.denoiser = instantiate_from_config(denoiser_config)
      self.sampler = (
          instantiate_from_config(sampler_config)
          if sampler_config is not None
          else None
      )
      self.conditioner = instantiate_from_config(
          default(conditioner_config, UNCONDITIONAL_CONFIG)
      )
      self.scheduler_config = scheduler_config
      self._init_first_stage(first_stage_config)
  
      self.loss_fn = (
          instantiate_from_config(loss_fn_config)
          if loss_fn_config is not None
          else None
      )
  
      self.use_ema = use_ema
      if self.use_ema:
          self.model_ema = LitEma(self.model, decay=ema_decay_rate)
          print(
              f"Keeping EMAs of {len(list(self.model_ema.buffers()))}."
          )
  
      self.scale_factor = scale_factor
      self.disable_first_stage_autocast = disable_first_stage_autocast
      self.no_cond_log = no_cond_log
  
      if ckpt_path is not None:
          self.init_from_ckpt(ckpt_path)
  
      self.en_and_decode_n_samples_a_time = (
          en_and_decode_n_samples_a_time
      )
  
imports: import math
  from contextlib import contextmanager
  from typing import Any, Dict, List, Optional, Tuple, Union
  import pytorch_lightning
  import torch
  from omegaconf import ListConfig, OmegaConf
  from safetensors.torch import load_file
  from torch.optim.lr_scheduler import LambdaLR
  from ..modules import UNCONDITIONAL_CONFIG
  from ..modules.autoencoding.temporal_ae import VideoDecoder
  from ..modules.diffusionmodules.wrappers import OPENAIUNETWRAPPER
  from ..modules.ema import LitEma
  from ..util import default, disabled_train, get_obj_from_str, instantiate_from_config, log_txt_as_img
label_desc: Use image node_id 1 to create a new DiffusionEngine object from inherited base classes: pl with example: obj = DiffusionEngine(network_config, denoiser_config, first_stage_config, conditioner_config, sampler_config, optimizer_config, scheduler_config, loss_fn_config, network_wrapper, ckpt_path, use_ema, ema_decay_rate, scale_factor, disable_first_stage_autocast, input_key, log_keys, no_cond_log, compile_model, en_and_decode_n_samples_a_time)