Code: Select all
#!/usr/local/bin/python
from __future__ import print_function
import serial
#######################################################################
# These are the variables I have managed to identify
# Submissions welcome.
kamstrup_402_var = {
0x003C: "Energy",
0x0050: "Power",
0x0056: "Temp1",
0x0057: "Temp2",
0x0059: "Tempdiff",
0x004A: "Flow",
0x0044: "Volume",
0x008D: "MinFlow_M",
0x008B: "MaxFlow_M",
0x008C: "MinFlowDate_M",
0x008A: "MaxFlowDate_M",
0x0091: "MinPower_M",
0x008F: "MaxPower_M",
0x0095: "AvgTemp1_M",
0x0096: "AvgTemp2_M",
0x0090: "MinPowerDate_M",
0x008E: "MaxPowerDate_M",
0x007E: "MinFlow_Y",
0x007C: "MaxFlow_Y",
0x007D: "MinFlowDate_Y",
0x007B: "MaxFlowDate_Y",
0x0082: "MinPower_Y",
0x0080: "MaxPower_Y",
0x0092: "AvgTemp1_Y",
0x0093: "AvgTemp2_Y",
0x0081: "MinPowerDate_Y",
0x007F: "MaxPowerDate_Y",
0x0061: "Temp1xm3",
0x006E: "Temp2xm3",
0x0071: "Infoevent",
0x03EC: "HourCounter",
}
#######################################################################
# Kamstrup uses the "true" CCITT CRC-16
#
def crc_1021(message):
poly = 0x1021
reg = 0x0000
for byte in message:
mask = 0x80
while(mask > 0):
reg<<=1
if byte & mask:
reg |= 1
mask>>=1
if reg & 0x10000:
reg &= 0xffff
reg ^= poly
return reg
#######################################################################
# Byte values which must be escaped before transmission
#
escapes = {
0x06: True,
0x0d: True,
0x1b: True,
0x40: True,
0x80: True,
}
#######################################################################
# And here we go....
#
class kamstrup(object):
def __init__(self, serial_port = "/dev/ttyUSB0"):
self.debug_fd = open("kamstrup", "wb")
self.debug_fd.write("\n\nStart\n")
self.debug_id = None
self.ser = serial.Serial(
port = serial_port,
baudrate = 2400,
timeout = 20,
bytesize = serial.EIGHTBITS,
parity = serial.PARITY_NONE,
stopbits = serial.STOPBITS_TWO)
# xonxoff = 0,
# rtscts = 0)
# timeout = 20
def debug(self, dir, b):
for i in b:
if dir != self.debug_id:
if self.debug_id != None:
self.debug_fd.write("\n")
self.debug_fd.write(dir + "\t")
self.debug_id = dir
self.debug_fd.write(" %02x " % i)
self.debug_fd.flush()
def debug_msg(self, msg):
if self.debug_id != None:
self.debug_fd.write("\n")
self.debug_id = "Msg"
self.debug_fd.write("Msg\t" + msg)
self.debug_fd.flush()
def wr(self, b):
b = bytearray(b)
self.debug("Wr", b);
self.ser.write(b)
def rd(self):
a = self.ser.read(1)
if len(a) == 0:
self.debug_msg("Rx Timeout")
return None
b = bytearray(a)[0]
self.debug("Rd", bytearray((b,)));
return b
def send(self, pfx, msg):
b = bytearray(msg)
b.append(0)
b.append(0)
c = crc_1021(b)
b[-2] = c >> 8
b[-1] = c & 0xff
c = bytearray()
c.append(pfx)
for i in b:
if i in escapes:
c.append(0x1b)
c.append(i ^ 0xff)
else:
c.append(i)
c.append(0x0d)
self.wr(c)
def recv(self):
b = bytearray()
while True:
d = self.rd()
if d == None:
return None
if d == 0x40:
b = bytearray()
b.append(d)
if d == 0x0d:
#break
c = bytearray()
i = 1;
while i < len(b) - 1:
if b[i] == 0x1b:
v = b[i + 1] ^ 0xff
if v not in escapes:
self.debug_msg(
"Missing Escape %02x" % v)
c.append(v)
i += 2
else:
c.append(b[i])
i += 1
if crc_1021(c):
self.debug_msg("CRC error")
return c[:-2]
def readvar(self, nbr):
self.send(0x80, (0x3f, 0x10, 0x01, nbr >> 8, nbr & 0xff))
b = self.recv()
if b == None:
return b
if b[0] != 0x3f or b[1] != 0x10:
return None
if b[2] != nbr >> 8 or b[3] != nbr & 0xff:
return None
x = 0
for i in b[7:]:
x <<= 8
x |= i
s = ""
for i in b[:4]:
s += " %02x" % i
s += " |"
for i in b[4:7]:
s += " %02x" % i
s += " |"
for i in b[7:]:
s += " %02x" % i
decimals = b[6] & 0x0f
while decimals > 0:
x *= .1
decimals -= 1
#print(s, "=", x)
return x
if __name__ == "__main__":
import time
foo = kamstrup()
for i in kamstrup_402_var:
x = foo.readvar(i)
#print("%-25s" % kamstrup_402_var[i], x)
print(kamstrup_402_var[i]+":", x)
