Add public accesor of measurements as dict for Collection.

PiperOrigin-RevId: 595487711

openhtf/core/measurements.py

@@ -59,14 +59,14 @@ def WidgetTestPhase(test):
 """
 
 import collections
+import copy
 import enum
 import functools
 import logging
 import typing
 from typing import Any, Callable, Dict, Iterator, List, Optional, Text, Tuple, Union
 
 import attr
-
 from openhtf import util
 from openhtf.util import data
 from openhtf.util import units as util_units
@@ -735,6 +735,42 @@ def to_dataframe(self, columns: Any = None) -> Any:
     return pandas.DataFrame.from_records(self.value, columns=columns)
 
 
+@attr.s(slots=True, frozen=True)
+class ImmutableMeasurement(object):
+  """Immutable copy of a measurement."""
+
+  name = attr.ib(type=Text)
+  value = attr.ib(type=Any)
+  units = attr.ib(type=Optional[util_units.UnitDescriptor])
+  dimensions = attr.ib(type=Optional[List[Dimension]])
+  outcome = attr.ib(type=Optional[Outcome])
+  docstring = attr.ib(type=Optional[Text], default=None)
+
+  @classmethod
+  def from_measurement(cls, measurement: Measurement) -> 'ImmutableMeasurement':
+    """Convert a Measurement into an ImmutableMeasurement."""
+    measured_value = measurement.measured_value
+    if isinstance(measured_value, DimensionedMeasuredValue):
+      value = data.attr_copy(
+          measured_value, value_dict=copy.deepcopy(measured_value.value_dict)
+      )
+    else:
+      value = (
+          copy.deepcopy(measured_value.value)
+          if measured_value.is_value_set
+          else None
+      )
+
+    return cls(
+        name=measurement.name,
+        value=value,
+        units=measurement.units,
+        dimensions=measurement.dimensions,
+        outcome=measurement.outcome,
+        docstring=measurement.docstring,
+    )
+
+
 @attr.s(slots=True)
 class Collection(object):
   """Encapsulates a collection of measurements.

openhtf/core/test_descriptor.py

@@ -569,7 +569,7 @@ def attach_from_file(
 
   def get_measurement(
       self,
-      measurement_name: Text) -> Optional[test_state.ImmutableMeasurement]:
+      measurement_name: Text) -> Optional[measurements.ImmutableMeasurement]:
     """Get a copy of a measurement value from current or previous phase.
 
     Measurement and phase name uniqueness is not enforced, so this method will
@@ -584,7 +584,7 @@ def get_measurement(
     return self._running_test_state.get_measurement(measurement_name)
 
   def get_measurement_strict(
-      self, measurement_name: Text) -> test_state.ImmutableMeasurement:
+      self, measurement_name: Text) -> measurements.ImmutableMeasurement:
     """Get a copy of the test measurement from current or previous phase.
 
     Measurement and phase name uniqueness is not enforced, so this method will

openhtf/core/test_state.py

@@ -33,10 +33,9 @@
 import os
 import socket
 import sys
-from typing import Any, Dict, Iterator, List, Optional, Set, Text, Tuple, TYPE_CHECKING, Union
+from typing import Any, Dict, Iterator, List, Optional, Set, TYPE_CHECKING, Text, Tuple, Union
 
 import attr
-
 import openhtf
 from openhtf import plugs
 from openhtf import util
@@ -48,7 +47,6 @@
 from openhtf.util import configuration
 from openhtf.util import data
 from openhtf.util import logs
-from openhtf.util import units
 from typing_extensions import Literal
 
 CONF = configuration.CONF
@@ -96,37 +94,6 @@ class InternalError(Exception):
   """An internal error."""
 
 
-@attr.s(slots=True, frozen=True)
-class ImmutableMeasurement(object):
-  """Immutable copy of a measurement."""
-
-  name = attr.ib(type=Text)
-  value = attr.ib(type=Any)
-  units = attr.ib(type=Optional[units.UnitDescriptor])
-  dimensions = attr.ib(type=Optional[List[measurements.Dimension]])
-  outcome = attr.ib(type=Optional[measurements.Outcome])
-
-  @classmethod
-  def from_measurement(
-      cls, measurement: measurements.Measurement) -> 'ImmutableMeasurement':
-    """Convert a Measurement into an ImmutableMeasurement."""
-    measured_value = measurement.measured_value
-    if isinstance(measured_value, measurements.DimensionedMeasuredValue):
-      value = data.attr_copy(
-          measured_value, value_dict=copy.deepcopy(measured_value.value_dict))
-    else:
-      value = (
-          copy.deepcopy(measured_value.value)
-          if measured_value.is_value_set else None)
-
-    return cls(
-        name=measurement.name,
-        value=value,
-        units=measurement.units,
-        dimensions=measurement.dimensions,
-        outcome=measurement.outcome)
-
-
 class TestState(util.SubscribableStateMixin):
   """This class handles tracking the state of a running Test.
 
@@ -263,8 +230,9 @@ def get_attachment(self,
     self.state_logger.warning('Could not find attachment: %s', attachment_name)
     return None
 
-  def get_measurement(self,
-                      measurement_name: Text) -> Optional[ImmutableMeasurement]:
+  def get_measurement(
+      self, measurement_name: Text
+  ) -> Optional[measurements.ImmutableMeasurement]:
     """Get a copy of a measurement value from current or previous phase.
 
     Measurement and phase name uniqueness is not enforced, so this method will
@@ -282,16 +250,17 @@ def get_measurement(self,
     # Check current running phase state
     if self.running_phase_state:
       if measurement_name in self.running_phase_state.measurements:
-        return ImmutableMeasurement.from_measurement(
-            self.running_phase_state.measurements[measurement_name])
+        return measurements.ImmutableMeasurement.from_measurement(
+            self.running_phase_state.measurements[measurement_name]
+        )
 
     # Iterate through phases in reversed order to return most recent (necessary
     # because measurement and phase names are not necessarily unique)
     for phase_record in reversed(self.test_record.phases):
       if (phase_record.result not in ignore_outcomes and
           measurement_name in phase_record.measurements):
         measurement = phase_record.measurements[measurement_name]
-        return ImmutableMeasurement.from_measurement(measurement)
+        return measurements.ImmutableMeasurement.from_measurement(measurement)
 
     self.state_logger.warning('Could not find measurement: %s',
                               measurement_name)