Source code for labgrid.util.qmp

import json
import logging
import attr


[docs]@attr.s(eq=False) class QMPMonitor: monitor_out = attr.ib() monitor_in = attr.ib()
[docs] def __attrs_post_init__(self): self.logger = logging.getLogger(f"{self}:") self._negotiate_capabilities()
def _negotiate_capabilities(self): greeting = self._read_parse_json() if not greeting.get('QMP'): raise QMPError("QMP greeting message invalid") self.monitor_in.write(json.dumps({"execute": "qmp_capabilities"}).encode("utf-8")) self.monitor_in.write("\n".encode("utf-8")) self.monitor_in.flush() answer = self._read_parse_json() if not "return" in answer: raise QMPError(f"Could not connect to QMP: {answer}") def _read_parse_json(self): line = self.monitor_out.readline().decode('utf-8') self.logger.debug("Received line: %s", line.rstrip("\r\n")) if not line: raise QMPError("Received empty response") return json.loads(line)
[docs] def execute(self, command): json_command = {"execute": command} self.monitor_in.write(json.dumps(json_command).encode("utf-8")) self.monitor_in.write("\n".encode("utf-8")) self.monitor_in.flush() answer = self._read_parse_json() # skip all asynchronous events while answer.get('event'): answer = self._read_parse_json() if "error" in answer: raise QMPError(answer['error']) return answer['return']
[docs]@attr.s(eq=False) class QMPError(Exception): msg = attr.ib(validator=attr.validators.instance_of(str))