diff options
| -rwxr-xr-x | .gitignore | 2 | ||||
| -rw-r--r-- | inserter.py | 68 | ||||
| -rw-r--r-- | test/test_usv.py | 65 | ||||
| -rw-r--r-- | udev/90-usv_usbuart.rules | 1 | ||||
| -rw-r--r-- | usv/__init__.py | 1 | ||||
| -rw-r--r-- | usv/usv.py | 56 |
6 files changed, 193 insertions, 0 deletions
@@ -35,3 +35,5 @@ nosetests.xml .mr.developer.cfg .project .pydevproject + +*.swp diff --git a/inserter.py b/inserter.py new file mode 100644 index 0000000..70088c9 --- /dev/null +++ b/inserter.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import sys +import argparser + +from time import time, sleep +from urllib.request import urlopen + +from usv import Usv +from pyinfluxdb import Line + +class UsvInserter(object): + TIME = 10 + + def __init__(self, port, url): + self.port = port + self.url = url + + self.usv = Usv(port) + self.duration = 0 + + def job(self): + start = time() + + fields = self.usv.read() + + data = [] + for name, value in fields.items(): + line = Line( + 'usv_parameters', + {'name': name, 'id': 'LI????VA' }, + {'value': value} + ) + data.append(line) + + post_data = '\n'.join(map(str, data)) + + print(post_data) + + with urlopen(self.url, post_data.encode('utf-8')) as fh: + print(fh.read().decode('utf-8')) + + self.duration = time() - start + + def get_duration(self): + return self.duration + + +if __name__ == '__main__': + """ inserter """ + + parser = argparse.ArgumentParser() + parser.add_argument('--url', help='URL') + parser.add_argument('--port', help='Serial port device', + default='/dev/ttyUSV') + args = parser.parse_args(sys.argv[1:]) + + usv_inserter = UsvInserter(args.port, args.url) + + print('Start inserter with url {} on port {}'.format(args.url, args.port)) + + while True: + usv_inserter.job() + + if usv_inserter.get_duration() > 0: + sleep(UsvInserter.TIME - UsvInserter.get_duration()) + diff --git a/test/test_usv.py b/test/test_usv.py new file mode 100644 index 0000000..5a013ae --- /dev/null +++ b/test/test_usv.py @@ -0,0 +1,65 @@ +import pytest + +from usv import Usv + +class MockUsvSerial(object): + def __init__(self, *args, **kwargs): + self.buf = [] + + def open(self): + pass + + def write(self, buf): + self.buf.insert(1, buf) + + def read(self, size): + cmd = self.buf.pop() + + if cmd == '\r': + return '\r=>' + elif cmd == 'd 1\r': + return '01 V In 219\r=>' + elif cmd == 'd 2\r': + return '02 V Out 215\r=>' + elif cmd == 'd 5\r': + return '05 VA Out 108\r=>' + elif cmd == 'd 7\r': + return '07 V Batt 52.7\r=>' + elif cmd == 'd 8\r': + return '08 Freq 50.01 Hz\r=>' + elif cmd == 'd 16\r': + return '16 FullLoad% 007\r=>' + elif cmd == 'd 17\r': + return '17 Watts 83\r=>' + elif cmd == 'd 18\r': + return '18 PF 1.00 ----\r=>' + elif cmd == 'd 19\r': + return '19 CrestF 1.44\r=>' + elif cmd == 'd 20\r': + return '20 #PwrOut 157\r=>' + elif cmd == 'd 23\r': + return '23 InvMin 0231.2\r=>' + + +class TestUsv(object): + def test_usv(self, monkeypatch): + monkeypatch.setattr('usv.usv.Serial', MockUsvSerial) + + usv_handle = Usv('/dev/usv') + + fields = usv_handle.read() + + assert len(fields) == 11 + + assert fields['acvoltsin'] == 219.0 + assert fields['acvoltsout'] == 215.0 + assert fields['vaout'] == 108.0 + assert fields['vbattery'] == 52.7 + assert fields['frequency'] == 50.01 + assert fields['fullload'] == 7.0 + assert fields['watts'] == 83.0 + assert fields['powerfactor'] == 1.0 + assert fields['crestfactor'] == 1.44 + assert fields['powerout'] == 157.0 + assert fields['inverterminutes'] == 231.2 + diff --git a/udev/90-usv_usbuart.rules b/udev/90-usv_usbuart.rules new file mode 100644 index 0000000..9cdc877 --- /dev/null +++ b/udev/90-usv_usbuart.rules @@ -0,0 +1 @@ +SUBSYSTEMS=="usb", ATTRS{idProduct}=="2303", ATTRS{idVendor}=="067b", SYMLINK="ttyUSV" diff --git a/usv/__init__.py b/usv/__init__.py new file mode 100644 index 0000000..d48dae4 --- /dev/null +++ b/usv/__init__.py @@ -0,0 +1 @@ +from .usv import Usv diff --git a/usv/usv.py b/usv/usv.py new file mode 100644 index 0000000..14b8735 --- /dev/null +++ b/usv/usv.py @@ -0,0 +1,56 @@ +import re +from serial import Serial + +parameters = { + 'acvoltsin': (1, '01 V In +([0-9]+[.]?[0-9]*)'), + 'acvoltsout': (2, '02 V Out +([0-9]+[.]?[0-9]*)'), + 'vaout': (5, '05 VA Out +([0-9]+[.]?[0-9]*)'), + 'vbattery': (7, '07 V Batt +([0-9]+[.]?[0-9]*)'), + 'frequency': (8, '08 Freq ([0-9]+[.]?[0-9]*) Hz'), + 'fullload': (16, '16 FullLoad% ([0-9]+[.]?[0-9]*)'), + 'watts': (17, '17 Watts +([0-9]+[.]?[0-9]*)'), + 'powerfactor': (18, '18 PF +([0-9]+[.]?[0-9]*) ----'), + 'crestfactor': (19, '19 CrestF +([0-9]+[.]?[0-9]*)'), + 'powerout': (20, '20 #PwrOut +([0-9]+[.]?[0-9]*)'), + 'inverterminutes': (23, '23 InvMin +([0-9]+[.]?[0-9]*)') + } + + +class UsvSerialHandler(object): + def __init__(self, device_path): + """ Handles the serial connection to the usv """ + + self._serial_port = Serial(device_path, + baudrate=600, + timeout=0.8) + + # force clean prompt + self._serial_port.write('\r') + self._serial_port.read(100) + + def request(self, index): + """ Write and read until prompt """ + + self._serial_port.write('d {}\r'.format(index)) + + return self._serial_port.read(21) + + +class Usv(UsvSerialHandler): + def __init__(self, device_path): + super().__init__(device_path) + + def read(self): + fields = {} + + for name, (index, regex) in parameters.items(): + result = self.request(index) + + match = re.search(regex, result, re.MULTILINE) + + if match: + value = float(match.group(1)) + + fields[name] = value + + return fields |