def dict_to_example()

in tfx/components/example_gen/utils.py [0:0]

• 43 lines of code
• 15 McCabe index (conditional complexity)

def dict_to_example(instance: Dict[str, Any]) -> example_pb2.Example:
  """Converts dict to tf example."""
  feature = {}
  for key, value in instance.items():
    # TODO(jyzhao): support more types.
    # Convert to Python Scalars
    if isinstance(value, np.ndarray):
      pyval = value.tolist()
    else:
      try:
        pyval = value.item()
      except AttributeError:
        pyval = value

    # Convert bytes to str
    if isinstance(pyval, bytes):
      pyval = pyval.decode(_DEFAULT_ENCODING)

    if pyval is None:
      feature[key] = feature_pb2.Feature()
    elif isinstance(pyval, int):
      feature[key] = feature_pb2.Feature(
          int64_list=feature_pb2.Int64List(value=[pyval]))
    elif isinstance(pyval, float):
      feature[key] = feature_pb2.Feature(
          float_list=feature_pb2.FloatList(value=[pyval]))
    elif isinstance(pyval, str):
      feature[key] = feature_pb2.Feature(
          bytes_list=feature_pb2.BytesList(
              value=[pyval.encode(_DEFAULT_ENCODING)]))
    elif isinstance(pyval, list):
      if not pyval:
        feature[key] = feature_pb2.Feature()
      elif isinstance(pyval[0], int):
        feature[key] = feature_pb2.Feature(
            int64_list=feature_pb2.Int64List(value=pyval))
      elif isinstance(pyval[0], float):
        feature[key] = feature_pb2.Feature(
            float_list=feature_pb2.FloatList(value=pyval))
      elif isinstance(pyval[0], str):
        feature[key] = feature_pb2.Feature(
            bytes_list=feature_pb2.BytesList(
                value=[v.encode(_DEFAULT_ENCODING) for v in pyval]))
      else:
        raise RuntimeError('Column type `list of {}` is not supported.'.format(
            type(value[0])))
    else:
      raise RuntimeError('Column type {} is not supported.'.format(type(value)))
  return example_pb2.Example(features=feature_pb2.Features(feature=feature))