python/chickenhouse
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

gate: pylint

Browse Source 
Signed-off-by: Thomas Klaehn <thomas.klaehn@u-blox.com>
This commit is contained in:
Thomas Klaehn 2017-04-10 10:38:37 +02:00
parent 40146343ea
commit 8e6a2550f0
2 changed files with 31 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gate_guard/__init__.py
View File

@ -4,13 +4,13 @@ Created on Dec 19, 2016
@author: klaehn @author: klaehn
''' '''
import time import time
import gate
import sys import sys
import logging import logging
import gate_guard.gate
def main(): def main():
logging.basicConfig(filename='/var/log/gate_guard.log', level=logging.DEBUG) logging.basicConfig(filename='/var/log/gate_guard.log', level=logging.DEBUG)
gate_state = gate.Gate() gate_state = gate_guard.gate.Gate()
try: try:
while True: while True:
gate_state.poll() gate_state.poll()

57
gate_guard/gate.py
View File

@ -5,10 +5,10 @@ Created on Dec 19, 2016
''' '''
import time import time
import mqtt import mqtt
import data_buffer import gate_guard.data_buffer
import light_sensor import gate_guard.light_sensor
import engine import gate_guard.engine
import power_sensor import gate_guard.power_sensor
import scipy import scipy
import scipy.stats import scipy.stats
import logging import logging
@ -33,11 +33,11 @@ LIGHT_SENSOR_I2C_ADDRESS = 0x23
POWER_SENSOR_I2C_BUS = 1 POWER_SENSOR_I2C_BUS = 1
POWER_SENSOR_I2C_ADDRESS = 0x40 POWER_SENSOR_I2C_ADDRESS = 0x40
SLOPE_COUNT = 100 SLOPE_COUNT = 10
SLOPE_CNT_MIN = 10 SLOPE_CNT_MIN = 2
MAX_GATE_RUNTIME = {"open":60, "close":50} MAX_GATE_RUNTIME = {"open":70, "close":60}
MAX_POWER_SLOPE = {"up":10, "down":2} MAX_POWER_SLOPE = {"up":40, "down":15}
class Gate(object): class Gate(object):
def __init__(self): def __init__(self):
@ -50,19 +50,20 @@ class Gate(object):
self.__next_state = STATE_INIT self.__next_state = STATE_INIT
self.__last_state = STATE_ERROR self.__last_state = STATE_ERROR
self.__light_sensor = light_sensor.LightSensor(LIGHT_SENSOR_I2C_BUS, \ self.__light_sensor = gate_guard.light_sensor.LightSensor(
LIGHT_SENSOR_I2C_ADDRESS) LIGHT_SENSOR_I2C_BUS, LIGHT_SENSOR_I2C_ADDRESS)
self.__light_data = data_buffer.DataBuffer(LIGHT_CONSECUTIVE_READS) self.__light_data = gate_guard.data_buffer.DataBuffer(
LIGHT_CONSECUTIVE_READS)
self.__comserver = mqtt.Mqtt(MQTT_HOST) self.__comserver = mqtt.Mqtt(MQTT_HOST)
self.__engine = engine.Engine(gpio_1=13, gpio_2=19) self.__engine = gate_guard.engine.Engine(gpio_1=13, gpio_2=19)
self.__power_sensor = power_sensor.PowerSensor(POWER_SENSOR_I2C_BUS, \ self.__power_sensor = gate_guard.power_sensor.PowerSensor(
POWER_SENSOR_I2C_ADDRESS) POWER_SENSOR_I2C_BUS, POWER_SENSOR_I2C_ADDRESS)
self.__gate_move_timeout = 0 self.__gate_move_timeout = 0
self.__light_read_timeout = 0 self.__light_read_timeout = 0
self.__error_count = 0 self.__error_count = 0
self.slope_power = data_buffer.DataBuffer(SLOPE_COUNT) self.slope_power = gate_guard.data_buffer.DataBuffer(SLOPE_COUNT)
self.slope_time = data_buffer.DataBuffer(SLOPE_COUNT) self.slope_time = gate_guard.data_buffer.DataBuffer(SLOPE_COUNT)
def poll(self): def poll(self):
current_time = time.time() current_time = time.time()
@ -78,7 +79,8 @@ class Gate(object):
def __is_transition(self): def __is_transition(self):
if self.__last_state != self.__next_state: if self.__last_state != self.__next_state:
logging.info('STATE: ' + self.__last_state + ' -> ' + self.__next_state) logging.info('STATE: ' + self.__last_state + ' -> ' + \
self.__next_state)
return True return True
return False return False
@ -100,7 +102,7 @@ class Gate(object):
next_state = self.__next_state next_state = self.__next_state
if self.__is_transition(): if self.__is_transition():
self.__engine.down() self.__engine.down()
time.sleep(5) time.sleep(0.5)
self.__engine.stop() self.__engine.stop()
slope, _, _, _, _ = scipy.stats.linregress(self.slope_time.get(), slope, _, _, _, _ = scipy.stats.linregress(self.slope_time.get(),
self.slope_power.get()) self.slope_power.get())
@ -120,14 +122,13 @@ class Gate(object):
next_state = self.__next_state next_state = self.__next_state
if self.__is_transition(): if self.__is_transition():
self.__engine.up() self.__engine.up()
time.sleep(5) time.sleep(0.5)
self.__engine.stop() self.__engine.stop()
slope, _, _, _, _ = scipy.stats.linregress(self.slope_time.get(), slope, _, _, _, _ = scipy.stats.linregress(self.slope_time.get(),
self.slope_power.get()) self.slope_power.get())
self.__comserver.transmit(MQTT_TOPIC, str(time.time()) + \ self.__comserver.transmit(MQTT_TOPIC, str(time.time()) + \
" Closed " + \ " Closed " + \
str(slope)) str(slope))
self.__power_data.clear()
self.slope_power.clear() self.slope_power.clear()
self.slope_time.clear() self.slope_time.clear()
@ -151,11 +152,11 @@ class Gate(object):
pwr = self.__power_sensor.power_mw() pwr = self.__power_sensor.power_mw()
self.slope_power.push(pwr) self.slope_power.push(pwr)
self.slope_time.push(tm) self.slope_time.push(tm)
logging.debug('up: ' + str(tm) + ' ' + str(pwr))
slope = 0 slope = 0
if self.slope_power.length() >= SLOPE_CNT_MIN: if self.slope_power.length() >= SLOPE_CNT_MIN:
slope, _, _, _, _ = scipy.stats.linregress(self.slope_time.get(), slope, _, _, _, _ = scipy.stats.linregress(
self.slope_power.get()) self.slope_time.get(), self.slope_power.get())
logging.debug('up: ' + str(tm) + ' ' + str(pwr) + ' ' + str(slope))
if slope > MAX_POWER_SLOPE["up"]: if slope > MAX_POWER_SLOPE["up"]:
next_state = STATE_OPENED next_state = STATE_OPENED
self.__update_state(next_state) self.__update_state(next_state)
@ -174,14 +175,14 @@ class Gate(object):
else: else:
pwr = self.__power_sensor.power_mw() pwr = self.__power_sensor.power_mw()
self.slope_power.push(pwr) self.slope_power.push(pwr)
self.slope_time.push(pwr) self.slope_time.push(tm)
logging.debug('dw: ' + str(tm) + ' ' + str(pwr))
slope = 0 slope = 0
if self.slope_power.length() >= SLOPE_CNT_MIN: if self.slope_power.length() >= SLOPE_CNT_MIN:
slope, _, _, _, _ = scipy.stats.linregress(self.slope_time.get(), slope, _, _, _, _ = scipy.stats.linregress(
self.slope_power.get()) self.slope_time.get(), self.slope_power.get())
logging.debug('dw: ' + str(tm) + ' ' + str(pwr) + ' ' + str(slope))
if slope > MAX_POWER_SLOPE["down"]: if slope > MAX_POWER_SLOPE["down"]:
next_state = STATE_OPENED next_state = STATE_CLOSED
self.__update_state(next_state) self.__update_state(next_state)