class_docstr: Control a Chain process from the main thread.
class_docstr_tok: ["Control","a","Chain","process","from","the","main","thread","."]
code: def __init__(
      self,
      draws: int,
      tune: int,
      step_method,
      step_method_pickled,
      chain: int,
      seed,
      start: Dict[str, np.ndarray],
      mp_ctx,
  ):
      self.chain = chain
      process_name = "worker_chain_%s" % chain
      self._msg_pipe, remote_conn = multiprocessing.Pipe()
  
      self._shared_point = {}
      self._point = {}
  
      for name, shape, dtype in DictToArrayBijection.map(
          start
      ).point_map_info:
          size = 1
          for dim in shape:
              size *= int(dim)
          size *= dtype.itemsize
          if size != ctypes.c_size_t(size).value:
              raise ValueError("Variable %s is too large" % name)
  
          array = mp_ctx.RawArray("c", size)
          self._shared_point[name] = (array, shape, dtype)
          array_np = np.frombuffer(array, dtype).reshape(shape)
          array_np[...] = start[name]
          self._point[name] = array_np
  
      self._readable = True
      self._num_samples = 0
  
      if step_method_pickled is not None:
          step_method_send = step_method_pickled
      else:
          if mp_ctx.get_start_method() == "spawn":
              raise ValueError(
                  "please provide a pre-pickled step method when multiprocessing start method is 'spawn'"
              )
          step_method_send = step_method
  
      self._process = mp_ctx.Process(
          daemon=True,
          name=process_name,
          target=_run_process,
          args=(
              process_name,
              remote_conn,
              step_method_send,
              step_method_pickled is not None,
              self._shared_point,
              draws,
              tune,
              seed,
          ),
      )
      self._process.start()
      # Close the remote pipe, so that we get notified if the other
      # end is closed.
      remote_conn.close()
  
imports: import ctypes
  import logging
  import multiprocessing
  import multiprocessing.sharedctypes
  import platform
  import time
  import traceback
  from collections import namedtuple
  from typing import Dict, List, Sequence
  import cloudpickle
  import numpy
  from fastprogress.fastprogress import progress_bar
  from pymc.blocking import DictToArrayBijection
  from pymc.exceptions import SamplingError
  from pymc.util import RandomSeed
label_desc: Use image node_id 1 to create a new ProcessAdapter object  with example: obj = ProcessAdapter(draws, tune, step_method, step_method_pickled, chain, seed, start, mp_ctx)