class_docstr: Merge repeated measurements for to obtain one observation per arm.
  
  Repeated measurements are merged via inverse variance weighting (e.g. over
  different trials). This intentionally ignores the trial index and assumes
  stationarity.
  
  TODO: Support inverse variance weighting correlated outcomes (full covariance).
  
  Note: this is not reversible.
class_docstr_tok: ["Merge","repeated","measurements","for","to","obtain","one","observation","per","arm",".","Repeated","measurements","are","merged","via","inverse","variance","weighting","(","e.g",".","over","different","trials",")",".","This","intentionally","ignores","the","trial","index","and","assumes","stationarity",".","TODO",":","Support","inverse","variance","weighting","correlated","outcomes","(","full","covariance",")",".","Note",":","this","is","not","reversible","."]
code: def transform_observations(
      self,
      observations: List[Observation],
  ) -> List[Observation]:
      # Transform observations
      new_observations = []
      observation_features, observation_data = separate_observations(
          observations
      )
      for j, obsd in enumerate(observation_data):
          key = Arm.md5hash(observation_features[j].parameters)
          # pop to ensure that the resulting observations list has one
          # observation per unique arm
          metric_dict = self.arm_to_merged.pop(key, None)
          if metric_dict is None:
              continue
          merged_means = np.zeros(len(obsd.metric_names))
          merged_covariance = np.zeros(
              (len(obsd.metric_names), len(obsd.metric_names))
          )
          for i, m in enumerate(obsd.metric_names):
              merged_metric = metric_dict[m]
              merged_means[i] = merged_metric["mean"]
              merged_covariance[i, i] = merged_metric["var"]
          new_obsd = ObservationData(
              metric_names=obsd.metric_names,
              means=merged_means,
              covariance=merged_covariance,
          )
          new_obs = Observation(
              features=observation_features[j], data=new_obsd
          )
          new_observations.append(new_obs)
      return new_observations
  
file_path: Ax/ax/modelbridge/transforms/merge_repeated_measurements.py
imports: from __future__ import annotations
  from collections import defaultdict
  from typing import DefaultDict, Dict, List, Optional
  import numpy
  from ax.core.arm import Arm
  from ax.core.observation import Observation, ObservationData, separate_observations
  from ax.core.search_space import SearchSpace
  from ax.modelbridge.base import ModelBridge
  from ax.modelbridge.transforms.base import Transform
  from ax.models.types import TConfig
label_desc: Use image node_id 2 for calling the MergeRepeatedMeasurements obj's underlying member method code with example usage: obj.transform_observations(observations) and returns: new_observations