diff options
| -rw-r--r-- | ebustest.py | 142 |
1 files changed, 87 insertions, 55 deletions
diff --git a/ebustest.py b/ebustest.py index 767ed9c..c9531eb 100644 --- a/ebustest.py +++ b/ebustest.py @@ -41,6 +41,23 @@ deviceDescription = [ {'address':0xfe, 'type':'broadcast', 'description':'Broadcast'}, ] +class fields: + class DataField(object): + def __init__(self, offset): + self.offset = offset + def value(self,payload): + raise NotImplemented() + + class Data2c(DataField): + def value(self,payload): + highByte = ord(payload[self.offset+1]) + lowByte= ord(payload[self.offset]) + return highByte*16 + (lowByte>>4) + (lowByte&0xf)/16.0 + + class Bit(DataField): + def value(self, payload): + return ord(payload[self.offset]) == 0x1 + packetDescription = [ # Service 0x05 (Brennersteuerbefehle) {'primary':0x5, 'secondary':0x3, 'name':'Betriebsdaten des Feuerungsautomaten an den Regler Block1'}, @@ -58,8 +75,10 @@ packetDescription = [ #p[3] = 10^4 #p[4] = 10^6 {'primary':0x3, 'secondary':0x8, 'name':'Gesamtbrennstoffmengenzähle lesen'}, - {'primary':''}, + {'primary':0x50, 'secondary':0x17, 'name':'Solar Daten', 'format':{ + 'solarPumpe':fields.Bit(0), 'tempKollektor':fields.Data2c(2), 'tempWarmwasserSolar':fields.Data2c(4)} }, ] + def formatHex(data): return " ".join(map(lambda byte: "%.2x"%ord(byte), data)) @@ -78,44 +97,83 @@ class EbusPacket(object): self.secondary_command = secondary_command def name(self): - desc = filter(lambda p: p['primary'] == self.primary_command \ - and p['secondary'] == self.secondary_command, packetDescription) - if len(desc) > 0: - return desc[0]['name'] - else: - return "UNKNOWN" + if self.description(): + return self.description()['name'] def __str__(self): - return "<%-18s name=\"%-15s\" source=\"%s\" destination=\"%s\" primary=0x%x secondary=0x%x length=0x%x>" % \ + return "<%-18s name=\"%-15s\" source=\"%s\" destination=\"%s\" primary=0x%x secondary=0x%x length=0x%x %s>" % \ (self.__class__.__name__, self.name(), getDsc(self.source), getDsc(self.destination), \ - self.primary_command, self.secondary_command, self.length) + self.primary_command, self.secondary_command, self.length, \ + " ".join(map(lambda name: "%s=%s" % (name, self.values()[name]),self.values()))) -class EbusService5017(EbusPacket): - # Contains Solars Pumpe and Kollektor Temperatur + def description(self): + desc = filter(lambda p: p['primary'] == self.primary_command \ + and p['secondary'] == self.secondary_command, packetDescription) + if len(desc) > 0: + return desc[0] + else: + return None + def values(self): + desc = self.description() + if not desc or not desc.has_key('format'): + return dict() + return dict( map(lambda name: (name, desc['format'][name].value(self.data) ), desc['format'].keys()) ) class EbusMasterMaster(EbusPacket): def __init__(self, source, destination, primary_command, secondary_command, data): EbusPacket.__init__(self, source, destination, primary_command, secondary_command) - self.length = len(data) - self.data = data - - + self.length = len(data) + self.data = data class EbusMasterSlave(EbusPacket): def __init__(self, source, destination, primary_command, secondary_command, request, response): EbusPacket.__init__(self, source, destination, primary_command, secondary_command) self.length = len(request) - self.request = request + self.request = request self.response = response class EbusBroadcast(EbusPacket): def __init__(self, source, destination, primary_command, secondary_command, data): EbusPacket.__init__(self, source, destination, primary_command, secondary_command) self.length = len(data) - self.data = data + self.data = data + +class EbusService(): + def __init__(self,p): + self.p = p + def convertData2c(data): - + # low nibble x&0xf + # high nibble x>>4 + """Beispiel für die Berechnung: + if ((x & 8000h) == 8000h) // y negativ + y = - [dez(High_Byte(!x)) 16 + dez(High_Nibble (Low_Byte (!x))) + + (dez(Low_Nibble (Low_Byte (!x))) +1 ) / 16] + else // y positiv + y = dez(High_Byte(x)) 16 + dez(High_ Nibble (Low Byte (x))) + + dez(Low_ Nibble (Low Byte (x))) / 16 + """ + #print "%s" % (formatHex(data)) + highByte = ord(data[1]) + lowByte = ord(data[0]) + + if (0): + #return (!highByte)*16 + (!lowByte)>>4 + (((!lowByte)&0xf)+1)/16.0 + return 0 + else: + return highByte*16 + (lowByte>>4) + (lowByte&0xf)/16.0 + +class EbusService5017(EbusService): + def __init__(self,p): + EbusPacket.__init__(self,p) + self.parsePayload() + def parsePayload(self): + self.solar_pumpe = ord(p.payload[0]) + self.collektor_temperature = convertData2c(p.payload[2:4]) + self.warmwasser_temperature = convertData2c(p.payload[4:6]) + def __str__(self): + return "Solarpumpe: %d Kollektor: %.1f°C Solarwarmwaser %.1f°C" % (self.solar_pumpe, self.collektor_temperature, self.warmwater_temperature) class Fetcher(asynchat.async_chat): def __init__(self): @@ -162,55 +220,29 @@ class Fetcher(asynchat.async_chat): secondaryCommand = ord(data[3]) payloadLength = ord(data[4]) payload = data[5:6+payloadLength] - #delete 0x50 packets for devel - #if (primaryCommand == ord("\x50")): -# return - #print "PR SC NN D0 D1 D2 D3 D4 D5 D6 D7 ..." - #print "%.2x" % (primaryCommand,) - #print "%.2x %.2x %.2x %s" % (primaryCommand,secondaryCommand,payloadLength,"bla") - #print "%.2x %.2x %.2x %s" % (primaryCommand,secondaryCommand,payloadLength,formatHex(payload)) + #delete 0x50 packets for devel + #if (primaryCommand == ord("\x50")): +# return + #print "PR SC NN D0 D1 D2 D3 D4 D5 D6 D7 ..." + #print "%.2x" % (primaryCommand,) + #print "%.2x %.2x %.2x %s" % (primaryCommand,secondaryCommand,payloadLength,"bla") + #print "%.2x %.2x %.2x %s" % (primaryCommand,secondaryCommand,payloadLength,formatHex(payload)) p = None if sourceDevice[0]['type'] == 'master' and destinationDevice[0]['type'] == 'master': p = EbusMasterMaster(source, destination, primaryCommand, secondaryCommand, payload) - print p - print "payload=%s" % formatHex(payload) + #print p + #print "payload=%s" % formatHex(payload) elif sourceDevice[0]['type'] == 'master' and destinationDevice[0]['type'] == 'slave': p = EbusMasterSlave(source, destination, primaryCommand, secondaryCommand, payload, None) - print p - print "payload=%s" % formatHex(payload) + #print p + ###print "payload=%s" % formatHex(payload) elif sourceDevice[0]['type'] == 'master' and destinationDevice[0]['type'] == 'broadcast': p = EbusBroadcast(source, destination, primaryCommand, secondaryCommand, payload) print p - print "payload=%s" % formatHex(payload) - if (primaryCommand == 0x50 and secondaryCommand == 0x17): - print "Kollektor Temperatur: %s %f °C" % (formatHex(payload[2:4]),self.convertData2c(payload[2:4])) else: print "KOMISCHES ZEUG" return - print "" - - def convertData2c(self,data): - - # low nibble x&0xf - # high nibble x>>4 - """Beispiel für die Berechnung: - if ((x & 8000h) == 8000h) // y negativ - y = - [dez(High_Byte(!x)) 16 + dez(High_Nibble (Low_Byte (!x))) - + (dez(Low_Nibble (Low_Byte (!x))) +1 ) / 16] - else // y positiv - y = dez(High_Byte(x)) 16 + dez(High_ Nibble (Low Byte (x))) - + dez(Low_ Nibble (Low Byte (x))) / 16 - """ - print "%s" % (formatHex(data)) - highByte = ord(data[1]) - lowByte = ord(data[0]) - - if (0): - #return (!highByte)*16 + (!lowByte)>>4 + (((!lowByte)&0xf)+1)/16.0 - return 0 - else: - return highByte*16 + (lowByte>>4) + (lowByte&0xf)/16.0 Fetcher() asyncore.loop() |