EVNotiPi/EVNotifyAPI/evnotify.py at master · EVNotify/EVNotiPi · GitHub

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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
""" Python interface for EVNotify API """

import requests

class CommunicationError(Exception): pass

class EVNotify:

    def __init__(self, akey=None, token=None):
        self._rest_url = 'https://app.evnotify.de/'
        self._session = requests.Session()
        self._session.headers.update({'User-Agent': 'PyEVNotifyApi/2'})
        self._akey = akey
        self._token = token
        self._timeout = 5

    def sendRequest(self, method, fnc, useAuthentication=False, data={}):
        params = {**data}

        if useAuthentication:
            params['akey'] = self._akey
            params['token'] = self._token

        try:
            if method == 'get':
                result = getattr(self._session, method)(self._rest_url + fnc,
                                                        params=params,
                                                        timeout=self._timeout)
            else:
                result = getattr(self._session, method)(self._rest_url + fnc,
                                                        json=params,
                                                        timeout=self._timeout)
            if result.status_code >= 400:
                raise CommunicationError("code({}) text({})"
                                         .format(result.status_code, result.text))

            return result.json()

        except requests.exceptions.ConnectionError:
            raise CommunicationError("connection failed")
        except requests.exceptions.Timeout:
            raise CommunicationError("timeout")


    def getKey(self):
        ret = self.sendRequest('get', 'key')

        if 'akey' not in ret:
            raise CommunicationError("return akey missing")

        return ret['akey']

    def getToken(self):
        return self._token

    def register(self, akey, password):
        ret = self.sendRequest('post', 'register', False, {
            "akey": akey,
            "password": password
        })

        if 'token' not in ret:
            raise CommunicationError("return token missing")

        self._token = ret['token']
        self._akey = akey

    def login(self, akey, password):
        ret = self.sendRequest('post', 'login', False, {
            "akey": akey,
            "password": password
        })

        if 'token' not in ret:
            raise CommunicationError("return token missing")

        self._token = ['token']
        self._akey = akey

    def changePassword(self, oldpassword, newpassword):
        ret = self.sendRequest('post', 'changepw', True, {
            "oldpassword": oldpassword,
            "newpassword": newpassword
        })

        return ret['changed'] if 'changed' in ret else None

    def getSettings(self):
        ret = self.sendRequest('get', 'settings', True)

        if 'settings' not in ret:
            raise CommunicationError("return settings missing")

        return ret['settings']

    def setSettings(self, settings):
        ret = self.sendRequest('put', 'settings', True, {
            "settings": settings
        })

        if 'settings' not in ret:
            raise CommunicationError()

    def setSOC(self, display, bms):
        ret = self.sendRequest('post', 'soc', True, {
            "display": display,
            "bms": bms
        })

        if 'synced' not in ret:
            raise CommunicationError("return settings missing")

    def getSOC(self):
        return self.sendRequest('get', 'soc', True)

    def setExtended(self, obj):
        ret = self.sendRequest('post', 'extended', True, obj)

        if 'synced' not in ret:
            raise CommunicationError("return synced missing")

    def getExtended(self):
        return self.sendRequest('get', 'extended', True)

    def getLocation(self):
        return self.sendRequest('get', 'location', True)

    def setLocation(self, obj):
        ret = self.sendRequest('post', 'location', True, obj)

        if 'synced' not in ret:
            raise CommunicationError("return synced missing")

    def renewToken(self, password):
        ret = self.sendRequest('put', 'renewtoken', True, {
            "password": password
        })

        if 'token' not in ret:
            raise CommunicationError("return token missing")

        self._token = ret['token']

        return self._token

    def sendNotification(self, abort=False):
        ret = self.sendRequest('post', 'notification', True, {
            "abort": abort
        })

        if 'notified' not in ret:
            raise CommunicationError("return notified missing")

        return ret['notified']