Created
August 7, 2016 12:19
-
-
Save piotrmaslanka/98899e356ec36b9e38ca9772209eb898 to your computer and use it in GitHub Desktop.
A script for SMOK4 (https://smok.co) to query GSM devices about current external temperature
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from sai.devices import Sensor, Pathpoint, get_all_devices, slTREND | |
from sai.orders import acADVISE, Section | |
from sai.basic import ur | |
def find_applicable_devices(): | |
""" | |
Finds devices that are applicable to be taken under QoR. | |
:return: list of Device objects | |
""" | |
# Step 1: Find all devices that have csqtrack installed | |
devices_with_csqtrack = [] | |
for device in get_all_devices(): | |
found_csqtrack = False | |
for cons in device.get_device_handler_api().get_associated_devices(): | |
if cons.get_device_handler_api().configuration.startswith('csqtrack'): | |
found_csqtrack = True | |
break | |
if found_csqtrack: | |
devices_with_csqtrack.append(device) | |
# Step 2: This device must declare 'temp external' | |
devices_with_external_and_csqtrack = [] | |
for device in devices_with_csqtrack: | |
try: | |
ts, v = Sensor(device.device_id, 'temp external').get() | |
except Sensor.DoesNotExist: | |
continue | |
except (Sensor.NotReadedError, Sensor.ReadFailedError): | |
continue | |
else: | |
devices_with_external_and_csqtrack.append(device) | |
return devices_with_external_and_csqtrack | |
def force_read_lac_cid(devices): | |
""" | |
This declares (if not existent) and forces a read of LAC and CID. | |
This forces read of external temperature as well. | |
:param devices: list of Device objects | |
""" | |
for device in devices: | |
device_id = device.device_id | |
# Do Watlac and Watcid exist? | |
if not Pathpoint(device_id, 'Watlac').exists(): | |
Pathpoint.define(device_id, 'Watlac', slTREND) | |
if not Pathpoint(device_id, 'Watcid').exists(): | |
Pathpoint.define(device_id, 'Watcid', slTREND) | |
# create a read barrage | |
q = Section() | |
q.absorb(Pathpoint(device_id, 'Watlac').read(acADVISE)) | |
q.absorb(Pathpoint(device_id, 'Watcid').read(acADVISE)) | |
q.absorb(Sensor(device_id, 'temp external').read(acADVISE)) | |
device.execute_sections([q]) | |
def show_results(devices): | |
"""Show results in reading shit""" | |
for device in devices: | |
device_id = device.device_id | |
try: | |
ts, lac = Pathpoint(device_id, 'Watlac').get() | |
ts, cid = Pathpoint(device_id, 'Watcid').get() | |
ts, temp = Sensor(device_id, 'temp external').get() | |
if isinstance(temp, ur.Quantity): temp = temp.magnitude # if unit-aware | |
except (Sensor.ReadFailedError, Sensor.DoesNotExist, Sensor.NotReadedError): | |
continue | |
print '%s:%s => %s C' % (lac, cid, temp) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment