Phase Branches Example

import os.path

import openhtf as htf
from openhtf.output.callbacks import json_factory
from openhtf.plugs import user_input


class DeviceType(htf.DiagResultEnum):
    PROTOTYPE = 'prototype'
    EVT = 'evt'
    DVT = 'dvt'


class OperatorError(Exception):
    pass


# --- Step 1: a diagnoser turns the operator's answer into a Diagnosis ---

@htf.PhaseDiagnoser(DeviceType)
def diagnose_test_branch(phase_rec: htf.PhaseRecord):
    answer = phase_rec.measurements['test_branch'].measured_value.value.lower()
    try:
        branch = DeviceType(answer)
    except ValueError as e:
        options = ', '.join(d.value for d in DeviceType)
        raise OperatorError(f"Input {answer!r} not recognized. Expected one of: {options}.") from e
    return htf.Diagnosis(branch)


@htf.diagnose(diagnose_test_branch)
@htf.measures(htf.Measurement('test_branch'))
@htf.plug(prompts=user_input.UserInput)
@htf.PhaseOptions(phase_name_case=htf.PhaseNameCase.CAMEL)
def select_testing_branch_phase(test: htf.TestApi, prompts: user_input.UserInput):
    options = ', '.join(f'"{d.value}"' for d in DeviceType)
    test.measurements.test_branch = prompts.prompt(
        f'Which branch is this unit? {options}', text_input=True)


# --- EVT and DVT: a prompt and a validation phase each ---

@htf.plug(prompts=user_input.UserInput)
def evt_observation_phase(test, prompts):
    test.logger.info('Operator observations: %s', prompts.prompt('EVT observations?', text_input=True))

def evt_validation_phase(test):
    test.logger.info('EVT parameters within limits.')

@htf.plug(prompts=user_input.UserInput)
def dvt_observation_phase(test, prompts):
    test.logger.info('Operator observations: %s', prompts.prompt('DVT observations?', text_input=True))

def dvt_validation_phase(test):
    test.logger.info('DVT validation complete.')


# --- Prototype: a second diagnoser drives a nested branch ---

class PrototypeOutcome(htf.DiagResultEnum):
    INCINERATE = 'incinerate'
    PROMISE_CAKE = 'promise_cake'


@htf.PhaseDiagnoser(PrototypeOutcome)
def diagnose_prototype_observation(phase_rec: htf.PhaseRecord):
    observation = phase_rec.measurements['prototype_observation'].measured_value.value.lower()
    if any(word in observation for word in ('fire', 'smoke', 'fail', 'bad')):
        return htf.Diagnosis(PrototypeOutcome.INCINERATE)
    return htf.Diagnosis(PrototypeOutcome.PROMISE_CAKE)


@htf.diagnose(diagnose_prototype_observation)
@htf.measures(htf.Measurement('prototype_observation'))
@htf.plug(prompts=user_input.UserInput)
def prototype_observation_phase(test, prompts):
    test.measurements.prototype_observation = prompts.prompt('Prototype observations?', text_input=True)

@htf.plug(prompts=user_input.UserInput)
def incinerate_phase(test, prompts):
    test.logger.warning('Dangerous anomalies detected.')
    prompts.prompt('Incineration sequence initiated. Press Okay.')
    return htf.PhaseResult.FAIL_AND_CONTINUE

def promise_cake_phase(test):
    test.logger.info('No failures detected. You remain eligible for cake.')


# --- Step 2: assemble the branches ---

def create_and_run_test(output_dir: str = '.'):
    prototype_branch = htf.BranchSequence(
        htf.DiagnosisCondition.on_all(DeviceType.PROTOTYPE),
        prototype_observation_phase,
        htf.BranchSequence(htf.DiagnosisCondition.on_all(PrototypeOutcome.PROMISE_CAKE), promise_cake_phase),
        htf.BranchSequence(htf.DiagnosisCondition.on_all(PrototypeOutcome.INCINERATE), incinerate_phase),
    )
    evt_branch = htf.BranchSequence(
        htf.DiagnosisCondition.on_all(DeviceType.EVT), evt_observation_phase, evt_validation_phase)
    dvt_branch = htf.BranchSequence(
        htf.DiagnosisCondition.on_all(DeviceType.DVT), dvt_observation_phase, dvt_validation_phase)

    test = htf.Test(select_testing_branch_phase, prototype_branch, evt_branch, dvt_branch)
    test.add_output_callbacks(
        json_factory.OutputToJSON(os.path.join(output_dir, '{dut_id}.phase_branches.json'), indent=2))
    test.execute(test_start=user_input.prompt_for_test_start())


if __name__ == '__main__':
    create_and_run_test()

$ python phase_branches.py -v
Enter a DUT ID in order to start the test.
--> SN1234
Which branch is this unit? "prototype", "evt", "dvt"
--> prototype
Prototype observations?
--> smoke from U7
W ... <phase: incinerate_phase> - Dangerous anomalies detected.
Incineration sequence initiated. Press Okay.
-->

======================= test: openhtf_test  outcome: FAIL ======================