blob: 5b9541ea4cf4d9a9a3399096371b06506d6fbeea (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
|
import socket
from enum import Enum
class Format(Enum):
JSON = 'json'
TABLE = 'table'
SIMPLE_TABLE = 'simple-table'
class Client:
def __init__(self, port=8305, host='127.0.0.1'):
self._host = host
self._port = port
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def __del__(self):
self.sock.close()
def connect(self):
self.sock.connect((self._host, self._port))
def _write(self, line):
self.sock.sendall((line+'\r\n').encode())
def _read(self):
buf = bytearray()
while True:
buf.extend(self.sock.recv(256))
if b'\r\n\r\n' in buf:
break
response = buf.decode().strip().split("\r\n")
status = response.pop(0)
if status not in ('ok', 'err'):
raise InverterError(f"Unexpected status '{status}'")
if status == 'err':
raise InverterError(response[0] if len(response) >= 0 else "Unknown error")
return '\r\n'.join(response)
def protocol(self, v):
self._write(f'v {v}')
return self._read()
def format(self, format):
self._write(f'format {format}')
return self._read()
def exec(self, command, arguments=()):
buf = f'exec {command}'
for arg in arguments:
buf += f' {arg}'
self._write(buf)
return self._read()
class InverterError(RuntimeError): pass
|
