<commit_before>"""Output or upload a TestRun proto for mfg-inspector.com"""

import json
import logging
import numbers
import zlib
import httplib2

from openhtf.io.output import json_factory
from openhtf.io.proto import testrun_pb2
from openhtf.io.proto import units_pb2

from openhtf import util
from openhtf.exe import test_state
from openhtf.util import validators

# pylint: disable=no-member
MIMETYPE_MAP = {
  'image/jpeg': testrun_pb2.InformationParameter.JPG,
  'image/png': testrun_pb2.InformationParameter.PNG,
  'audio/x-wav': testrun_pb2.InformationParameter.WAV,
  'text/plain': testrun_pb2.InformationParameter.TEXT_UTF8,
  'image/tiff': testrun_pb2.InformationParameter.TIFF,
  'video/mp4': testrun_pb2.InformationParameter.MP4,
}
OUTCOME_MAP = {
  'State.ERROR': testrun_pb2.ERROR,
  'State.TIMEOUT': testrun_pb2.ERROR,
  'State.ABORTED': testrun_pb2.ERROR,
  'State.FAIL': testrun_pb2.FAIL,
  'State.PASS': testrun_pb2.PASS,
}
UOM_CODE_MAP = {
  u.GetOptions().Extensions[units_pb2.uom_code]: num
  for num, u in units_pb2.Units.UnitCode.DESCRIPTOR.values_by_number.iteritems()
}
# pylint: enable=no-member

# Control how many flattened parameters we'll output per multidimensional
# measurement.
MAX_PARAMS_PER_MEASUREMENT = 100


class UploadFailedError(Exception):
  """Raised when an upload to mfg-inspector fails."""


# pylint: disable=invalid-name
def _PopulateHeader(record, testrun):
  """Populate header-like info in testrun from record.

  Mostly obvious, some stuff comes from metadata, see docstring of
  TestRunFromTestRecord for details.
  """
  testrun.dut_serial = record.dut_id
  testrun.tester_name = record.station_id
  if 'test_name' in record.metadata:
    testrun.test_info.name = record.metadata['test_name']
  else:
    # Default to copying tester_name into test_info.name.
    testrun.test_info.name = record.station_id
  if 'test_description' in record.metadata:
    testrun.test_info.description = record.metadata['test_description']
  if 'test_version' in record.metadata:
    testrun.test_info.version_string = record.metadata['test_version']
  testrun.test_status = OUTCOME_MAP.get(record.outcome, testrun_pb2.ERROR)
  testrun.start_time_millis = record.start_time_millis
  testrun.end_time_millis = record.end_time_millis
  if 'run_name' in record.metadata:
    testrun.run_name = record.metadata['run_name']
  for details in record.outcome_details:
    testrun_code = testrun.failure_codes.add()
    if details.code_type == test_state.TestState.State.ERROR:
      testrun_code.code = details.code
      testrun_code.details = details.description
    else:
      testrun_code.code = details.code_type
      testrun_code.details = '%s: %s' % (details.code, details.description)


def _AttachJson(record, testrun):
  """Attach a copy of the JSON-ified record as an info parameter.

  Save a copy of the JSON-ified record in an attachment so we can access
  un-mangled fields later if we want.  Remove attachments since those get
  copied over and can potentially be quite large.
  """
  record_dict = util.convert_to_dict(record, ignore_keys=('attachments',))
  record_json = json_factory.OutputToJSON(sort_keys=True).encode(record_dict)
  testrun_param = testrun.info_parameters.add()
  testrun_param.name = 'OpenHTF_record.json'
  testrun_param.value_binary = record_json
  # pylint: disable=no-member
  testrun_param.type = testrun_pb2.InformationParameter.TEXT_UTF8
  # pylint: enable=no-member


def _ExtractAttachments(phase, testrun, used_parameter_names):
  """Extract attachments, just copy them over."""
  for name, (data, mimetype) in sorted(phase.attachments.items()):
    while name in used_parameter_names:
      name += '_'  # Hack to avoid collisions between phases.
    used_parameter_names.add(name)
    testrun_param = testrun.info_parameters.add()
    testrun_param.name = name
    testrun_param.value_binary = data
    if mimetype in MIMETYPE_MAP:
      testrun_param.type = MIMETYPE_MAP[mimetype]
    else:
      # pylint: disable=no-member
      testrun_param.type = testrun_pb2.InformationParameter.BINARY
      # pylint: enable=no-member


def _MangleMeasurement(name, value, measurement, mangled_parameters):
  """Flatten parameters for backwards compatibility, watch for collisions.

  We generate these by doing some name mangling, using some sane limits for
  very large multidimensional measurements.
  """
  for current_value in value[:MAX_PARAMS_PER_MEASUREMENT]:
    # Mangle names so they look like 'myparameter_Xsec_Ynm_ZHz'
    mangled_name = '_'.join([name] + [
        '%s%s' % (
          dim_val,
          dim_units.uom_suffix if dim_units.uom_suffix else '') for
        dim_val, dim_units in zip(
          current_value[:-1], measurement.dimensions)])
    while mangled_name in mangled_parameters:
      logging.warning('Mangled name %s already in use', mangled_name)
      mangled_name += '_'
    mangled_param = testrun_pb2.TestParameter()
    mangled_param.name = mangled_name
    mangled_param.numeric_value = float(current_value[-1])
    if measurement.units:
      mangled_param.unit_code = UOM_CODE_MAP[measurement.units.uom_code]
    mangled_param.description = (
        'Mangled parameter from measurement %s with dimensions %s' % (
        name, tuple(d.uom_suffix for d in measurement.dimensions)))
    for validator in measurement.validators:
      mangled_param.description += '\nValidator: ' + str(validator)
    mangled_parameters[mangled_name] = mangled_param


def _ExtractParameters(record, testrun, used_parameter_names):
  """Extract parameters from phases.

  Generate mangled parameters afterwards so we give real measurements priority
  getting names.
  """
  mangled_parameters = {}
  for phase in record.phases:
    testrun_phase = testrun.phases.add()
    testrun_phase.name = phase.name
    testrun_phase.description = phase.code
    testrun_phase.timing.start_time_millis = phase.start_time_millis
    testrun_phase.timing.end_time_millis = phase.end_time_millis

    _ExtractAttachments(phase, testrun, used_parameter_names)
    for name, measurement in sorted(phase.measurements.items()):
      tr_name = name
      while tr_name in used_parameter_names:
        tr_name += '_'
      used_parameter_names.add(tr_name)
      testrun_param = testrun.test_parameters.add()
      testrun_param.name = tr_name
      testrun_param.status = (
          testrun_pb2.PASS if measurement.outcome else testrun_pb2.FAIL)
      if measurement.docstring:
        testrun_param.description = measurement.docstring
      if measurement.units:
        testrun_param.unit_code = UOM_CODE_MAP[measurement.units.uom_code]

      if name not in phase.measured_values:
        testrun_param.status = testrun_pb2.ERROR
        continue
      value = phase.measured_values[name]
      if measurement.dimensions is None:
        # Just a plain ol' value.
        if isinstance(value, numbers.Number):
          testrun_param.numeric_value = float(value)
        else:
          testrun_param.text_value = str(value)
        # Check for validators we know how to translate.
        for validator in measurement.validators:
          if isinstance(validator, validators.InRange):
            if validator.minimum is not None:
              testrun_param.numeric_minimum = float(validator.minimum)
            if validator.maximum is not None:
              testrun_param.numeric_maximum = float(validator.maximum)
          elif isinstance(validator, validators.RegexMatcher):
            testrun_param.expected_text = validator.regex
          else:
            testrun_param.description += '\nValidator: ' + str(validator)
      else:
        _MangleMeasurement(name, value, measurement, mangled_parameters)
  return mangled_parameters


def _AddMangledParameters(testrun, mangled_parameters, used_parameter_names):
  """Add any mangled parameters we generated from multidim measurements."""
  for mangled_name, mangled_param in sorted(mangled_parameters.items()):
    while mangled_name in used_parameter_names:
      logging.warning('Mangled name %s in use by non-mangled parameter',
                      mangled_name)
      mangled_name += '_'
    testrun_param = testrun.test_parameters.add()
    testrun_param.CopyFrom(mangled_param)


def _AddLogLines(record, testrun):
  """Copy log records over, this is a fairly straightforward mapping."""
  for log in record.log_records:
    testrun_log = testrun.test_logs.add()
    testrun_log.timestamp_millis = log.timestamp_millis
    testrun_log.log_message = log.message
    testrun_log.logger_name = log.logger_name
    testrun_log.levelno = log.level
    # pylint: disable=no-member
    if log.level <= logging.DEBUG:
      testrun_log.level = testrun_pb2.TestRunLogMessage.DEBUG
    elif log.level <= logging.INFO:
      testrun_log.level = testrun_pb2.TestRunLogMessage.INFO<commit_msg>add failed parameters test_run failure_codes in mfg_inspector<commit_after>"""Output or upload a TestRun proto for mfg-inspector.com"""

import json
import logging
import numbers
import zlib
import httplib2

from openhtf.io.output import json_factory
from openhtf.io.proto import testrun_pb2
from openhtf.io.proto import units_pb2

from openhtf import util
from openhtf.exe import test_state
from openhtf.util import validators

# pylint: disable=no-member
MIMETYPE_MAP = {
  'image/jpeg': testrun_pb2.InformationParameter.JPG,
  'image/png': testrun_pb2.InformationParameter.PNG,
  'audio/x-wav': testrun_pb2.InformationParameter.WAV,
  'text/plain': testrun_pb2.InformationParameter.TEXT_UTF8,
  'image/tiff': testrun_pb2.InformationParameter.TIFF,
  'video/mp4': testrun_pb2.InformationParameter.MP4,
}
OUTCOME_MAP = {
  'State.ERROR': testrun_pb2.ERROR,
  'State.TIMEOUT': testrun_pb2.ERROR,
  'State.ABORTED': testrun_pb2.ERROR,
  'State.FAIL': testrun_pb2.FAIL,
  'State.PASS': testrun_pb2.PASS,
}
UOM_CODE_MAP = {
  u.GetOptions().Extensions[units_pb2.uom_code]: num
  for num, u in units_pb2.Units.UnitCode.DESCRIPTOR.values_by_number.iteritems()
}
# pylint: enable=no-member

# Control how many flattened parameters we'll output per multidimensional
# measurement.
MAX_PARAMS_PER_MEASUREMENT = 100


class UploadFailedError(Exception):
  """Raised when an upload to mfg-inspector fails."""


# pylint: disable=invalid-name
def _PopulateHeader(record, testrun):
  """Populate header-like info in testrun from record.

  Mostly obvious, some stuff comes from metadata, see docstring of
  TestRunFromTestRecord for details.
  """
  testrun.dut_serial = record.dut_id
  testrun.tester_name = record.station_id
  if 'test_name' in record.metadata:
    testrun.test_info.name = record.metadata['test_name']
  else:
    # Default to copying tester_name into test_info.name.
    testrun.test_info.name = record.station_id
  if 'test_description' in record.metadata:
    testrun.test_info.description = record.metadata['test_description']
  if 'test_version' in record.metadata:
    testrun.test_info.version_string = record.metadata['test_version']
  testrun.test_status = OUTCOME_MAP.get(record.outcome, testrun_pb2.ERROR)
  testrun.start_time_millis = record.start_time_millis
  testrun.end_time_millis = record.end_time_millis
  if 'run_name' in record.metadata:
    testrun.run_name = record.metadata['run_name']
  for details in record.outcome_details:
    testrun_code = testrun.failure_codes.add()
    if details.code_type == test_state.TestState.State.ERROR:
      testrun_code.code = details.code
      testrun_code.details = details.description
    else:
      testrun_code.code = details.code_type
      testrun_code.details = '%s: %s' % (details.code, details.description)


def _AttachJson(record, testrun):
  """Attach a copy of the JSON-ified record as an info parameter.

  Save a copy of the JSON-ified record in an attachment so we can access
  un-mangled fields later if we want.  Remove attachments since those get
  copied over and can potentially be quite large.
  """
  record_dict = util.convert_to_dict(record, ignore_keys=('attachments',))
  record_json = json_factory.OutputToJSON(sort_keys=True).encode(record_dict)
  testrun_param = testrun.info_parameters.add()
  testrun_param.name = 'OpenHTF_record.json'
  testrun_param.value_binary = record_json
  # pylint: disable=no-member
  testrun_param.type = testrun_pb2.InformationParameter.TEXT_UTF8
  # pylint: enable=no-member


def _ExtractAttachments(phase, testrun, used_parameter_names):
  """Extract attachments, just copy them over."""
  for name, (data, mimetype) in sorted(phase.attachments.items()):
    while name in used_parameter_names:
      name += '_'  # Hack to avoid collisions between phases.
    used_parameter_names.add(name)
    testrun_param = testrun.info_parameters.add()
    testrun_param.name = name
    testrun_param.value_binary = data
    if mimetype in MIMETYPE_MAP:
      testrun_param.type = MIMETYPE_MAP[mimetype]
    else:
      # pylint: disable=no-member
      testrun_param.type = testrun_pb2.InformationParameter.BINARY
      # pylint: enable=no-member


def _MangleMeasurement(name, value, measurement, mangled_parameters):
  """Flatten parameters for backwards compatibility, watch for collisions.

  We generate these by doing some name mangling, using some sane limits for
  very large multidimensional measurements.
  """
  for current_value in value[:MAX_PARAMS_PER_MEASUREMENT]:
    # Mangle names so they look like 'myparameter_Xsec_Ynm_ZHz'
    mangled_name = '_'.join([name] + [
        '%s%s' % (
          dim_val,
          dim_units.uom_suffix if dim_units.uom_suffix else '') for
        dim_val, dim_units in zip(
          current_value[:-1], measurement.dimensions)])
    while mangled_name in mangled_parameters:
      logging.warning('Mangled name %s already in use', mangled_name)
      mangled_name += '_'
    mangled_param = testrun_pb2.TestParameter()
    mangled_param.name = mangled_name
    mangled_param.numeric_value = float(current_value[-1])
    if measurement.units:
      mangled_param.unit_code = UOM_CODE_MAP[measurement.units.uom_code]
    mangled_param.description = (
        'Mangled parameter from measurement %s with dimensions %s' % (
        name, tuple(d.uom_suffix for d in measurement.dimensions)))
    for validator in measurement.validators:
      mangled_param.description += '\nValidator: ' + str(validator)
    mangled_parameters[mangled_name] = mangled_param


def _ExtractParameters(record, testrun, used_parameter_names):
  """Extract parameters from phases.

  Generate mangled parameters afterwards so we give real measurements priority
  getting names.
  """
  mangled_parameters = {}
  for phase in record.phases:
    testrun_phase = testrun.phases.add()
    testrun_phase.name = phase.name
    testrun_phase.description = phase.code
    testrun_phase.timing.start_time_millis = phase.start_time_millis
    testrun_phase.timing.end_time_millis = phase.end_time_millis

    _ExtractAttachments(phase, testrun, used_parameter_names)
    for name, measurement in sorted(phase.measurements.items()):
      tr_name = name
      while tr_name in used_parameter_names:
        tr_name += '_'
      used_parameter_names.add(tr_name)
      testrun_param = testrun.test_parameters.add()
      testrun_param.name = tr_name
      testrun_param.status = (
          testrun_pb2.PASS if measurement.outcome else testrun_pb2.FAIL)
      if measurement.docstring:
        testrun_param.description = measurement.docstring
      if measurement.units:
        testrun_param.unit_code = UOM_CODE_MAP[measurement.units.uom_code]

      if name not in phase.measured_values:
        testrun_param.status = testrun_pb2.ERROR
        continue
      value = phase.measured_values[name]
      if measurement.dimensions is None:
        # Just a plain ol' value.
        if isinstance(value, numbers.Number):
          testrun_param.numeric_value = float(value)
        else:
          testrun_param.text_value = str(value)
        # Check for validators we know how to translate.
        for validator in measurement.validators:
          if isinstance(validator, validators.InRange):
            if validator.minimum is not None:
              testrun_param.numeric_minimum = float(validator.minimum)
            if validator.maximum is not None:
              testrun_param.numeric_maximum = float(validator.maximum)
          elif isinstance(validator, validators.RegexMatcher):
            testrun_param.expected_text = validator.regex
          else:
            testrun_param.description += '\nValidator: ' + str(validator)
      else:
        _MangleMeasurement(name, value, measurement, mangled_parameters)
      if testrun_param.status == testrun_pb2.FAIL:
        testrun_code = testrun.failure_codes.add()
        testrun_code.code = testrun_param.name
        if measurement.dimensions is None:
          if testrun_param.numeric_value:
            testrun_code.details = str(testrun_param.numeric_value)
          else:
            testrun_code.details = testrun_param.text_value
  return mangled_parameters


def _AddMangledParameters(testrun, mangled_parameters, used_parameter_names):
  """Add any mangled parameters we generated from multidim measurements."""
  for mangled_name, mangled_param in sorted(mangled_parameters.items()):
    while mangled_name in used_parameter_names:
      logging.warning('Mangled name %s in use by non-mangled parameter',
                      mangled_name)
      mangled_name += '_'
    testrun_param = testrun.test_parameters.add()
    testrun_param.CopyFrom(mangled_param)


def _AddLogLines(record, testrun):
  """Copy log records over, this is a fairly straightforward mapping."""
  for log in record.log_records:
    testrun_log = testrun.test_logs.add()
    testrun_log.timestamp_millis = log.timestamp_millis
    testrun_log.log_message = log.message
    testrun_log.logger_name = log.logger_name
    testrun_log.levelno = log.level
    # pylint: disable=no-member
    if log.level <= logging.DEBUG:
      testrun_log.level = testrun_pb2.TestRunLogMessage.DEBUG
    elif log.level <= logging.INFO:
      testrun_log.level = testrun_pb2.TestRunLogMessage.INFO
{'fraction_non_alphanumeric': 0.0859478568272205, 'fraction_numerical': 0.0038113124171453823, 'mean_word_length': 3.911828540423223, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7538274', 'n_tokens_mistral': 5694, 'n_tokens_neox': 5602, 'n_words': 1257}