code: def apply_transform(
      self,
      data: CPMBeeInputType,
      transform: Union[
          Dict[str, Any],
          Callable[[CPMBeeInputType], CPMBeeInputType],
          None,
      ],
  ) -> CPMBeeInputType:
      if transform is None:
          return data
      if not isinstance(transform, dict):
          # transform function
          return transform(data)
  
      mapping_list: List[Tuple[str, str]] = []
  
      def _walk_transform_dict(
          data: Union[Dict[str, Any], str], prefix: str = ""
      ):
          if isinstance(data, dict):
              for k, v in data.items():
                  if len(prefix) > 0:
                      _walk_transform_dict(v, prefix + "." + k)
                  else:
                      _walk_transform_dict(v, k)
          else:
              assert isinstance(
                  data, str
              ), "Invalid transform {}".format(data)
              mapping_list.append((prefix, data))
  
      _walk_transform_dict(transform)
  
      expanded_mapping_list: List[Tuple[str, Any]] = []
  
      def _expand_mapping(
          data: CPMBeeInputType,
          stars: List[str],
          path: List[str],
          target: List[str],
      ):
          if len(path) == 0:
              num_stars = 0
              for it in target:
                  if it == "*":
                      num_stars += 1
              if num_stars != len(stars):
                  raise ValueError(
                      "Invalid transform {}".format(".".join(target))
                  )
  
              nw_tgt = []
              num_stars = 0
              for it in target:
                  if it == "*":
                      nw_tgt.append(stars[num_stars])
                      num_stars += 1
                  else:
                      nw_tgt.append(it)
              expanded_mapping_list.append((".".join(nw_tgt), data))
          else:
              if not isinstance(data, dict):
                  raise ValueError("Invalid data {}".format(data))
              if path[0] == "*":
                  for k, v in data.items():
                      _expand_mapping(v, stars + [k], path[1:], target)
              else:
                  _expand_mapping(
                      data[path[0]], stars, path[1:], target
                  )
  
      # expand mapping list
      for tgt, src in mapping_list:
          if src.startswith("$"):
              # copy from src
              _expand_mapping(
                  data, [], src[1:].split("."), tgt.split(".")
              )
          else:
              if "*" in tgt:
                  raise ValueError(
                      "Constant value is not allowed to have `*` in prefix"
                  )
              expanded_mapping_list.append((tgt, src))
  
      ret = {}
      for tgt, val in expanded_mapping_list:
          tgt = tgt.split(".")
          cur = ret
          while len(tgt) > 1:
              cur = cur[tgt[0]]
              tgt = tgt[1:]
          cur[tgt[0]] = val
      return ret
  
imports: from collections import OrderedDict
  import multiprocessing
  import os
  from queue import Empty
  from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
  from typing_extensions import TypedDict
  from ...dataset import DistributedDataset
  from ...tokenizers import CPMBeeTokenizer
  from ...utils.config import load_dataset_config
  import numpy
  import time
  from numpy.typing import NDArray
  import torch
  import bmtrain
  import importlib.machinery
  import importlib.util
  import types
  import random
label_desc: Use image node_id 3 for calling the _MixedDatasetBatchPacker obj's underlying member method code with example usage: obj.apply_transform(data, transform) and returns: ret, data, transform