(dct, update_ids_from_dap=False)
| 82 | |
| 83 | |
| 84 | def from_dict(dct, update_ids_from_dap=False): |
| 85 | msg_type = dct.get("type") |
| 86 | if msg_type is None: |
| 87 | raise ValueError("Unable to make sense of message: %s" % (dct,)) |
| 88 | |
| 89 | if msg_type == "request": |
| 90 | to_type = _requests_to_types |
| 91 | use = dct["command"] |
| 92 | |
| 93 | elif msg_type == "response": |
| 94 | to_type = _responses_to_types |
| 95 | use = dct["command"] |
| 96 | |
| 97 | else: |
| 98 | to_type = _event_to_types |
| 99 | use = dct["event"] |
| 100 | |
| 101 | cls = to_type.get(use) |
| 102 | if cls is None: |
| 103 | raise ValueError("Unable to create message from dict: %s. %s not in %s" % (dct, use, sorted(to_type.keys()))) |
| 104 | try: |
| 105 | return cls(update_ids_from_dap=update_ids_from_dap, **dct) |
| 106 | except: |
| 107 | msg = "Error creating %s from %s" % (cls, dct) |
| 108 | debug_exception(msg) |
| 109 | raise |
| 110 | |
| 111 | |
| 112 | def from_json(json_msg, update_ids_from_dap=False, on_dict_loaded=lambda dct: None): |
no test coverage detected