123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
import paho.mqtt.client as mqtt
|
|
import serial
|
|
import time
|
|
|
|
|
|
class ShellerRPMDisplay:
|
|
mqtt_host = "172.22.112.94"
|
|
|
|
COM_PORTS = [ "/dev/ttyACM0", "/dev/ttyUSB0" ]
|
|
buffer = ""
|
|
def init(self):
|
|
self.drum = 0
|
|
self.paddle = 0
|
|
try:
|
|
self.serials = [
|
|
serial.Serial(
|
|
self.COM_PORTS[1],
|
|
9600,
|
|
timeout=1,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
bytesize=serial.EIGHTBITS
|
|
),
|
|
serial.Serial(
|
|
self.COM_PORTS[0],
|
|
9600,
|
|
timeout=1,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
bytesize=serial.EIGHTBITS
|
|
)
|
|
]
|
|
self.mqtt_inform()
|
|
for s in self.serials:
|
|
s.write(bytearray("@D1\r\n", encoding='ascii'))
|
|
return True
|
|
except Exception as e:
|
|
print(e)
|
|
time.sleep(1)
|
|
if hasattr(self, "serials"):
|
|
del self.serials
|
|
return False
|
|
|
|
def start(self):
|
|
self.isrunning = True
|
|
self.mqtt_inform()
|
|
|
|
def stop(self):
|
|
self.isrunning = False
|
|
if hasattr(self, "serials"):
|
|
try:
|
|
for serial in self.serials:
|
|
serial.close()
|
|
except:
|
|
pass
|
|
del self.serials
|
|
|
|
def main(self):
|
|
while self.isrunning:
|
|
if not hasattr(self, "serials"):
|
|
self.init()
|
|
try:
|
|
for i, serial in enumerate(self.serials):
|
|
if serial.in_waiting > 0:
|
|
self.buffer += serial.read(serial.in_waiting).decode(
|
|
"utf-8"
|
|
)
|
|
print(self.buffer)
|
|
self.process_text(self.buffer[: self.buffer.index("\r")], i)
|
|
self.buffer = ""
|
|
except Exception as e:
|
|
print(e)
|
|
time.sleep(1)
|
|
|
|
def mqtt_inform(self):
|
|
# Publish MQTT messages for Moisture Percentage
|
|
self.client = mqtt.Client()
|
|
self.client.connect(self.mqtt_host, 1883)
|
|
self.client.publish(
|
|
"homeassistant/sensor/sheller-paddle-rpm/config",
|
|
'{\
|
|
"name": "Paddle RPM", \
|
|
"state_topic": "homeassistant/sensor/sheller/state", \
|
|
"unit_of_measurement": "RPM", \
|
|
"value_template": "{{ value_json.paddle }}", \
|
|
"unique_id": "sheller-paddle-rpm", \
|
|
"device":{ \
|
|
"identifiers": ["shelling-machine"], \
|
|
"name": "Shelling Machine", \
|
|
"manufacturer": "ME&E" \
|
|
} \
|
|
}', retain=True
|
|
)
|
|
self.client.publish(
|
|
"homeassistant/sensor/sheller-drum-rpm/config",
|
|
'{\
|
|
"name": "Drum RPM", \
|
|
"state_topic": "homeassistant/sensor/sheller/state", \
|
|
"unit_of_measurement": "RPM", \
|
|
"value_template": "{{ value_json.drum }}", \
|
|
"unique_id": "sheller-drum-rpm", \
|
|
"device":{ \
|
|
"identifiers": ["shelling-machine"] \
|
|
} \
|
|
}', retain=True
|
|
)
|
|
|
|
def process_text(self, text, index):
|
|
print(text)
|
|
if index == 0:
|
|
self.paddle = text
|
|
elif index == 1:
|
|
self.drum = text
|
|
|
|
self.client.publish(
|
|
"homeassistant/sensor/sheller/state",
|
|
'{"paddle": "%s", "drum": "%s"}'
|
|
% (self.paddle, self.drum), retain=True
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
obj = ShellerRPMDisplay()
|
|
obj.start()
|
|
obj.main()
|