Add public accessor of measurements through ImmutableMeasurements in …

…Collection.

PiperOrigin-RevId: 595488744

openhtf/core/measurements.py

@@ -59,14 +59,15 @@ 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
+from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional, Text, Tuple, Union
 
 import attr
-
+import immutabledict
 from openhtf import util
 from openhtf.util import data
 from openhtf.util import units as util_units
@@ -735,6 +736,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.
@@ -820,6 +857,12 @@ def __getitem__(self, name: Text) -> Any:
     # Return the MeasuredValue's value, MeasuredValue will raise if not set.
     return m.measured_value.value
 
+  def immutable_measurements(self) -> Mapping[Text, ImmutableMeasurement]:
+    return immutabledict.immutabledict({
+        name: ImmutableMeasurement.from_measurement(meas)
+        for name, meas in self._measurements.items()
+    })
+
 
 # Work around for attrs bug in 20.1.0; after the next release, this can be
 # removed and `Collection._custom_setattr` can be renamed to `__setattr__`.

openhtf/core/test_descriptor.py

@@ -34,18 +34,16 @@
 
 import attr
 import colorama
-
 from openhtf import util
 from openhtf.core import base_plugs
 from openhtf.core import diagnoses_lib
-from openhtf.core import measurements
+from openhtf.core import measurements as htf_measurements
 from openhtf.core import phase_collections
 from openhtf.core import phase_descriptor
 from openhtf.core import phase_executor
 from openhtf.core import test_executor
 from openhtf.core import test_record as htf_test_record
 from openhtf.core import test_state
-
 from openhtf.util import configuration
 from openhtf.util import console_output
 from openhtf.util import logs
@@ -462,10 +460,10 @@ class TestApi(object):
       stdout (configurable) and the frontend via the Station API, if it's
       enabled, in addition to the 'log_records' attribute of the final
       TestRecord output by the running test.
-    measurements: A measurements.Collection object used to get/set measurement
-      values.  See util/measurements.py for more implementation details, but in
-      the simple case, set measurements directly as attributes on this object
-      (see examples/measurements.py for examples).
+    measurements: A htf_measurements.Collection object used to get/set
+      measurement values.  See util/measurements.py for more implementation
+      details, but in the simple case, set measurements directly as attributes
+      on this object (see examples/measurements.py for examples).
     attachments: Dict mapping attachment name to test_record.Attachment instance
       containing the data that was attached (and the MIME type that was assumed
       based on extension, if any).  Only attachments that have been attached in
@@ -486,7 +484,7 @@ class TestApi(object):
           https://github.com/google/openhtf/issues/new
   """
 
-  measurements = attr.ib(type=measurements.Collection)
+  measurements = attr.ib(type=htf_measurements.Collection)
 
   # Internal state objects.  If you find yourself needing to use these, please
   # use required_state=True for the phase to use the test_state object instead.
@@ -568,8 +566,8 @@ def attach_from_file(
         filename, name=name, mimetype=mimetype)
 
   def get_measurement(
-      self,
-      measurement_name: Text) -> Optional[test_state.ImmutableMeasurement]:
+      self, measurement_name: Text
+  ) -> Optional[htf_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 +582,8 @@ 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
+  ) -> htf_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)