try new data class
This commit is contained in:
parent
208a82142c
commit
2b1c12f273
2
.idea/ATC_Sensor_Gateway.iml
generated
2
.idea/ATC_Sensor_Gateway.iml
generated
@ -2,7 +2,7 @@
|
||||
<module type="PYTHON_MODULE" version="4">
|
||||
<component name="NewModuleRootManager">
|
||||
<content url="file://$MODULE_DIR$" />
|
||||
<orderEntry type="jdk" jdkName="Python 3.12" jdkType="Python SDK" />
|
||||
<orderEntry type="inheritedJdk" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
</module>
|
||||
2
.idea/misc.xml
generated
2
.idea/misc.xml
generated
@ -3,5 +3,5 @@
|
||||
<component name="Black">
|
||||
<option name="sdkName" value="Python 3.12 (autopicture-v3)" />
|
||||
</component>
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12" project-jdk-type="Python SDK" />
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10" project-jdk-type="Python SDK" />
|
||||
</project>
|
||||
@ -1,12 +1,30 @@
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class Data:
|
||||
def __init__(self, data:dict):
|
||||
self.timestamp = data['timestamp']
|
||||
self.mac = data['mac']
|
||||
self.temperature = data['temperature']
|
||||
self.humidity = data['humidity']
|
||||
self.battery_percent = data['battery_percent']
|
||||
self.battery_volt = data['battery_volt']
|
||||
self.count = data['count']
|
||||
def __init__(self, data):
|
||||
self.timestamp = None
|
||||
self.mac = None
|
||||
self.temperature = None
|
||||
self.humidity = None
|
||||
self.battery_percent = None
|
||||
self.battery_volt = None
|
||||
self.count = None
|
||||
self.parse_data(data)
|
||||
|
||||
def parse_data(self, val):
|
||||
data_bytes = [int(val[i:i + 2], 16) for i in range(0, len(val), 2)]
|
||||
|
||||
if data_bytes[8] > 127:
|
||||
print(data_bytes)
|
||||
data_bytes[8] -= 256
|
||||
self.timestamp = datetime.now().astimezone().replace(microsecond=0).isoformat()
|
||||
self.mac = ":".join(["{:02X}".format(data_bytes[i]) for i in range(2, 8)])
|
||||
self.temperature = (data_bytes[8] * 256 + data_bytes[9]) / 10
|
||||
self.humidity = data_bytes[10]
|
||||
self.battery_percent = data_bytes[11]
|
||||
self.battery_volt = (data_bytes[12] * 256 + data_bytes[13]) / 1000
|
||||
self.count = data_bytes[14]
|
||||
|
||||
def print_data(self):
|
||||
print(self.to_json())
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
from bluepy.btle import DefaultDelegate
|
||||
from bluepy.btle import Scanner
|
||||
from datetime import datetime
|
||||
|
||||
from data_class import Data
|
||||
from devices import get_devices as get_device_from_config
|
||||
@ -18,10 +17,11 @@ class ScanDelegate(DefaultDelegate):
|
||||
|
||||
for (sdid, desc, val) in dev.getScanData():
|
||||
if self.is_temperature(sdid, val):
|
||||
data_obj = Data(self.parse_data(val))
|
||||
data_obj = Data(val)
|
||||
|
||||
if self.is_atc_device(dev, data_obj):
|
||||
device_from_config = self.get_device(dev)
|
||||
devices.append([dev, data_obj])
|
||||
devices.append([dev, data_obj, device_from_config])
|
||||
|
||||
@staticmethod
|
||||
def is_temperature(sdid, val):
|
||||
@ -42,35 +42,22 @@ class ScanDelegate(DefaultDelegate):
|
||||
print(f'\tTemp: {data_obj.temperature}°C, Humid: {data_obj.humidity}%, Batt: {data_obj.battery_percent}%\n')
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def parse_data(val):
|
||||
bytes = [int(val[i:i + 2], 16) for i in range(0, len(val), 2)]
|
||||
if bytes[8] > 127:
|
||||
bytes[8] -= 256
|
||||
return {
|
||||
'timestamp': datetime.now().astimezone().replace(microsecond=0).isoformat(),
|
||||
'mac': ":".join(["{:02X}".format(bytes[i]) for i in range(2, 8)]),
|
||||
'temperature': (bytes[8] * 256 + bytes[9]) / 10,
|
||||
'humidity': bytes[10],
|
||||
'battery_percent': bytes[11],
|
||||
'battery_volt': (bytes[12] * 256 + bytes[13]) / 1000,
|
||||
'count': bytes[14],
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def get_device(dev):
|
||||
return next((d for d in get_device_from_config() if d.mac == dev.addr.upper()), None)
|
||||
|
||||
|
||||
class Discovery:
|
||||
def __init__(self):
|
||||
pass
|
||||
def cleanup():
|
||||
global devices
|
||||
devices = []
|
||||
|
||||
def start_discovery(self, timeout=20.0):
|
||||
|
||||
def start_discovery(timeout=20.0):
|
||||
cleanup()
|
||||
global devices
|
||||
print(f'Start discovery with timout {timeout}s...')
|
||||
|
||||
scanner = Scanner().withDelegate(ScanDelegate())
|
||||
scanner.scan(timeout=timeout, passive=True)
|
||||
scanner.scan(timeout=timeout, passive=False)
|
||||
|
||||
return devices
|
||||
|
||||
@ -1,8 +1,9 @@
|
||||
from time import sleep
|
||||
|
||||
from discovery import start_discovery
|
||||
|
||||
|
||||
def start_loop(interval=60):
|
||||
while True:
|
||||
from discovery import start_discovery
|
||||
start_discovery()
|
||||
sleep(interval)
|
||||
|
||||
@ -1,13 +1,7 @@
|
||||
from discovery import Discovery
|
||||
from discovery import start_discovery
|
||||
from data_class import Data
|
||||
from loop import start_loop
|
||||
from time import sleep
|
||||
|
||||
while True:
|
||||
discovery = Discovery()
|
||||
devices = discovery.start_discovery()
|
||||
sleep(10)
|
||||
|
||||
|
||||
|
||||
devices = start_discovery()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user