wn/MeterbusLib
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

some more tests

Browse Source
This commit is contained in:
hg
2015-06-12 16:02:49 +02:00
parent 88e36b0170
commit c3a7c19d33
3 changed files with 65 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
MeterbusLib.py
View File

@ -97,7 +97,7 @@ class LongFrame(ControlFrame):
def __init__(self, telegram, devices=[]): def __init__(self, telegram, devices=[]):
super(LongFrame, self).__init__(telegram) super(LongFrame, self).__init__(telegram)
self.devices = devices self.devices = devices
if self.devices != None: if self.devices is not None:
for device in self.devices: for device in self.devices:
if not isinstance(device, Device): if not isinstance(device, Device):
raise InvalidDevicesStructureException() raise InvalidDevicesStructureException()
@ -307,7 +307,7 @@ class LongFrame(ControlFrame):
# print vifeCode # print vifeCode
(mask, value, descr, unit, factorMask, factorFunc) = vifeCode (mask, value, descr, unit, factorMask, factorFunc) = vifeCode
if (vife & mask) == value: if (vife & mask) == value:
if factorFunc == None: if factorFunc is None:
retFactor = 1 retFactor = 1
else: else:
factorHelper = vife & factorMask factorHelper = vife & factorMask
@ -339,7 +339,7 @@ class LongFrame(ControlFrame):
else: else:
retDescr = descr retDescr = descr
retUnit = unit retUnit = unit
if factorFunc != None: if factorFunc is not None:
factorHelper = vif & factorMask factorHelper = vif & factorMask
retFactor = factorFunc(factorHelper) retFactor = factorFunc(factorHelper)
break break
@ -379,7 +379,7 @@ class LongFrame(ControlFrame):
self.consumed += userDataLength self.consumed += userDataLength
dataConvFunc = LongFrame.DataInformationBlock.DATA_FIELD_CODES[self.dif & 0x0f][2] dataConvFunc = LongFrame.DataInformationBlock.DATA_FIELD_CODES[self.dif & 0x0f][2]
if dataConvFunc == None: if dataConvFunc is None:
self.rawValue = 0 self.rawValue = 0
else: else:
self.rawValue = dataConvFunc(self.userData) self.rawValue = dataConvFunc(self.userData)
@ -413,10 +413,10 @@ class LongFrame(ControlFrame):
device = None device = None
for d in self.devices: for d in self.devices:
if d.address == self.address: if d.address == self.address:
print("device found") #print("device found")
device = d device = d
break break
if device != None: if device is not None:
self.comment = device.comment self.comment = device.comment
else: else:
self.comment = '-' self.comment = '-'
@ -425,8 +425,11 @@ class LongFrame(ControlFrame):
dibIndex = 0 dibIndex = 0
#print("Telegram length: %d" % len(self.telegram)) #print("Telegram length: %d" % len(self.telegram))
while True: while True:
if device != None: if device is not None:
try:
comment = device.dataItems[dibIndex] comment = device.dataItems[dibIndex]
except IndexError:
raise DeviceItemsMismatchException()
else: else:
comment = '-' comment = '-'
dibIndex += 1 dibIndex += 1

69
test1.py
View File

@ -10,8 +10,29 @@ import MeterbusTypeConversion
import MeterbusLibExceptions import MeterbusLibExceptions
class TestFrameParsing(unittest.TestCase): class TestCaseWFloat(unittest.TestCase):
def assertBetween(self, v, bottom, top):
self.assertTrue(v > bottom and v < top )
class TestFrameParsing(TestCaseWFloat):
def setUp(self): def setUp(self):
#devices
self.devices = []
device_dishwasher_electric = MeterbusLib.Device(0x53, "Dishwasher", ["Energy total", "Energy partial", "Voltage", "Current", "Power", "img. Power"])
self.devices.append(device_dishwasher_electric)
device_3phase_electric = MeterbusLib.Device(0x50, "3 Phase Electric", ["Energy T1 total", "Energy T1 partial", "Energy T2 total", "Energy T2 partial",
"Voltage phase 1", "Current phase 1", "Power phase 1", "img. Power phase 1",
"Voltage phase 2", "Current phase 2", "Power phase 2", "img. Power phase 2",
"Voltage phase 3", "Current phase 3", "Power phase 3", "img. Power phase 3",
"converter ratio", "Power total", "img. Power total", "tariff"
])
self.devices.append(device_3phase_electric)
self.brokenDevices = []
device_dishwasher_electric = MeterbusLib.Device(0x53, "Dishwasher", ["Energy total", "Energy partial", "Voltage", "Current", "Power"])
self.brokenDevices.append(device_dishwasher_electric)
# dishwasher, electric # dishwasher, electric
self.inputOkLongFrame_1phase_electric = "68 38 38 68 08 53 72 17 00 13 00 2E 19 24 02 D6 00 00 00 8C 10 04 01 02 00 00 8C 11 04 01 02 00 00 02 FD C9 FF 01 E4 00 02 FD DB FF 01 03 00 02 AC FF 01 01 00 82 40 AC FF 01 FA FF 20 16" self.inputOkLongFrame_1phase_electric = "68 38 38 68 08 53 72 17 00 13 00 2E 19 24 02 D6 00 00 00 8C 10 04 01 02 00 00 8C 11 04 01 02 00 00 02 FD C9 FF 01 E4 00 02 FD DB FF 01 03 00 02 AC FF 01 01 00 82 40 AC FF 01 FA FF 20 16"
self.inputNOkLongFrame_1phase_electric_wrong_medium = "68 38 38 68 08 53 72 17 00 13 00 2E 19 24 FE D6 00 00 00 8C 10 04 01 02 00 00 8C 11 04 01 02 00 00 02 FD C9 FF 01 E4 00 02 FD DB FF 01 03 00 02 AC FF 01 01 00 82 40 AC FF 01 FA FF 1c 16" self.inputNOkLongFrame_1phase_electric_wrong_medium = "68 38 38 68 08 53 72 17 00 13 00 2E 19 24 FE D6 00 00 00 8C 10 04 01 02 00 00 8C 11 04 01 02 00 00 02 FD C9 FF 01 E4 00 02 FD DB FF 01 03 00 02 AC FF 01 01 00 82 40 AC FF 01 FA FF 1c 16"
@ -46,7 +67,7 @@ class TestFrameParsing(unittest.TestCase):
def test_OK_LongFrame_1phase_electric(self): def test_OK_LongFrame_1phase_electric(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkLongFrame_1phase_electric) telegram.fromHexString(self.inputOkLongFrame_1phase_electric)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame) self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame)
@ -57,22 +78,36 @@ class TestFrameParsing(unittest.TestCase):
self.assertEqual(telegram.frame.fixedDataHeader.manufacturer, "FIN") self.assertEqual(telegram.frame.fixedDataHeader.manufacturer, "FIN")
self.assertEqual(telegram.frame.fixedDataHeader.version, 36) self.assertEqual(telegram.frame.fixedDataHeader.version, 36)
self.assertEqual(telegram.frame.fixedDataHeader.medium, "Electrity") self.assertEqual(telegram.frame.fixedDataHeader.medium, "Electrity")
self.assertEqual(telegram.frame.comment, "Dishwasher")
self.assertEqual(telegram.frame.dib[0].comment, "Energy total")
self.assertEqual(telegram.frame.dib[0].rawValue, 201)
self.assertEqual(telegram.frame.dib[0].value, 2010)
self.assertEqual(telegram.frame.dib[3].comment, "Current")
self.assertEqual(telegram.frame.dib[3].rawValue, 3)
v = telegram.frame.dib[3].value
self.assertBetween(v, 0.2999, 0.3001)
def test_OK_LongFrame_1phase_electric_wrong_medium(self): def test_OK_LongFrame_1phase_electric_wrong_medium(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOkLongFrame_1phase_electric_wrong_medium) telegram.fromHexString(self.inputNOkLongFrame_1phase_electric_wrong_medium)
with self.assertRaises(MeterbusLibExceptions.MediumConversionException): with self.assertRaises(MeterbusLibExceptions.MediumConversionException):
telegram.parse() telegram.parse()
def test_OK_LongFrame_1phase_electric_wrong_signature(self): def test_OK_LongFrame_1phase_electric_wrong_signature(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOkLongFrame_1phase_electric_wrong_signature) telegram.fromHexString(self.inputNOkLongFrame_1phase_electric_wrong_signature)
with self.assertRaises(MeterbusLibExceptions.InvalidFrameException): with self.assertRaises(MeterbusLibExceptions.InvalidFrameException):
telegram.parse() telegram.parse()
def test_OK_LongFrame_1phase_electric_device_items_wrong(self):
telegram = MeterbusLib.Telegram(self.brokenDevices)
telegram.fromHexString(self.inputOkLongFrame_1phase_electric)
with self.assertRaises(MeterbusLibExceptions.DeviceItemsMismatchException):
telegram.parse()
def test_OK_LongFrame_3phase_electric(self): def test_OK_LongFrame_3phase_electric(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkLongFrame_3phase_electric) telegram.fromHexString(self.inputOkLongFrame_3phase_electric)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame) self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame)
@ -83,9 +118,11 @@ class TestFrameParsing(unittest.TestCase):
self.assertEqual(telegram.frame.fixedDataHeader.manufacturer, "FIN") self.assertEqual(telegram.frame.fixedDataHeader.manufacturer, "FIN")
self.assertEqual(telegram.frame.fixedDataHeader.version, 22) self.assertEqual(telegram.frame.fixedDataHeader.version, 22)
self.assertEqual(telegram.frame.fixedDataHeader.medium, "Electrity") self.assertEqual(telegram.frame.fixedDataHeader.medium, "Electrity")
self.assertEqual(telegram.frame.comment, "3 Phase Electric")
def test_OK_LongFrame_water(self): def test_OK_LongFrame_water(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkLongFrame_water) telegram.fromHexString(self.inputOkLongFrame_water)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame) self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame)
@ -99,7 +136,7 @@ class TestFrameParsing(unittest.TestCase):
def test_OK_LongFrame_gas(self): def test_OK_LongFrame_gas(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkLongFrame_gas) telegram.fromHexString(self.inputOkLongFrame_gas)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame) self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame)
@ -112,7 +149,7 @@ class TestFrameParsing(unittest.TestCase):
self.assertEqual(telegram.frame.fixedDataHeader.medium, "Gas") self.assertEqual(telegram.frame.fixedDataHeader.medium, "Gas")
def test_OK_LongFrame_thermometer(self): def test_OK_LongFrame_thermometer(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkLongFrame_thermometer) telegram.fromHexString(self.inputOkLongFrame_thermometer)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame) self.assertIsInstance(telegram.frame, MeterbusLib.LongFrame)
@ -125,13 +162,13 @@ class TestFrameParsing(unittest.TestCase):
self.assertEqual(telegram.frame.fixedDataHeader.medium, "Other") self.assertEqual(telegram.frame.fixedDataHeader.medium, "Other")
def test_OK_SingleCharacter(self): def test_OK_SingleCharacter(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkSingleCharacter) telegram.fromHexString(self.inputOkSingleCharacter)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.SingleCharacter) self.assertIsInstance(telegram.frame, MeterbusLib.SingleCharacter)
def test_OK_Shortframe(self): def test_OK_Shortframe(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOkShortframe) telegram.fromHexString(self.inputOkShortframe)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.ShortFrame) self.assertIsInstance(telegram.frame, MeterbusLib.ShortFrame)
@ -139,7 +176,7 @@ class TestFrameParsing(unittest.TestCase):
self.assertEqual(telegram.frame.address, 2); self.assertEqual(telegram.frame.address, 2);
def test_OK_Controlframe(self): def test_OK_Controlframe(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputOKControlframe) telegram.fromHexString(self.inputOKControlframe)
telegram.parse() telegram.parse()
self.assertIsInstance(telegram.frame, MeterbusLib.ControlFrame) self.assertIsInstance(telegram.frame, MeterbusLib.ControlFrame)
@ -148,31 +185,31 @@ class TestFrameParsing(unittest.TestCase):
self.assertEqual(telegram.frame.ciField, 3); self.assertEqual(telegram.frame.ciField, 3);
def test_NOk_Shortframe_checksum_failure(self): def test_NOk_Shortframe_checksum_failure(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOkShortframe_checksum_failure) telegram.fromHexString(self.inputNOkShortframe_checksum_failure)
with self.assertRaises(MeterbusLibExceptions.InvalidChecksumException): with self.assertRaises(MeterbusLibExceptions.InvalidChecksumException):
telegram.parse() telegram.parse()
def test_NOk_LongFrame_thermometer_checksum_failure(self): def test_NOk_LongFrame_thermometer_checksum_failure(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOkLongFrame_thermometer_checksum_failure) telegram.fromHexString(self.inputNOkLongFrame_thermometer_checksum_failure)
with self.assertRaises(MeterbusLibExceptions.InvalidChecksumException): with self.assertRaises(MeterbusLibExceptions.InvalidChecksumException):
telegram.parse() telegram.parse()
def test_NOk_Controlframe_wrong_stopcode(self): def test_NOk_Controlframe_wrong_stopcode(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOKControlframe_wrong_stopcode) telegram.fromHexString(self.inputNOKControlframe_wrong_stopcode)
with self.assertRaises(MeterbusLibExceptions.InvalidStopCharException): with self.assertRaises(MeterbusLibExceptions.InvalidStopCharException):
telegram.parse() telegram.parse()
def test_NOk_Controlframe_wrong_secondlength(self): def test_NOk_Controlframe_wrong_secondlength(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOKControlframe_wrong_secondlength) telegram.fromHexString(self.inputNOKControlframe_wrong_secondlength)
with self.assertRaises(MeterbusLibExceptions.InvalidSecondLengthException): with self.assertRaises(MeterbusLibExceptions.InvalidSecondLengthException):
telegram.parse() telegram.parse()
def test_NOk_wrong_startcode(self): def test_NOk_wrong_startcode(self):
telegram = MeterbusLib.Telegram() telegram = MeterbusLib.Telegram(self.devices)
telegram.fromHexString(self.inputNOk_wrong_startcode) telegram.fromHexString(self.inputNOk_wrong_startcode)
with self.assertRaises(MeterbusLibExceptions.InvalidStartCharException): with self.assertRaises(MeterbusLibExceptions.InvalidStartCharException):
telegram.parse() telegram.parse()

2
test2.py
View File

@ -30,7 +30,7 @@ inputOkLongFrame_water = "68 46 46 68 08 30 72 45 71 43 00 24 23 25 07 F6 00 00
inputOkLongFrame_gas = "68 56 56 68 08 40 72 43 60 52 00 77 04 14 03 FF 10 00 00 0C 78 76 03 01 10 0D 7C 08 44 49 20 2E 74 73 75 63 0A 30 30 30 30 30 30 30 30 30 30 04 6D 06 0E EB 16 02 7C 09 65 6D 69 74 20 2E 74 61 62 A3 09 04 13 98 AE 04 00 04 93 7F 4E 01 00 00 44 13 FC A5 04 00 0F 01 00 1F 51 16" inputOkLongFrame_gas = "68 56 56 68 08 40 72 43 60 52 00 77 04 14 03 FF 10 00 00 0C 78 76 03 01 10 0D 7C 08 44 49 20 2E 74 73 75 63 0A 30 30 30 30 30 30 30 30 30 30 04 6D 06 0E EB 16 02 7C 09 65 6D 69 74 20 2E 74 61 62 A3 09 04 13 98 AE 04 00 04 93 7F 4E 01 00 00 44 13 FC A5 04 00 0F 01 00 1F 51 16"
telegram = MeterbusLib.Telegram(devices) telegram = MeterbusLib.Telegram(devices)
telegram.fromHexString(inputOkLongFrame_thermometer) telegram.fromHexString(inputOkLongFrame_1phase_electric)
telegram.parse() telegram.parse()
print(json.dumps(telegram.getJSON(), indent=2)) print(json.dumps(telegram.getJSON(), indent=2))