diff --git a/.gitignore b/.gitignore index 13fbcc0..c32263c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ .idea/ -web/EVNotifyAPI/ \ No newline at end of file +web/EVNotifyAPI/ +__pycache__/ diff --git a/cars/IONIQ_BEV.py b/cars/IONIQ_BEV.py index 7809273..f9b624e 100644 --- a/cars/IONIQ_BEV.py +++ b/cars/IONIQ_BEV.py @@ -1,70 +1,71 @@ -class IONIQ_BEV: +from car import * +from time import time + +POLL_DELAY_2180 = 60 # Rate limit b2180 to once a minute +b2101 = bytes.fromhex(hex(0x2101)[2:]) +b2105 = bytes.fromhex(hex(0x2105)[2:]) +b2180 = bytes.fromhex(hex(0x2180)[2:]) + +class IONIQ_BEV(car): def __init__(self, dongle): self.dongle = dongle self.dongle.setProtocol('CAN_11_500') - self.dongle.setCANRxFilter('7EC') - self.dongle.setCANRxMask('7FF') + self.dongle.setCANRxFilter(0x7ec) + self.dongle.setCANRxMask(0x7ff) + self.last_raw = {} + self.last_poll_2180 = 0 def getData(self): + now = time() raw = {} - for cmd in [2101,2105]: - raw[cmd] = self.dongle.sendCommand(str(cmd)) - - chargingBits = raw[2101][0x7EC21][5] \ - if 0x7EC21 in raw[2101] else None - dcBatteryCurrent = int.from_bytes(raw[2101][0x7EC21][6:7] + raw[2101][0x7EC22][0:1], - byteorder='big', signed=True) / 10.0 \ - if 0x7EC21 in raw[2101] and 0x7EC22 in raw[2101] else None - dcBatteryVoltage = int.from_bytes(raw[2101][0x7EC22][1:3], - byteorder='big', signed=False) / 10.0 \ - if 0x7EC22 in raw[2101] else None - - data = {'SOC_BMS': raw[2101][0x7EC21][0] / 2.0 \ - if 0x7EC21 in raw[2101] else None, - 'SOC_DISPLAY': raw[2105][0x7EC24][6] / 2.0 \ - if 0x7EC24 in raw[2105] else None, - 'EXTENDED': { - 'auxBatteryVoltage': raw[2101][0x7EC24][4] / 10.0 \ - if 0x7EC24 in raw[2105] else None, - - 'batteryInletTemperature': int.from_bytes(raw[2101][0x7EC23][2:3], - byteorder='big', signed=True) \ - if 0x7EC23 in raw[2105] else None, - - 'batteryMaxTemperature': int.from_bytes(raw[2101][0x7EC22][3:4], - byteorder='big', signed=True) \ - if 0x7EC22 in raw[2105] else None, - - 'batteryMinTemperature': int.from_bytes(raw[2101][0x7EC22][4:5], - byteorder='big', signed=True) \ - if 0x7EC22 in raw[2105] else None, - - 'charging': 1 if chargingBits != None and \ - chargingBits & 0x80 == 0x80 else 0, - - 'normalChargePort': 1 if chargingBits != None and \ - chargingBits & 0x20 == 0x20 else 0, - - 'rapidChargePort': 1 if chargingBits != None and \ - chargingBits & 0x40 == 0x40 else 0, - - 'dcBatteryCurrent': dcBatteryCurrent, - - 'dcBatteryPower': dcBatteryCurrent * dcBatteryVoltage / 1000.0 \ - if dcBatteryCurrent!= None and dcBatteryVoltage != None else None, - - 'dcBatteryVoltage': dcBatteryVoltage, - - 'soh': int.from_bytes(raw[2105][0x7EC24][0:2], - byteorder='big', signed=False) / 10.0 \ - if 0x7EC24 in raw[2105] else None, - } + self.dongle.setCANRxFilter(0x7ec) + self.dongle.setCanID(0x7e4) + for cmd in [b2101,b2105]: + raw[cmd] = self.dongle.sendCommand(cmd) + + if now - self.last_poll_2180 > POLL_DELAY_2180 or b2180 not in self.last_raw: + self.last_poll_2180 = now + self.dongle.setCANRxFilter(0x7ee) + self.dongle.setCanID(0x7e6) + raw[b2180] = self.dongle.sendCommand(b2180) + else: + raw[b2180] = self.last_raw[b2180] + + if len(raw[b2101][0x7ec]) != 9 or \ + len(raw[b2105][0x7ec]) != 7 or \ + len(raw[b2180][0x7ee]) != 4: + raise IONIQ_BEV.NULL_BLOCK("Got wrong count of frames!\n"+str(raw)) + + self.last_raw[b2180] = raw[b2180] + + data = self.getBaseData() + + data['SOC_BMS'] = raw[b2101][0x7ec][1][0] / 2.0 + data['SOC_DISPLAY'] = raw[b2105][0x7ec][4][6] / 2.0 + + chargingBits = raw[b2101][0x7ec][1][5] + dcBatteryCurrent = int.from_bytes(raw[b2101][0x7ec][1][6:7] + raw[b2101][0x7ec][2][0:1], byteorder='big', signed=True) / 10.0 + dcBatteryVoltage = int.from_bytes(raw[b2101][0x7ec][2][1:3], byteorder='big', signed=False) / 10.0 + + data['EXTENDED'] = { + 'auxBatteryVoltage': raw[b2101][0x7ec][4][4] / 10.0, + 'batteryInletTemperature': int.from_bytes(raw[b2101][0x7ec][3][2:3], byteorder='big', signed=True), + 'batteryMaxTemperature': int.from_bytes(raw[b2101][0x7ec][2][3:4], byteorder='big', signed=True), + 'batteryMinTemperature': int.from_bytes(raw[b2101][0x7ec][2][4:5], byteorder='big', signed=True), + 'cumulativeEnergyCharged': int.from_bytes(raw[b2101][0x7ec][5][6:7] + raw[b2101][0x7ec][6][0:3], byteorder='big', signed=False) / 10.0, + 'cumulativeEnergyDischarged': int.from_bytes(raw[b2101][0x7ec][6][3:7], byteorder='big', signed=False) / 10.0, + 'charging': 1 if chargingBits & 0x80 else 0, + 'normalChargePort': 1 if chargingBits & 0x20 else 0, + 'rapidChargePort': 1 if chargingBits & 0x40 else 0, + 'dcBatteryCurrent': dcBatteryCurrent, + 'dcBatteryPower': dcBatteryCurrent * dcBatteryVoltage / 1000.0, + 'dcBatteryVoltage': dcBatteryVoltage, + 'outsideTemp': (raw[b2180][0x7ee][2][1] - 80) / 2, + 'soh': int.from_bytes(raw[b2105][0x7ec][4][0:2], byteorder='big', signed=False) / 10.0, } - data.update(self.getBaseData()) - return data def getBaseData(self): diff --git a/cars/KONA_EV.py b/cars/KONA_EV.py index bc494ad..13a3685 100644 --- a/cars/KONA_EV.py +++ b/cars/KONA_EV.py @@ -1,62 +1,60 @@ -class KONA_EV: +from car import * + +b220101 = bytes.fromhex(hex(0x220101)[2:]) +b220105 = bytes.fromhex(hex(0x220105)[2:]) + +class KONA_EV(Car): def __init__(self, dongle): self.dongle = dongle self.dongle.setProtocol('CAN_11_500') self.dongle.setCANRxFilter('7EC') self.dongle.setCANRxMask('7FF') + self.dongle.setCanID(0x7e4) def getData(self): raw = {} - for cmd in [220101,220105]: - raw[cmd] = self.dongle.sendCommand(str(cmd)) - - chargingBits = raw[220101][0x7EC27][5] \ - if 0x7EC27 in raw[220101] else None + for cmd in [b220101,b220105]: + raw[cmd] = self.dongle.sendCommand(cmd) - normalChargePort = raw[220101][0x7EC21][6] == 3 \ - if 0x7EC21 in raw[220101] else None - normalChargeBit = chargingBits & 0x02 == 0x02 + # print("Raw lens {} {}".format(len(raw[b220101][0x7ec]),len(raw[b220105][0x7ec]))) + #if len(raw[b220101][0x7ec]) != 9 or \ # XXX: Need to put in correct line count for Kona + # len(raw[b220105][0x7ec]) != 7: + # raise KONA_EV.NULL_BLOCK("Got wrong count of frames!\n"+str(raw)) - dcBatteryCurrent = int.from_bytes(raw[220101][0x7EC22][0:2], byteorder='big', signed=True) / 10.0 \ - if 0x7EC22 in raw[220101] else None + data = self.getBaseData() - dcBatteryVoltage = int.from_bytes(raw[220101][0x7EC22][2:4], byteorder='big', signed=False) / 10.0 \ - if 0x7EC22 in raw[220101] else None + data['SOC_BMS'] = raw[b220101][0x7ec][1][1] / 2.0 + data['SOC_DISPLAY'] = raw[b220105][0x7ec][5][0] / 2.0 - data = {'SOC_BMS': raw[220101][0x7EC21][1] / 2.0 \ - if 0x7EC21 in raw[220101] else None, - 'SOC_DISPLAY': raw[220105][0x7EC25][0] / 2.0 \ - if 0x7EC25 in raw[220105] else None, - 'EXTENDED': { - 'auxBatteryVoltage': raw[220101][0x7EC24][5] / 10.0 \ - if 0x7EC24 in raw[220101] else None, - 'batteryInletTemperature': int.from_bytes(raw[220101][0x7EC23][5:6], byteorder='big', signed=True) \ - if 0x7EC23 in raw[220101] else None, - 'batteryMaxTemperature': int.from_bytes(raw[220101][0x7EC22][4:5], byteorder='big', signed=True) \ - if 0x7EC22 in raw[220101] else None, - 'batteryMinTemperature': int.from_bytes(raw[220101][0x7EC22][5:6], byteorder='big', signed=True) \ - if 0x7EC22 in raw[220101] else None, - 'charging': 1 if chargingBits & 0xc == 0x8 else 0, - 'normalChargePort': 1 if normalChargeBit and normalChargePort else 0, - 'rapidChargePort': 1 if normalChargeBit and not normalChargePort else 0, - 'dcBatteryCurrent': dcBatteryCurrent, - 'dcBatteryPower': dcBatteryCurrent * dcBatteryVoltage / 1000.0 \ - if dcBatteryCurrent!= None and dcBatteryVoltage != None else None, - 'dcBatteryVoltage': dcBatteryVoltage, - 'soh': int.from_bytes(raw[220105][0x7EC24][1:3], byteorder='big', signed=False) / 10.0 \ - if 0x7EC24 in raw[220105] else None, - } + chargingBits = raw[b220101][0x7ec][7][5] + normalChargePort = raw[b220101][0x7ec][1][6] == 3 + normalChargeBit = chargingBits & 0x02 == 0x02 + dcBatteryCurrent = int.from_bytes(raw[b220101][0x7ec][2][0:2], byteorder='big', signed=True) / 10.0 + dcBatteryVoltage = int.from_bytes(raw[b220101][0x7ec][2][2:4], byteorder='big', signed=False) / 10.0 + + data['EXTENDED'] = { + 'auxBatteryVoltage': raw[b220101][0x7ec][4][5] / 10.0, + 'batteryInletTemperature': int.from_bytes(raw[b220101][0x7ec][3][5:6], byteorder='big', signed=True), + 'batteryMaxTemperature': int.from_bytes(raw[b220101][0x7ec][2][4:5], byteorder='big', signed=True), + 'batteryMinTemperature': int.from_bytes(raw[b220101][0x7ec][2][5:6], byteorder='big', signed=True), + 'cumulativeEnergyCharged': int.from_bytes(raw[b220101][0x7ec][6][0:4], byteorder='big', signed=False) / 10.0, + 'cumulativeEnergyDischarged': int.from_bytes(raw[b220101][0x7ec][6][4:7] + raw[b220101][0x7ec][7][0:1], byteorder='big', signed=False) / 10.0, + 'charging': 1 if (chargingBits & 0xc) == 0x8 else 0, + 'normalChargePort': 1 if normalChargeBit and normalChargePort else 0, + 'rapidChargePort': 1 if normalChargeBit and not normalChargePort else 0, + 'dcBatteryCurrent': dcBatteryCurrent, + 'dcBatteryPower': dcBatteryCurrent * dcBatteryVoltage / 1000.0, + 'dcBatteryVoltage': dcBatteryVoltage, + 'soh': int.from_bytes(raw[b220105][0x7ec][4][1:3], byteorder='big', signed=False) / 10.0, } - data.update(self.getBaseData()) - return data def getBaseData(self): return { - "CAPACITY": 28, + "CAPACITY": 64, "SLOW_SPEED": 2.3, "NORMAL_SPEED": 4.6, "FAST_SPEED": 50 diff --git a/cars/NIRO_EV.py b/cars/NIRO_EV.py new file mode 100644 index 0000000..a743f06 --- /dev/null +++ b/cars/NIRO_EV.py @@ -0,0 +1,5 @@ +from KONA_EV import * + +class NIRO_EV(KONA_EV): + pass + diff --git a/cars/car.py b/cars/car.py new file mode 100644 index 0000000..4063149 --- /dev/null +++ b/cars/car.py @@ -0,0 +1,7 @@ + +class Car: + + class NULL_BLOCK(Exception): pass + class LOW_VOLTAGE(Exception): pass + + diff --git a/dongles/ELM327.py b/dongles/ELM327.py index 26a8c9a..a2340c1 100644 --- a/dongles/ELM327.py +++ b/dongles/ELM327.py @@ -1,32 +1,102 @@ from serial import Serial import logging from pexpect import fdpexpect +from binascii import hexlify class ELM327: + class CAN_ERROR(Exception): pass + class NO_DATA(Exception): pass + def __init__(self, dongle): + print("Init Dongle") self.serial = Serial(dongle['port'], baudrate=dongle['speed'], timeout=5) self.exp = fdpexpect.fdspawn(self.serial.fd) self.initDongle() def sendAtCmd(self, cmd, expect='OK'): - expect = bytes(expect, 'utf-8') cmd = bytes(cmd, 'utf-8') + expect = bytes(expect, 'utf-8') + try: + while self.serial.in_waiting: # Clear the input buffer + print("Stray data in buffer: " + \ + str(self.serial.read(self.serial.in_waiting))) + sleep(0.2) + self.exp.send(cmd + b'\r\n') + self.exp.expect('>') + ret = self.exp.before.strip(b'\r\n') + if expect not in ret: + raise Exception('Expected %s, got %s' % (expect,ret)) - logging.info('Write %s' % str(cmd)) - self.exp.send(cmd + b'\r\n') - self.exp.expect('>') - ret = self.exp.before.strip(b'\r\n') - logging.debug('Got %s' % str(cmd)) - if expect not in ret: - raise Exception('Expected %s, got %s' % (expect,ret)) + except pexpect.exceptions.TIMEOUT: + ret = b'TIMEOUT' - def sendCommand(self, cmd): + return ret.split(b"\r\n")[-1] + + expect = bytes(expect, 'utf-8') cmd = bytes(cmd, 'utf-8') - logging.info('Write %s' % str(cmd)) - self.exp.send(cmd + b'\r\n') - self.exp.expect('>') - return self.exp.before.strip(b'\r\n').split(b'\r\n') + + def sendCommand(self, cmd): + cmd = hexlify(cmd) + try: + while self.serial.in_waiting: # Clear the input buffer + print("Stray data in buffer: " + \ + str(self.serial.read(self.serial.in_waiting))) + sleep(0.2) + self.exp.send(cmd + b'\r\n') + self.exp.expect('>') + ret = self.exp.before.strip(b'\r\n') + except pexpect.exceptions.TIMEOUT: + ret = b'TIMEOUT' + + if ret in [b'NO DATA', b'DATA ERROR', b'ACT ALERT']: + raise ELM327.NO_DATA(ret) + elif ret in [b'BUFFER FULL', B'BUS BUSY', b'BUS ERROR', b'CAN ERROR', + b'ERR', b'FB ERROR', b'LP ALERT', b'LV RESET', b'STOPPED', + b'UNABLE TO CONNECT']: + raise ELM327.CAN_ERROR("Failed Command {}\n{}".format(cmd,ret)) + + try: + data = {} + raw = ret.split(b'\r\n') + lines = None + + for line in raw: + if len(line) != 19: + raise ValueError + + identifier = int(line[0:3],16) + frame_type = int(line[3:4],16) + + if frame_type == 0: # Single frame + idx = 0 + lines = 1 + elif frame_type == 1: # First frame + lines = math.ceil((int.from_bytes(bytes.fromhex(str(b'0' + line[4:7], 'ascii')), + byteorder='big', signed=False) + 1) / 7) + idx = 0 + if len(raw) != lines: + raise ValueError + elif frame_type == 2: # Consecutive frame + idx = int(line[4:5],16) + else: # Unexpected frame + raise ValueError + + if not identifier in data: + data[identifier] = [None] * lines + + data[identifier][idx] = bytes.fromhex(str(line[5:],'ascii')) + + # Check if all entries are filled + for key, val in data.items(): + for i in val: + if i == None: + raise ValueError + + except ValueError: + raise ELM327.CAN_ERROR("Failed Command {}\n{}".format(cmd,ret)) + + return data def initDongle(self): cmds = [['ATZ','OK'], @@ -49,12 +119,38 @@ def setProtocol(self, prot): else: raise Exception('Unsupported protocol %s' % prot) - def setIDFilter(self, filter): - self.sendAtCmd('ATCF' + str(filter)) + def setCanID(self, can_id): + if isinstance(can_id, bytes): + can_id = str(can_id) + elif isinstance(can_id, int): + can_id = format(can_id, 'X') + + self.sendAtCmd('ATTA' + can_id) + + def setIDFilter(self, id_filter): + if isinstance(id_filter, bytes): + id_filter = str(id_filter) + elif isinstance(id_filter, int): + id_filter = format(id_filter, 'X') + + self.sendAtCmd('ATCF' + id_filter) def setCANRxMask(self, mask): - self.sendAtCmd('ATCM' + str(mask)) + if isinstance(mask, bytes): + mask = str(mask) + elif isinstance(mask, int): + mask = format(mask, 'X') + + self.sendAtCmd('ATCM' + mask) def setCANRxFilter(self, addr): - self.sendAtCmd('ATCRA' + str(addr)) + if isinstance(addr, bytes): + addr = str(addr) + elif isinstance(addr, int): + addr = format(addr, 'X') + + self.sendAtCmd('ATCRA' + addr) + + def getObdVoltage(self): + return None diff --git a/dongles/PiOBD2Hat.py b/dongles/PiOBD2Hat.py index f2ae145..4873a36 100644 --- a/dongles/PiOBD2Hat.py +++ b/dongles/PiOBD2Hat.py @@ -1,6 +1,9 @@ from serial import Serial +from time import sleep import pexpect from pexpect import fdpexpect +from binascii import hexlify +import math class PiOBD2Hat: @@ -9,37 +12,47 @@ class NO_DATA(Exception): pass def __init__(self, dongle): print("Init Dongle") - self.serial = Serial(dongle['port'], baudrate=dongle['speed'], timeout=5) - self.exp = fdpexpect.fdspawn(self.serial.fd) + self.serial = Serial(dongle['port'], baudrate=dongle['speed']) + self.exp = fdpexpect.fdspawn(self.serial) self.initDongle() def sendAtCmd(self, cmd, expect='OK'): - expect = bytes(expect, 'utf-8') cmd = bytes(cmd, 'utf-8') - + expect = bytes(expect, 'utf-8') try: + while self.serial.in_waiting: # Clear the input buffer + print("Stray data in buffer: " + \ + str(self.serial.read(self.serial.in_waiting))) + sleep(0.2) self.exp.send(cmd + b'\r\n') - self.exp.expect('>', timeout=5) + self.exp.expect('>') ret = self.exp.before.strip(b'\r\n') - if expect not in ret: raise Exception('Expected %s, got %s' % (expect,ret)) except pexpect.exceptions.TIMEOUT: ret = b'TIMEOUT' + return ret.split(b"\r\n")[-1] + def sendCommand(self, cmd): - cmd = bytes(cmd, 'utf-8') - print("Send Command "+str(cmd)) + """ + @cmd: should be hex-encoded + """ + + cmd = hexlify(cmd) try: + while self.serial.in_waiting: # Clear the input buffer + print("Stray data in buffer: " + \ + str(self.serial.read(self.serial.in_waiting))) + sleep(0.2) self.exp.send(cmd + b'\r\n') - self.exp.expect('>', timeout=5) + self.exp.expect('>') ret = self.exp.before.strip(b'\r\n') - print(ret) except pexpect.exceptions.TIMEOUT: ret = b'TIMEOUT' - if ret in [b'NO DATA', b'TIMEOUT']: + if ret in [b'NO DATA', b'TIMEOUT', b'CAN NO ACK']: raise PiOBD2Hat.NO_DATA(ret) elif ret in [b'INPUT TIMEOUT', b'NO INPUT CHAR', b'UNKNOWN COMMAND', b'WRONG HEXCHAR COUNT', b'ILLEGAL COMMAND', b'SYNTAX ERROR', @@ -48,25 +61,58 @@ def sendCommand(self, cmd): b'NO ADDRESSBYTE', b'WRONG PROTOCOL', b'DATA ERROR', b'CHECKSUM ERROR', b'NO ANSWER', b'COLLISION DETECT', b'CAN NO ANSWER', b'PRTOTOCOL 8 OR 9 REQUIRED', - b'CAN ERROR', b'CAN NO ACK']: - raise PiOBD2Hat.CAN_ERROR(ret) + b'CAN ERROR']: + raise PiOBD2Hat.CAN_ERROR("Failed Command {}\n{}".format(cmd,ret)) try: - raw = {} - for line in ret.split(b'\r\n'): - raw[int(line[:5],16)] = bytes.fromhex(str(line[5:],'ascii')) + data = {} + raw = ret.split(b'\r\n') + lines = None + + for line in raw: + if len(line) != 19: + raise ValueError + + identifier = int(line[0:3],16) + frame_type = int(line[3:4],16) + + if frame_type == 0: # Single frame + idx = 0 + lines = 1 + elif frame_type == 1: # First frame + lines = math.ceil((int.from_bytes(bytes.fromhex(str(b'0' + line[4:7], 'ascii')), + byteorder='big', signed=False) + 1) / 7) + idx = 0 + if len(raw) != lines: + raise ValueError + elif frame_type == 2: # Consecutive frame + idx = int(line[4:5],16) + else: # Unexpected frame + raise ValueError + + if not identifier in data: + data[identifier] = [None] * lines + + data[identifier][idx] = bytes.fromhex(str(line[5:],'ascii')) + + # Check if all entries are filled + for key, val in data.items(): + for i in val: + if i == None: + raise ValueError + except ValueError: - raise PiOBD2Hat.CAN_ERROR(ret) + raise PiOBD2Hat.CAN_ERROR("Failed Command {}\n{}".format(cmd,ret)) - return raw + return data def initDongle(self): - cmds = [['ATZ','PI-OBD v1.0'], - ['ATE0','OK'], - ['ATL1','OK'], - ['ATOHS0','OK'], - ['ATH1','OK'], - ['ATSTFF','OK']] + cmds = [['ATRST','PI-OBD'], # Cold start + ['ATE0','OK'], # Disable echo + ['ATL1','OK'], # Use \r\n + ['ATOHS0','OK'], # Disable space between HEX bytes + ['ATH1','OK'], # Display header + ['ATST64','OK']] # Input timeout (10 sec) for c,r in cmds: self.sendAtCmd(c, r) @@ -78,16 +124,43 @@ def setAllowLongMessages(self, value): def setProtocol(self, prot): if prot == 'CAN_11_500': self.sendAtCmd('ATP6','6 = ISO 15765-4, CAN (11/500)') + self.sendAtCmd('ATONI1') # No init sequence else: raise Exception('Unsupported protocol %s' % prot) - def setIDFilter(self, filter): - self.sendAtCmd('ATSF' + str(filter)) + def setCanID(self, can_id): + if isinstance(can_id, bytes): + can_id = str(can_id) + elif isinstance(can_id, int): + can_id = format(can_id, 'X') + + self.sendAtCmd('ATCT' + can_id) + + def setIDFilter(self, id_filter): + if isinstance(id_filter, bytes): + id_filter = str(id_filter) + elif isinstance(id_filter, int): + id_filter = format(id_filter, 'X') + + self.sendAtCmd('ATSF' + id_filter) def setCANRxMask(self, mask): - self.sendAtCmd('ATCM' + str(mask)) + if isinstance(mask, bytes): + mask = str(mask) + elif isinstance(mask, int): + mask = format(mask, 'X') + + self.sendAtCmd('ATCM' + mask) def setCANRxFilter(self, addr): - self.sendAtCmd('ATCR' + str(addr)) + if isinstance(addr, bytes): + addr = str(addr) + elif isinstance(addr, int): + addr = format(addr, 'X') + + self.sendAtCmd('ATCR' + addr) + def getObdVoltage(self): + ret = self.sendAtCmd('AT!10','V') + return float(ret[:-1]) * 0.694 # strip the 'V' diff --git a/evnotipi.py b/evnotipi.py index 837de36..54c8434 100755 --- a/evnotipi.py +++ b/evnotipi.py @@ -14,6 +14,9 @@ LOOP_DELAY = 15 ABORT_NOTIFICATION_DELAY = 60 +POLL_THRESHOLD_VOLT = 13.0 + +class SKIP_POLL(Exception): pass # load config with open('config.json', encoding='utf-8') as config_file: @@ -74,25 +77,31 @@ abortNotificationSent = False last_charging = time() print("Notification threshold: {}".format(socThreshold)) + +# Set up signal handling def exit_gracefully(signum, frame): sys.exit(0) + signal.signal(signal.SIGTERM, exit_gracefully) + +car_off_skip_poll = False + +print("Starting main loop") try: while main_running: now = time() try: + if car_off_skip_poll: # Skip polling until car on voltage is detected again + if dongle.getObdVoltage() < POLL_THRESHOLD_VOLT: + raise SKIP_POLL + else: + print("Car on detected. Resume polling.") + car_off_skip_poll = False + data = car.getData() fix = gps.fix() - except DONGLE.CAN_ERROR as e: - print(e) - main_running = False - except DONGLE.NO_DATA as e: - print(e) - except: - raise - else: - print(data) + #print(data) try: EVNotify.setSOC(data['SOC_DISPLAY'], data['SOC_BMS']) currentSOC = data['SOC_DISPLAY'] or data['SOC_BMS'] @@ -146,8 +155,21 @@ def exit_gracefully(signum, frame): except EVNotify.CommunicationError as e: print(e) - except: - raise + + except DONGLE.CAN_ERROR as e: + print(e) + print("Stopping programm") + main_running = False + except DONGLE.NO_DATA as e: + print(e) + volt = dongle.getObdVoltage() + if volt and volt < POLL_THRESHOLD_VOLT: + print("Car off detected. Stop polling until car on.") + car_off_skip_poll = True + except SKIP_POLL: + pass + except CAR.NULL_BLOCK as e: + print(e) finally: try: @@ -162,12 +184,13 @@ def exit_gracefully(signum, frame): sys.stdout.flush() - if main_running: sleep(LOOP_DELAY) + if main_running: + loop_delay = LOOP_DELAY - (time()-now) + if loop_delay > 0: sleep(loop_delay) except (KeyboardInterrupt, SystemExit): #when you press ctrl+c main_running = False -except: - raise + finally: print("Exiting ...") gps.stop() diff --git a/gpspoller.py b/gpspoller.py index f6aba22..4a33f85 100644 --- a/gpspoller.py +++ b/gpspoller.py @@ -2,6 +2,7 @@ sys.path.append('/usr/local/lib/python3.7/site-packages') import gps import threading +from time import time,sleep class GpsPoller(threading.Thread): def __init__(self):