code: def test_explicit_sample(mp_start_method):
      with pm.Model() as model:
          a = pm.Normal("a", shape=1)
          b = pm.HalfNormal("b")
          step1 = pm.NUTS([model.rvs_to_values[a]])
          step2 = pm.Metropolis([model.rvs_to_values[b]])
  
      step = pm.CompoundStep([step1, step2])
  
      # on Windows we cannot fork
      if platform.system() == "Windows" and mp_start_method == "fork":
          return
      if mp_start_method == "spawn":
          step_method_pickled = cloudpickle.dumps(step, protocol=-1)
      else:
          step_method_pickled = None
  
      ctx = multiprocessing.get_context(mp_start_method)
      proc = ps.ProcessAdapter(
          10,
          10,
          step,
          chain=3,
          seed=1,
          mp_ctx=ctx,
          start={
              "a": floatX(np.array([1.0])),
              "b_log__": floatX(np.array(2.0)),
          },
          step_method_pickled=step_method_pickled,
      )
      proc.start()
      while True:
          proc.write_next()
          out = ps.ProcessAdapter.recv_draw([proc])
          view = proc.shared_point_view
          for name in view:
              view[name].copy()
          if out[1]:
              break
      proc.join()
  
imports: import multiprocessing
  import os
  import platform
  import sys
  import warnings
  import cloudpickle
  import numpy
  import pytensor
  import pytensor.tensor
  import pytest
  from pytensor.compile.ops import as_op
  from pytensor.tensor.type import TensorType
  import pymc
  import pymc.sampling.parallel
  from pymc.pytensorf import floatX
label_desc: Use image node_id 6 for calling a global function with example usage: test_explicit_sample(mp_start_method) without return types