diff --git a/src/energenie/Devices.py b/src/energenie/Devices.py index 668e4c9..e2b99af 100644 --- a/src/energenie/Devices.py +++ b/src/energenie/Devices.py @@ -276,6 +276,7 @@ class Capabilities(): pass self.capabilities = Capabilities() self.updated_cb = None + self.rxseq = 0 def get_config(self): raise RuntimeError("There is no configuration for a base Device") @@ -353,9 +354,13 @@ """An estimate of the next time we expect a message from this device""" pass + def get_receive_count(self): + return self.rxseq + def incoming_message(self, payload): """Entry point for a message to be processed""" #This is the base-class entry point, don't override this, but override handle_message + self.rxseq += 1 self.handle_message(payload) if self.updated_cb != None: self.updated_cb(self, payload) diff --git a/src/energenie/Devices_test.py b/src/energenie/Devices_test.py index 6e437df..456b0b8 100644 --- a/src/energenie/Devices_test.py +++ b/src/energenie/Devices_test.py @@ -2,43 +2,42 @@ # # Test harness for Devices module -#TODO: Turn into unittest.TestCase - - import time -from Devices import * +import Devices +import unittest +from lifecycle import * -# cooperative loop could be energenie_radio.loop() -# or wrap a thread around it with start() but beware of thread context -# and thread safety. +class TestDevices(unittest.TestCase): -import radio -radio.DEBUG = True + @test_1 + def test_without_registry(self): + """A simple on/off test with some devices from the device factory""" + tv = Devices.DeviceFactory.get_device_from_name("GreenButton", device_id=(0xC8C8C, 1)) + fan = Devices.DeviceFactory.get_device_from_name("AdaptorPlus", device_id=0x68b) + xbox = Devices.DeviceFactory.get_device_from_id(Devices.PRODUCTID_MIHO005, device_id=10) -def test_without_registry(): - - tv = DeviceFactory.get_device_from_name("GreenButton", device_id=(0xC8C8C, 1)) - fan = DeviceFactory.get_device_from_name("AdaptorPlus", device_id=0x68b) - xbox = DeviceFactory.get_device_from_id(PRODUCTID_MIHO005, device_id=10) - - while True: print("ON") tv.turn_on() fan.turn_off() xbox.turn_off() - time.sleep(2) print("OFF") tv.turn_off() fan.turn_on() xbox.turn_on() - time.sleep(1) + + +def init(): + """Start the Energenie system running""" + import OpenThings, radio + radio.DEBUG = True + radio.init() + OpenThings.init(Devices.CRYPT_PID) if __name__ == "__main__": - import OpenThings, Devices - OpenThings.init(Devices.CRYPT_PID) - test_without_registry() + init() + unittest.main() # END