From 7e96d60ed1bbd638b0bb93c9ce4a3e2af39dd6cd Mon Sep 17 00:00:00 2001 From: tkl Date: Tue, 20 Dec 2016 22:09:01 +0100 Subject: [PATCH] math helper functions started --- src/communiate/protocol.py | 6 +- src/data_buffer/__init__.py | 25 +++ .../test}/__init__.py | 0 .../test/test_data_buffer.py} | 12 +- src/engine/__init__.py | 24 +++ src/engine/engine.py | 24 --- src/gate/gate_handler.py | 32 +--- src/gate/gate_state.py | 171 +++++++++++++++++- src/gate_guard.py | 75 ++------ src/gpio/__init__.py | 32 ++++ src/gpio/gpio.py | 25 --- src/light_data/light_data.py | 20 -- src/light_data/test/__init__.py | 0 src/light_sensor/light_sensor.py | 8 +- src/math/__init__.py | 0 src/math/slope.py | 15 -- src/mymath/__init__.py | 13 ++ src/mymath/unittest/__init__.py | 29 +++ src/power_sensor/__init__.py | 49 +++++ src/power_sensor/power_sensor.py | 37 ---- 20 files changed, 371 insertions(+), 226 deletions(-) create mode 100644 src/data_buffer/__init__.py rename src/{light_data => data_buffer/test}/__init__.py (100%) rename src/{light_data/test/test_light_data.py => data_buffer/test/test_data_buffer.py} (88%) delete mode 100644 src/engine/engine.py delete mode 100644 src/gpio/gpio.py delete mode 100644 src/light_data/light_data.py delete mode 100644 src/light_data/test/__init__.py delete mode 100644 src/math/__init__.py delete mode 100644 src/math/slope.py create mode 100644 src/mymath/__init__.py create mode 100644 src/mymath/unittest/__init__.py delete mode 100755 src/power_sensor/power_sensor.py diff --git a/src/communiate/protocol.py b/src/communiate/protocol.py index f71de5f..f211f4b 100644 --- a/src/communiate/protocol.py +++ b/src/communiate/protocol.py @@ -31,9 +31,13 @@ class mqtt: return False def transmit(self, topic, payload): + was_connected = True if not self.__is_connected: - return False + was_connected = False + self.connect() result = self.__client.publish(topic, payload, self.__qos, self.__retain) + if not was_connected: + self.disconnect() if result == 0: return True return False diff --git a/src/data_buffer/__init__.py b/src/data_buffer/__init__.py new file mode 100644 index 0000000..f78c802 --- /dev/null +++ b/src/data_buffer/__init__.py @@ -0,0 +1,25 @@ +import mymath + +class DataBuffer(object): + def __init__(self, length): + self.__max = length + self.__data = [] + + def push(self, element): + if self.__max == 0: + return False + if len(self.__data) == self.__max: + _ = self.__data.pop(0) + self.__data.append(element) + return True + + def average(self): + if len(self.__data) != self.__max: + return None + return mymath.mean(self.__data) + + def length(self): + return len(self.__data) + + def clear(self): + self.__data = [] diff --git a/src/light_data/__init__.py b/src/data_buffer/test/__init__.py similarity index 100% rename from src/light_data/__init__.py rename to src/data_buffer/test/__init__.py diff --git a/src/light_data/test/test_light_data.py b/src/data_buffer/test/test_data_buffer.py similarity index 88% rename from src/light_data/test/test_light_data.py rename to src/data_buffer/test/test_data_buffer.py index 9ecc1f0..b43a040 100644 --- a/src/light_data/test/test_light_data.py +++ b/src/data_buffer/test/test_data_buffer.py @@ -5,20 +5,20 @@ Created on Dec 19, 2016 ''' import unittest import random -from light_data.light_data import light_data +from data_buffer import DataBuffer class Test(unittest.TestCase): def test_zero_space(self): - dut = light_data(0) + dut = DataBuffer(0) result = dut.push(10) self.assertFalse(result, "test_zero_space - exp: False res: True") def test_max_data(self): buffer_size = 10 - dut = light_data(buffer_size) + dut = DataBuffer(buffer_size) for _ in range(11): dut.push(random.random()) result = dut.length() @@ -26,7 +26,7 @@ class Test(unittest.TestCase): str(buffer_size) + " res: " + str(result)) def test_average_min(self): - dut = light_data(10) + dut = DataBuffer(10) rnd = random.random() dut.push(rnd) result = dut.average() @@ -36,7 +36,7 @@ class Test(unittest.TestCase): def test_average_mid(self): buffer_size = 10 res_buffer = [] - dut = light_data(buffer_size) + dut = DataBuffer(buffer_size) for _ in range(buffer_size / 2): rnd = random.random() dut.push(rnd) @@ -48,7 +48,7 @@ class Test(unittest.TestCase): def test_average_max(self): buffer_size = 10 res_buffer = [] - dut = light_data(buffer_size) + dut = DataBuffer(buffer_size) for _ in range(buffer_size): rnd = random.random() dut.push(rnd) diff --git a/src/engine/__init__.py b/src/engine/__init__.py index e69de29..eeb5793 100644 --- a/src/engine/__init__.py +++ b/src/engine/__init__.py @@ -0,0 +1,24 @@ +from gpio import Gpio + +class Engine: + def __init__(self, gpio_1 = 13, gpio_2 = 19): + self.gpio_1 = Gpio(gpio_1) + self.gpio_2 = Gpio(gpio_2) + + self.gpio_1.export() + self.gpio_1.direction(Gpio.DIRECTION_OUT) + self.gpio_2.export() + self.gpio_2.direction(Gpio.DIRECTION_OUT) + self.stop() + + def stop(self): + self.gpio_1.set(0) + self.gpio_2.set(0) + + def up(self): + self.gpio_1.set(1) + self.gpio_2.set(0) + + def down(self): + self.gpio_1.set(0) + self.gpio_2.set(1) diff --git a/src/engine/engine.py b/src/engine/engine.py deleted file mode 100644 index 6cf76fe..0000000 --- a/src/engine/engine.py +++ /dev/null @@ -1,24 +0,0 @@ -from gpio.gpio import gpio - -class engine: - def __init__(self, gpio_1 = 13, gpio_2 = 19): - self.gpio_1 = gpio(gpio_1) - self.gpio_2 = gpio(gpio_2) - - self.gpio_1.export() - self.gpio_1.direction("out") - self.gpio_2.export() - self.gpio_2.direction("out") - self.stop() - - def stop(self): - self.gpio_1.set(0) - self.gpio_2.set(0) - - def run_up(self): - self.gpio_1.set(1) - self.gpio_2.set(0) - - def run_down(self): - self.gpio_1.set(0) - self.gpio_2.set(1) diff --git a/src/gate/gate_handler.py b/src/gate/gate_handler.py index caf119c..137e63a 100644 --- a/src/gate/gate_handler.py +++ b/src/gate/gate_handler.py @@ -4,37 +4,21 @@ Created on Dec 19, 2016 @author: klaehn ''' -from time import time -import numpy as np -from engine.engine import engine -from power_sensor.power_sensor import power_sensor +from engine import Engine class GateHandler(object): """Gate handler class""" def __init__(self): - self.__engine = engine(13, 19) - self.__power_sensor = power_sensor(1, 0x40) - self.__max_current = {"up":315, "down":280} - self.__max_time = {"open":250, "close":250} - return + self.__engine = Engine(13, 19) def open(self): """Open the gate""" - timeout_open = time() + self.__max_time["open"] - self.__engine.run_up() - power = np.array([]) - timestamps = np.array([]) - while timeout_open > time(): - timestamps = np.append(timestamps, time()) - power = np.append(power, self.__power_sensor.power_mw()) + self.__engine.up() def close(self): """Close the gate""" - timeout_open = time() + self.__max_time["close"] - self.__engine.run_down() - power = np.array([]) - timestamps = np.array([]) - while timeout_open > time(): - timestamps = np.append(timestamps, time()) - power = np.append(power, self.__power_sensor.power_mw()) - \ No newline at end of file + self.__engine.down() + + def stop(self): + """Stop the Engine""" + self.__engine.stop() diff --git a/src/gate/gate_state.py b/src/gate/gate_state.py index 2967250..02c9d96 100644 --- a/src/gate/gate_state.py +++ b/src/gate/gate_state.py @@ -3,17 +3,168 @@ Created on Dec 19, 2016 @author: klaehn ''' -class gate_state: - def __init__(self): - self.possible_states = ["unknown", "open", "close", "opening", "closing"] - self.state = "unknown" - return +from time import time +from communiate.protocol import mqtt +from data_buffer import DataBuffer +from light_sensor.light_sensor import light_sensor +from gate.gate_handler import GateHandler +from power_sensor import PowerSensor - def set_state(self, new_state): - if new_state in self.possible_states: - self.state = new_state +STATE_INIT = "init" +STATE_OPENED = "open" +STATE_CLOSED = "close" +STATE_OPENING = "opening" +STATE_CLOSING = "closing" +STATE_ERROR = "error" + +LIGHT_READ_DELAY_S = 60 +LIGHT_CONSECUTIVE_READS = 10 +LIGHT_LX_THRESHOLD = {"open":0, "close":0} + +MQTT_HOST = "gitlab" +MQTT_TOPIC = "outdoor/chickenhouse/gate" + +LIGHT_SENSOR_I2C_BUS = 1 +LIGHT_SENSOR_I2C_ADDRESS = 0x23 + +POWER_SENSOR_I2C_BUS = 1 +POWER_SENSOR_I2C_ADDRESS = 0x40 +CONSECUTIVE_CURRENT_READS = 100 + +MAX_ENGINE_POWER = {"up":320, "down":290} +MAX_GATE_RUNTIME = {"open":250, "close":250} + + +class GateState(object): + def __init__(self): + self.__state_handler = {STATE_INIT:self.__init_handler, \ + STATE_OPENED:self.__opened_handler, \ + STATE_CLOSED:self.__closed_handler, \ + STATE_OPENING:self.__opening_handler, \ + STATE_CLOSING:self.__closing_handler, \ + STATE_ERROR:self.__error_handler} + self.__next_state = "init" + self.__last_state = "error" + + self.__light_sensor = light_sensor(LIGHT_SENSOR_I2C_BUS, \ + LIGHT_SENSOR_I2C_ADDRESS) + self.__light_data = DataBuffer(LIGHT_CONSECUTIVE_READS) + self.__comserver = mqtt(MQTT_HOST) + self.__gate_handler = GateHandler() + self.__power_sensor = PowerSensor(POWER_SENSOR_I2C_BUS, \ + POWER_SENSOR_I2C_ADDRESS) + self.__power_data = DataBuffer(CONSECUTIVE_CURRENT_READS) + self.__gate_move_timeout = 0 + self.__light_read_timeout = 0 + + def poll(self): + current_time = time() + if current_time >= self.__light_read_timeout: + self.__light_read_timeout = current_time + LIGHT_READ_DELAY_S + self.__light_data.push(self.__light_sensor.read()) + self.__state_handler[self.__next_state](self.__light_data.average()) + + def __update_state(self, new_state): + self.__last_state = self.__next_state + self.__next_state = new_state + + + def __is_transition(self): + if self.__last_state != self.__next_state: return True return False - def get_state(self): - return self.state + + def __init_handler(self, light_avg): + ''' + In init we don't know anything neither about gate state nor about + light. So first we try to reach STATE_CLOSED. + ''' + #pylint: disable=unused-argument + self.__comserver.connect() + self.__comserver.transmit(MQTT_TOPIC, str(time()) + \ + " gate gard initiated") + self.__comserver.disconnect() + self.__update_state(STATE_CLOSING) + + + def __opened_handler(self, light_avg): + next_state = self.__next_state + if self.__is_transition(): + self.__gate_handler.stop() + self.__power_data.clear() + self.__comserver.transmit(MQTT_TOPIC, str(time()) + \ + " Opening gate finished") + + if light_avg <= LIGHT_LX_THRESHOLD["close"]: + next_state = STATE_CLOSING + + self.__update_state(next_state) + + + def __closed_handler(self, light_avg): + next_state = self.__next_state + if self.__is_transition(): + self.__gate_handler.stop() + self.__power_data.clear() + self.__comserver.transmit(MQTT_TOPIC, str(time()) + \ + " Closing gate finished") + + if light_avg > LIGHT_LX_THRESHOLD["open"]: + next_state = STATE_OPENING + + self.__update_state(next_state) + + + def __opening_handler(self, light_avg): + #pylint: disable=unused-argument + next_state = self.__next_state + if self.__is_transition(): + self.__gate_handler.open() + self.__gate_move_timeout = time() + MAX_GATE_RUNTIME["open"] + self.__comserver.transmit(MQTT_TOPIC, str(time()) + \ + " Opening gate beginning") + + if time() > self.__gate_move_timeout: + next_state = STATE_ERROR + else: + self.__power_data.push(self.__power_sensor.power_mw()) + current_avg = self.__power_data.average() + if current_avg != None: + print "current_avg: " + str(current_avg) + if current_avg > MAX_ENGINE_POWER["up"]: + next_state = STATE_OPENED + + self.__update_state(next_state) + + def __closing_handler(self, light_avg): + #pylint: disable=unused-argument + next_state = self.__next_state + if self.__is_transition(): + self.__gate_handler.close() + self.__gate_move_timeout = time() + MAX_GATE_RUNTIME["close"] + self.__comserver.transmit(MQTT_TOPIC, str(time()) + \ + " Closing gate beginning") + + if time() > self.__gate_move_timeout: + next_state = STATE_ERROR + else: + self.__power_data.push(self.__power_sensor.power_mw()) + current_avg = self.__power_data.average() + if current_avg != None: + print "current_avg: " + str(current_avg) + if current_avg > MAX_ENGINE_POWER["down"]: + next_state = STATE_CLOSED + elif current_avg > MAX_ENGINE_POWER["up"]: + ''' TODO: Screwed up! while closing we are reaching the upper limit!!!''' + + self.__update_state(next_state) + + + def __error_handler(self, light_avg): + #pylint: disable=unused-argument + if self.__is_transition(): + self.__gate_handler.stop() + self.__comserver.transmit(MQTT_TOPIC, str(time()) + \ + " Error handler!!!") + self.__update_state(STATE_INIT) diff --git a/src/gate_guard.py b/src/gate_guard.py index 44d5206..2e217d5 100644 --- a/src/gate_guard.py +++ b/src/gate_guard.py @@ -1,67 +1,22 @@ -from communiate.protocol import mqtt -from gate.gate_state import gate_state -from light_data.light_data import light_data -from light_sensor.light_sensor import light_sensor +''' +Created on Dec 19, 2016 +@author: klaehn +''' import sys import time +from gate.gate_state import GateState -class gate_guard: - def __init__(self): - self.__light_read_delay_s = 60 - self.__consequtive_light_reads = 10 - self.__light_lx_close = 0 - self.__light_lx_open = 0 +def main(): + gate_state = GateState() + try: + while True: + gate_state.poll() + time.sleep(0.001) - self.__light_sensor = light_sensor(1, 0x23) - self.__light_data = light_data(self.__consequtive_light_reads) - self.__comserver = mqtt("gitlab") - self.__gate_state = gate_state() - - def close_gate(self): - topic = "outdoor/chickenhouse/gate" - payload = str(time.time()) + " closing" - self.__comserver.connect() - self.__comserver.transmit(topic, payload) - self.__comserver.disconnect() - self.__gate_state.set_state("close") - return None - - def open_gate(self): - topic = "outdoor/chickenhouse/gate" - payload = str(time.time()) + " opening" - self.__comserver.connect() - self.__comserver.transmit(topic, payload) - self.__comserver.disconnect() - self.__gate_state.set_state("open") - return None - - def handle_gate_state(self, light_avg=0): - if self.__gate_state.get_state() == "open": - if light_avg <= self.__light_lx_close: - self.close_gate() - elif self.__gate_state.get_state() == "close": - if light_avg > self.__light_lx_open: - self.open_gate() - elif self.__gate_state.get_state() == "unknown": - '''TODO: bring gate in a defined position''' - - def run(self): - if self.__gate_state.get_state() == "unknown": - self.__gate_state.set_state("open") - - try: - while True: - rd = self.__light_sensor.read() - self.__light_data.push(rd) - light_avg = self.__light_data.average() - if light_avg != None: - self.handle_gate_state(light_avg) - time.sleep(self.__light_read_delay_s) - - except KeyboardInterrupt: - return None + except KeyboardInterrupt: + print "key exit" + return None if __name__ == "__main__": - gate_guard = gate_guard() - sys.exit(gate_guard.run()) + sys.exit(main()) diff --git a/src/gpio/__init__.py b/src/gpio/__init__.py index e69de29..bf38b9e 100644 --- a/src/gpio/__init__.py +++ b/src/gpio/__init__.py @@ -0,0 +1,32 @@ +from os.path import islink, isfile + +class Gpio: + DIRECTION_OUT = "out" + DIRECTION_IN = "in" + + def __init__(self, pin): + self.pin = pin + + def export(self): + if not islink("/sys/class/gpio/gpio" + str(self.pin)): + f = open("/sys/class/gpio/export", "w") + f.write(str(self.pin)) + f.close() + + def unexport(self): + if islink("/sys/class/gpio/gpio" + str(self.pin)): + f = open("/sys/class/gpio/unexport", "w") + f.write(str(self.pin)) + f.close() + + def direction(self, direction = DIRECTION_OUT): + if isfile("/sys/class/gpio/gpio" + str(self.pin) + "/direction"): + f = open("/sys/class/gpio/gpio" + str(self.pin) + "/direction", "w") + f.write(direction) + f.close() + + def set(self, value = 0): + if isfile("/sys/class/gpio/gpio" + str(self.pin) + "/value"): + f = open("/sys/class/gpio/gpio" + str(self.pin) + "/value", "w") + f.write(str(value)) + f.close() diff --git a/src/gpio/gpio.py b/src/gpio/gpio.py deleted file mode 100644 index cca3a14..0000000 --- a/src/gpio/gpio.py +++ /dev/null @@ -1,25 +0,0 @@ -from os.path import islink, isfile - -class gpio: - def __init__(self, pin): - self.pin = pin - - def export(self): - if not islink("/sys/class/gpio/gpio" + str(self.pin)): - f = open("/sys/class/gpio/export", "w") - f.write(str(self.pin)) - - def unexport(self): - if islink("/sys/class/gpio/gpio" + str(self.pin)): - f = open("/sys/class/gpio/unexport", "w") - f.write(str(self.pin)) - - def direction(self, direction = "out"): - if isfile("/sys/class/gpio/gpio" + str(self.pin) + "/direction"): - f = open("/sys/class/gpio/gpio" + str(self.pin) + "/direction", "w") - f.write(direction) - - def set(self, value = 0): - if isfile("/sys/class/gpio/gpio" + str(self.pin) + "/value"): - f = open("/sys/class/gpio/gpio" + str(self.pin) + "/value", "w") - f.write(str(value)) diff --git a/src/light_data/light_data.py b/src/light_data/light_data.py deleted file mode 100644 index 94e33f1..0000000 --- a/src/light_data/light_data.py +++ /dev/null @@ -1,20 +0,0 @@ -class light_data: - def __init__(self, length=10): - self.max = length - self.data = [] - - def push(self, element): - if self.max == 0: - return False - if len(self.data) == self.max: - _ = self.data.pop(0) - self.data.append(element) - return True - - def average(self): - if len(self.data) != self.max: - return None - return sum(self.data) / self.max - - def length(self): - return len(self.data) \ No newline at end of file diff --git a/src/light_data/test/__init__.py b/src/light_data/test/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/light_sensor/light_sensor.py b/src/light_sensor/light_sensor.py index 8c346b9..07b8a66 100644 --- a/src/light_sensor/light_sensor.py +++ b/src/light_sensor/light_sensor.py @@ -2,10 +2,10 @@ import smbus class light_sensor: def __init__(self, bus = 1, addr = 0x23): - self.bus = bus - self.addr = addr - self.bus = smbus.SMBus(self.bus) + self.__i2c_device = bus + self.__i2c_addr = addr + self.__i2c_device = smbus.SMBus(self.__i2c_device) def read(self): - data = self.bus.read_i2c_block_data(self.addr, 0x10) + data = self.__i2c_device.read_i2c_block_data(self.__i2c_addr, 0x10) return int(round((data[0] * 256 + data[1]) / 1.2, 0)) diff --git a/src/math/__init__.py b/src/math/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/math/slope.py b/src/math/slope.py deleted file mode 100644 index 3cc4c28..0000000 --- a/src/math/slope.py +++ /dev/null @@ -1,15 +0,0 @@ -''' -Created on Dec 20, 2016 - -@author: klaehn -''' - -import numpy as np - -def calc_slope(x, y, x_dist): - """ - Calculate the slope of a curve given by x and y arrays. The x_distance - gives the start and end point for the slope. - Returns x and y arrays of the slope - """ - return diff --git a/src/mymath/__init__.py b/src/mymath/__init__.py new file mode 100644 index 0000000..f3168d5 --- /dev/null +++ b/src/mymath/__init__.py @@ -0,0 +1,13 @@ +""" +Math helper functions +""" + +def mean(x_array): + """ + Calculate the mean. + Input: x: array of x-values + Return: mean + """ + if len(x_array) > 0: + return sum(x_array) / len(x_array) + return None diff --git a/src/mymath/unittest/__init__.py b/src/mymath/unittest/__init__.py new file mode 100644 index 0000000..02c2f28 --- /dev/null +++ b/src/mymath/unittest/__init__.py @@ -0,0 +1,29 @@ +""" Unit tests for mymath module """ +import unittest +import mymath + +class Test(unittest.TestCase): + """ Test class for untit tests """ + def test_mean_even(self): + """ Unit test for mean function """ + test_arr = [1,2,3,4,5] + result = mymath.mean(test_arr) + self.assertEqual(result, 3, "test_mean_even - exp: 3, res: " + str(result)) + + def test_mean_odd(self): + """ Unit test for mean function """ + test_arr = [-1,2,-3,4,-5] + result = mymath.mean(test_arr) + self.assertEqual(result, -1, "test_mean_odd - exp: -1, res: " + str(result)) + + def test_mean_float(self): + """ Unit test for mean function """ + test_arr = [1.9,2.007,3.4,4,50.678] + result = mymath.mean(test_arr) + self.assertEqual(result, 12.397, "test_mean_float - exp: 12.397, res: " + str(result)) + + def test_mean_empty(self): + """ Unit test for mean function """ + test_arr = [] + result = mymath.mean(test_arr) + self.assertIsNone(result) diff --git a/src/power_sensor/__init__.py b/src/power_sensor/__init__.py index e69de29..b9e2bec 100644 --- a/src/power_sensor/__init__.py +++ b/src/power_sensor/__init__.py @@ -0,0 +1,49 @@ +''' +Created on Dec 19, 2016 + +@author: klaehn +''' + +import smbus + +class PowerSensor(object): + ''' + Power sensor wrapper + ''' + def __init__(self, bus = 1, addr = 0x40): + self.__bus = smbus.SMBus(bus) + self.__addr = addr + + value = [(0x1000 >> 8) & 0xFF, 0x1000 & 0xFF] + self.__bus.write_i2c_block_data(self.__addr, 0x05, value) + config = 0x2000 | 0x1800 | 0x0400 | 0x0018 | 0x0007 + value = [(config >> 8) & 0xFF, config & 0xFF] + self.__bus.write_i2c_block_data(self.__addr, 0x00, value) + + def shunt_voltage_mv(self): + ''' Read the voltage at the shunt resistor [mV] ''' + data = self.__bus.read_i2c_block_data(self.__addr, 0x01) + voltage = data[0] * 256 + data[1] + return voltage * 0.01 + + def current_ma(self): + ''' Read the current [mA] ''' + data = self.__bus.read_i2c_block_data(self.__addr, 0x04) + if data[0] >> 7 == 1: + current = data[0] * 256 + data[1] + if current & (1 << 15): + current = current - (1 << 16) + else: + current = (data[0] << 8) | (data[1]) + return current / 10 + + def power_mw(self): + ''' Read the power [mW] ''' + data = self.__bus.read_i2c_block_data(self.__addr, 0x03) + if data[0] >> 7 == 1: + power = data[0] * 256 + data[1] + if power & (1 << 15): + power = power - (1 << 16) + else: + power = (data[0] << 8) | (data[1]) + return power / 2 diff --git a/src/power_sensor/power_sensor.py b/src/power_sensor/power_sensor.py deleted file mode 100755 index 87fc40a..0000000 --- a/src/power_sensor/power_sensor.py +++ /dev/null @@ -1,37 +0,0 @@ -import smbus - -class power_sensor: - def __init__(self, bus = 1, addr = 0x40): - self.bus = smbus.SMBus(bus) - self.addr = addr - - bytes = [(0x1000 >> 8) & 0xFF, 0x1000 & 0xFF] - self.bus.write_i2c_block_data(self.addr, 0x05, bytes) - config = 0x2000 | 0x1800 | 0x0400 | 0x0018 | 0x0007 - bytes = [(config >> 8) & 0xFF, config & 0xFF] - self.bus.write_i2c_block_data(self.addr, 0x00, bytes) - - def shunt_voltage_mv(self): - data = self.bus.read_i2c_block_data(self.addr, 0x01) - voltage = data[0] * 256 + data[1] - return voltage * 0.01 - - def current_ma(self): - data = self.bus.read_i2c_block_data(self.addr, 0x04) - if(data[0] >> 7 == 1): - current = data[0] * 256 + data[1] - if(current & (1 << 15)): - current = current - (1 << 16) - else: - current = (data[0] << 8) | (data[1]) - return current / 10 - - def power_mw(self): - data = self.bus.read_i2c_block_data(self.addr, 0x03) - if(data[0] >> 7 == 1): - power = data[0] * 256 + data[1] - if(power & (1 << 15)): - power = power - (1 << 16) - else: - power = (data[0] << 8) | (data[1]) - return power / 2