code: def test_particle_filtering_tractography(rng):
      """This tests that the ParticleFilteringTracking produces
      more streamlines connecting the gray matter than LocalTracking.
      """
      sphere = get_sphere("repulsion100")
      step_size = 0.2
  
      # Simple tissue masks
      simple_wm = np.array(
          [
              [0, 0, 0, 0, 0, 0],
              [0, 0, 1, 0, 0, 0],
              [0, 1, 1, 1, 0, 0],
              [0, 1, 1, 1, 0, 0],
              [0, 0, 0, 0, 0, 0],
          ]
      )
      simple_wm = np.dstack(
          [
              np.zeros(simple_wm.shape),
              simple_wm,
              simple_wm,
              simple_wm,
              np.zeros(simple_wm.shape),
          ]
      )
      simple_gm = np.array(
          [
              [1, 1, 0, 0, 0, 0],
              [1, 1, 0, 0, 0, 0],
              [0, 1, 0, 0, 1, 0],
              [0, 0, 0, 0, 1, 0],
              [0, 0, 0, 0, 0, 0],
          ]
      )
      simple_gm = np.dstack(
          [
              np.zeros(simple_gm.shape),
              simple_gm,
              simple_gm,
              simple_gm,
              np.zeros(simple_gm.shape),
          ]
      )
      simple_csf = np.ones(simple_wm.shape) - simple_wm - simple_gm
  
      sc = ActStoppingCriterion.from_pve(
          simple_wm, simple_gm, simple_csf
      )
      seeds = seeds_from_mask(simple_wm, np.eye(4), density=2)
  
      # Random pmf in every voxel
      shape_img = list(simple_wm.shape)
      shape_img.extend([sphere.vertices.shape[0]])
      pmf = rng.random(shape_img)
  
      # Test that PFT recover equal or more streamlines than localTracking
      dg = ProbabilisticDirectionGetter.from_pmf(pmf, 60, sphere)
      local_streamlines_generator = LocalTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size,
          max_cross=1,
          return_all=False,
      )
      local_streamlines = Streamlines(local_streamlines_generator)
  
      pft_streamlines_generator = ParticleFilteringTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size,
          max_cross=1,
          return_all=False,
          pft_back_tracking_dist=1,
          pft_front_tracking_dist=0.5,
      )
      pft_streamlines = Streamlines(pft_streamlines_generator)
  
      npt.assert_(np.array([len(pft_streamlines) > 0]))
      npt.assert_(
          np.array([len(pft_streamlines) >= len(local_streamlines)])
      )
  
      # Test PFT with a PTT direction getter
      dg_ptt = PTTDirectionGetter.from_pmf(pmf, 60, sphere)
      pft_ptt_streamlines = Streamlines(
          ParticleFilteringTracking(
              dg_ptt,
              sc,
              seeds,
              np.eye(4),
              step_size,
              max_cross=1,
              return_all=False,
              pft_back_tracking_dist=1,
              pft_front_tracking_dist=0.5,
          )
      )
  
      npt.assert_(np.array([len(pft_ptt_streamlines) > 0]))
      npt.assert_(
          np.array([len(pft_ptt_streamlines) >= len(local_streamlines)])
      )
  
      # Test that all points are equally spaced
      for l in [2, 3, 5, 10, 100]:
          pft_streamlines = ParticleFilteringTracking(
              dg,
              sc,
              seeds,
              np.eye(4),
              step_size,
              max_cross=1,
              return_all=True,
              maxlen=l,
          )
          for s in pft_streamlines:
              for i in range(len(s) - 1):
                  npt.assert_almost_equal(
                      np.linalg.norm(s[i] - s[i + 1]), step_size
                  )
  
      # Test that all points are within the image volume
      seeds = seeds_from_mask(
          np.ones(simple_wm.shape), np.eye(4), density=1
      )
      pft_streamlines_generator = ParticleFilteringTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size,
          max_cross=1,
          return_all=True,
      )
      pft_streamlines = Streamlines(pft_streamlines_generator)
  
      for s in pft_streamlines:
          npt.assert_(np.all((s + 0.5).astype(int) >= 0))
          npt.assert_(np.all((s + 0.5).astype(int) < simple_wm.shape))
  
      # Test that the number of streamline return with return_all=True equal the
      # number of seeds places
      npt.assert_(np.array([len(pft_streamlines) == len(seeds)]))
  
      # Test min and max length
      pft_streamlines_generator = ParticleFilteringTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size,
          maxlen=20,
          minlen=3,
          return_all=False,
      )
      pft_streamlines = Streamlines(pft_streamlines_generator)
  
      for s in pft_streamlines:
          npt.assert_(len(s) >= 3)
          npt.assert_(len(s) <= 20)
  
      # Test non WM seed position
      seeds = [[0, 5, 4], [0, 0, 1], [50, 50, 50]]
      pft_streamlines_generator = ParticleFilteringTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size,
          max_cross=1,
          return_all=True,
      )
      pft_streamlines = Streamlines(pft_streamlines_generator)
  
      npt.assert_equal(len(pft_streamlines[0]), 3)  # INVALIDPOINT
      npt.assert_equal(len(pft_streamlines[1]), 3)  # ENDPOINT
      npt.assert_equal(len(pft_streamlines[2]), 1)  # OUTSIDEIMAGE
  
      # Test with wrong StoppingCriterion type
      sc_bin = BinaryStoppingCriterion(simple_wm)
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg, sc_bin, seeds, np.eye(4), step_size
          ),
      )
      # Test with invalid back/front tracking distances
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg,
              sc,
              seeds,
              np.eye(4),
              step_size,
              pft_back_tracking_dist=0,
              pft_front_tracking_dist=0,
          ),
      )
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg,
              sc,
              seeds,
              np.eye(4),
              step_size,
              pft_back_tracking_dist=-1,
          ),
      )
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg,
              sc,
              seeds,
              np.eye(4),
              step_size,
              pft_back_tracking_dist=0,
              pft_front_tracking_dist=-2,
          ),
      )
  
      # Test with invalid affine shape
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg, sc, seeds, np.eye(3), step_size
          ),
      )
  
      # Test with invalid maxlen
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg, sc, seeds, np.eye(4), step_size, maxlen=0
          ),
      )
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg, sc, seeds, np.eye(4), step_size, maxlen=-1
          ),
      )
  
      # Test with invalid particle count
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg, sc, seeds, np.eye(4), step_size, particle_count=0
          ),
      )
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg, sc, seeds, np.eye(4), step_size, particle_count=-1
          ),
      )
  
      # Test reproducibility
      tracking1 = Streamlines(
          ParticleFilteringTracking(
              dg, sc, seeds, np.eye(4), step_size, random_seed=0
          )
      )._data
      tracking2 = Streamlines(
          ParticleFilteringTracking(
              dg, sc, seeds, np.eye(4), step_size, random_seed=0
          )
      )._data
      npt.assert_equal(tracking1, tracking2)
  
      # Test min_wm_pve_before_stopping parameter
      expected = [
          np.array([[1.0, 0.0, 1.0], [1.0, 1.0, 1.0], [1.0, 2.0, 1.0]]),
          np.array(
              [
                  [1.0, 0.0, 1.0],
                  [1.0, 1.0, 1.0],
                  [1.0, 2.0, 1.0],
                  [1.0, 3.0, 1.0],
                  [1.0, 4.0, 1.0],
              ]
          ),
      ]
  
      simple_wm = np.array(
          [
              [0, 0, 0, 0, 0, 0],
              [0, 0.4, 0.4, 1, 0.4, 0],
              [0, 0, 0, 0, 0, 0],
          ]
      )
      simple_wm = np.dstack(
          [
              np.zeros(simple_wm.shape),
              simple_wm,
              simple_wm,
              simple_wm,
              np.zeros(simple_wm.shape),
          ]
      )
      simple_gm = np.array(
          [
              [0, 0, 0, 0, 0, 0],
              [1, 0.6, 0.6, 0, 0.6, 1],
              [0, 0, 0, 0, 0, 0],
          ]
      )
      simple_gm = np.dstack(
          [
              np.zeros(simple_gm.shape),
              simple_gm,
              simple_gm,
              simple_gm,
              np.zeros(simple_gm.shape),
          ]
      )
      simple_csf = np.ones(simple_wm.shape) - simple_wm - simple_gm
      sc = ActStoppingCriterion.from_pve(
          simple_wm, simple_gm, simple_csf
      )
      seeds = np.array([[1, 1, 1]])
      sphere = HemiSphere.from_sphere(unit_octahedron)
      pmf = np.zeros(list(simple_gm.shape) + [3])
      pmf[:, :, :, 1] = 1  # horizontal bundle
  
      dg = ProbabilisticDirectionGetter.from_pmf(pmf, 30, sphere)
  
      pft_streamlines_generator = ParticleFilteringTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size=1,
          max_cross=1,
          return_all=True,
          min_wm_pve_before_stopping=0,
      )
      pft_streamlines = Streamlines(pft_streamlines_generator)
      npt.assert_(np.allclose(pft_streamlines[0], expected[0]))
  
      pft_streamlines_generator = ParticleFilteringTracking(
          dg,
          sc,
          seeds,
          np.eye(4),
          step_size=1,
          max_cross=1,
          return_all=True,
          min_wm_pve_before_stopping=1,
      )
      pft_streamlines = Streamlines(pft_streamlines_generator)
      npt.assert_(np.allclose(pft_streamlines[0], expected[1]))
  
      # Test invalid min_wm_pve_before_stopping parameters
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg,
              sc,
              seeds,
              np.eye(4),
              step_size,
              min_wm_pve_before_stopping=-1,
          ),
      )
      npt.assert_raises(
          ValueError,
          lambda: ParticleFilteringTracking(
              dg,
              sc,
              seeds,
              np.eye(4),
              step_size,
              min_wm_pve_before_stopping=2,
          ),
      )
  
imports: from logging import warn
  import warnings
  import nibabel
  import numpy
  import numpy.testing
  from dipy.core.gradients import gradient_table
  from dipy.core.sphere import HemiSphere, unit_octahedron
  from dipy.data import get_fnames, get_sphere
  from dipy.direction import BootDirectionGetter, ClosestPeakDirectionGetter, DeterministicMaximumDirectionGetter, PeaksAndMetrics, ProbabilisticDirectionGetter, PTTDirectionGetter
  from dipy.reconst.csdeconv import ConstrainedSphericalDeconvModel
  from dipy.reconst.shm import descoteaux07_legacy_msg
  from dipy.tracking.local_tracking import LocalTracking, ParticleFilteringTracking
  from dipy.tracking.streamline import Streamlines
  from dipy.tracking.stopping_criterion import ActStoppingCriterion, BinaryStoppingCriterion, ThresholdStoppingCriterion, StreamlineStatus
  from dipy.tracking.utils import random_seeds_from_mask, seeds_from_mask
  from dipy.sims.voxel import single_tensor, multi_tensor
  from dipy.testing.decorators import set_random_number_generator
label_desc: Use image node_id 6 for calling a global function with example usage: test_particle_filtering_tractography(rng) without return types