dream

1
import logging
2
from typing import Dict, List
3

4
logger = logging.getLogger(__name__)
5

6
LAST_N_TURNS = 5  # number of turns to consider in annotator/skill.
7

8

9
def catcher_formatter(dialog: Dict) -> Dict:
10
    return [{"x": [dialog["utterances"][-1]["text"]]}]
11

12

13
def last_utt_dialog(dialog: Dict) -> Dict:
14
    return [{"sentences": [dialog["utterances"][-1]["text"]]}]
15

16

17
def base_response_selector_formatter_service(payload: List) -> Dict:
18
    if len(payload) == 3:
19
        return {"skill_name": payload[0], "text": payload[1], "confidence": payload[2]}
20
    elif len(payload) == 5:
21
        return {
22
            "skill_name": payload[0],
23
            "text": payload[1],
24
            "confidence": payload[2],
25
            "human_attributes": payload[3],
26
            "bot_attributes": payload[4],
27
        }
28

29

30
def full_dialog(dialog: Dict):
31
    return [{"dialogs": [dialog]}]
32

33

34
def base_skill_formatter(payload: Dict) -> Dict:
35
    return [{"text": payload[0], "confidence": payload[1]}]
36

37

38
def simple_formatter_service(payload: List):
39
    logging.info("answer " + str(payload))
40
    return payload
41

42

43
def entity_linking_formatter(payload: List):
44
    response = []
45
    for entity_name, wikidata_ids, id_types in zip(*payload):
46
        item = {
47
            "entity_name": entity_name,
48
            "wikidata_ids": [{"id": id, "instance_of": instance_of} for id, instance_of in zip(wikidata_ids, id_types)],
49
        }
50
        response.append(item)
51
    return response
52

53

54
def hypotheses_list(dialog: Dict) -> Dict:
55
    hypotheses = dialog["utterances"][-1]["hypotheses"]
56
    hypots = [h["text"] for h in hypotheses]
57
    return [{"sentences": hypots}]
58

59

60
def programy_formatter_dialog(dialog: Dict) -> List:
61
    return [{"sentences_batch": [[u["text"] for u in dialog["utterances"][-5:]]]}]
62

63

64
def skill_with_attributes_formatter_service(payload: Dict):
65
    """
66
    Formatter should use `"state_manager_method": "add_hypothesis"` in config!!!
67
    Because it returns list of hypothesis even if the payload is returned for one sample!
68

69
    Args:
70
        payload: if one sample, list of the following structure:
71
            (text, confidence, ^human_attributes, ^bot_attributes, attributes) [by ^ marked optional elements]
72
                if several hypothesis, list of lists of the above structure
73

74
    Returns:
75
        list of dictionaries of the following structure:
76
            {"text": text, "confidence": confidence_value,
77
             ^"human_attributes": {}, ^"bot_attributes": {},
78
             **attributes},
79
             by ^ marked optional elements
80
    """
81
    # Used by: skill_with_attributes_formatter, news_skill, meta_script_skill, dummy_skill
82
    if isinstance(payload[0], list) and isinstance(payload[1], list):
83
        result = [{"text": hyp[0], "confidence": hyp[1]} for hyp in zip(*payload)]
84
    else:
85
        result = [{"text": payload[0], "confidence": payload[1]}]
86

87
    if len(payload) >= 4:
88
        if isinstance(payload[2], dict) and isinstance(payload[3], dict):
89
            result[0]["human_attributes"] = payload[2]
90
            result[0]["bot_attributes"] = payload[3]
91
        elif isinstance(payload[2], list) and isinstance(payload[3], list):
92
            for i, hyp in enumerate(zip(*payload)):
93
                result[i]["human_attributes"] = hyp[2]
94
                result[i]["bot_attributes"] = hyp[3]
95

96
    if len(payload) == 3 or len(payload) == 5:
97
        if isinstance(payload[-1], dict):
98
            for key in payload[-1]:
99
                result[0][key] = payload[-1][key]
100
        elif isinstance(payload[-1], list):
101
            for i, hyp in enumerate(zip(*payload)):
102
                for key in hyp[-1]:
103
                    result[i][key] = hyp[-1][key]
104

105
    return result
106