code: def test_model_can_use_observation_history(self):
      observations = np.array(
          [0.1, 3.0, -0.7, 1.1, 0.0, 14.0, -3.0, 5.8]
      ).astype(self.dtype)
      weights = np.array([0.1, -0.2, 0.7]).astype(self.dtype)
  
      # Define an autoregressive model on observations. This ignores the
      # state entirely; it depends only on previous observations.
      initial_state_prior = jdn.JointDistributionNamed(
          {"dummy_state": deterministic.Deterministic(0.0)}
      )
  
      def dummy_transition_fn(_, state, **kwargs):
          del kwargs
          return jdn.JointDistributionNamed(
              tf.nest.map_structure(deterministic.Deterministic, state)
          )
  
      @particle_filter_augmentation.augment_with_observation_history(
          observations=observations, history_size=len(weights)
      )
      def autoregressive_observation_fn(
          step, _, observation_history=None
      ):
          num_terms = prefer_static.minimum(step, len(weights))
          usable_weights = tf.convert_to_tensor(weights)[
              len(weights) - num_terms :
          ]
          loc = tf.reduce_sum(usable_weights * observation_history)
          return normal.Normal(loc, 1.0)
  
      # Manually compute the conditional log-probs of a series of observations
      # under the autoregressive model.
      expected_locs = []
      for current_step in range(len(observations)):
          start_step = max(0, current_step - len(weights))
          context_length = current_step - start_step
          expected_locs.append(
              np.sum(
                  observations[start_step:current_step]
                  * weights[len(weights) - context_length :]
              )
          )
      expected_lps = self.evaluate(
          normal.Normal(expected_locs, scale=1.0).log_prob(observations)
      )
  
      # Check that the particle filter gives the same log-probs.
      _, _, _, lps = self.evaluate(
          particle_filter.particle_filter(
              observations,
              initial_state_prior=initial_state_prior,
              transition_fn=dummy_transition_fn,
              observation_fn=autoregressive_observation_fn,
              num_particles=2,
              seed=test_util.test_seed(),
          )
      )
      self.assertAllClose(expected_lps, lps)
  
file_path: probability/tensorflow_probability/python/experimental/mcmc/particle_filter_augmentation_test.py
imports: import numpy
  import tensorflow.compat.v2
  from tensorflow_probability.python.distributions import deterministic
  from tensorflow_probability.python.distributions import joint_distribution_named
  from tensorflow_probability.python.distributions import normal
  from tensorflow_probability.python.distributions import poisson
  from tensorflow_probability.python.experimental.mcmc import particle_filter
  from tensorflow_probability.python.experimental.mcmc import particle_filter_augmentation
  from tensorflow_probability.python.internal import prefer_static
  from tensorflow_probability.python.internal import test_util
label_desc: Use image node_id 3 for calling the _ParticleFilterUtilTest obj's underlying member method code with example usage: obj.test_model_can_use_observation_history() and returns: jdn, normal