Temper USB Sensor Setup

File Contents:

1. temper.service: [Unit]
Description=Temper Service
After=multi-user.target

[Service]
Type=idle
ExecStart=/usr/bin/python3.6 /root/temper-metrics.py<c/ode>

[Install]
WantedBy=multi-user.target

2. temper-metrics.py:
import time, http.server
#from sense_hat import SenseHat
#from tempSensor import tempf
#from pressureSensor import pressure
#from humiditySensor import humidity
import socket
import os
import subprocess

#temper_command = '/usr/bin/python3.6 /root/temper.py | awk "{print tempsensor, $8 }" | sed "s/F//g"'


#temper_command = subprocess.check_output(["/usr/bin/python3.6", "-u", "/root/temper.py"])

class Metrics(http.server.BaseHTTPRequestHandler):
def do_GET(self):
# so that we can make a best-effort to respond to the caller.
# in order to troubleshoot any errors, comment this and call
# parseget() without a try / catch.
try:
response, body = parseget(self.path)
except Exception as e:
response = 500
body = ""

self.send_response(response or 500)
self.send_header("Content-Type", "text/plain; charset=utf-8")
# must be led with a newline. otherwise, "message corrupt".
# there may be other caveats I haven't explored / encoutered.
self.wfile.write("\n{0}".format(body or "").encode())


def getStats():
#this is the func() for however you get the stats from your sensehat
#return a string that lists the stats that you want
#the rest of this code should work without any change
bashCommand = 'echo "$(/root/temper.py | awk \'{print $8}\' | sed \'s/F//g\')" '
output = subprocess.check_output(['bash','-c', bashCommand])
results = ('[your metric] ' + str(output.decode()))
return(results)

def parseget(path):
uri = path.split("?")
if uri[0] == "/metrics":
body = getStats()
# elif uri[0] == "/senseHat":
# body = getInnerTemp()
else:
# all other endpoints in the URI are "not implemented"
return 501, None

return 200, body


port = 9210
server = http.server.HTTPServer(("", port), Metrics)
try:
server.serve_forever()
finally:
server.server_close()

3. temper.py:
import argparse
import binascii
import json
import os
import re
import select
import struct
import syse>
# Non-standard modules
try:
import serial
except ImportError:
print('Cannot import "serial". Please sudo apt-get install python3-serial')
sys.exit(1)

class USBList(object):
'''Get a list of all of the USB devices on a system, along with their
associated hidraw or serial (tty) devices.
'''

SYSPATH = '/sys/bus/usb/devices'

def _readfile(self, path):
'''Read data from 'path' and return it as a string. Return the empty string
if the file does not exist, cannot be read, or has an error.
'''
try:
with open(path, 'r') as fp:
return fp.read().strip()
except:
return ''

def _find_devices(self, dirname):
'''Scan a directory hierarchy for names that start with "tty" or "hidraw".
Return these names in a set.
'''
devices = set()
for entry in os.scandir(dirname):
if entry.is_dir() and not entry.is_symlink():
devices |= self._find_devices(os.path.join(dirname, entry.name))
if re.search('tty.*[0-9]', entry.name):
devices.add(entry.name)
if re.search('hidraw[0-9]', entry.name):
devices.add(entry.name)
return devices

def _get_usb_device(self, dirname):
'''Examine the files in 'dirname', looking for files with well-known
names expected to be in the /sys hierarchy under Linux for USB devices.
Return a dictionary of the information gathered. If no information is found
(i.e., because the directory is not for a USB device) return None.
'''
info = dict()
vendorid = self._readfile(os.path.join(dirname, 'idVendor'))
if vendorid == '':
return None
info['vendorid'] = int(vendorid, 16)
productid = self._readfile(os.path.join(dirname, 'idProduct'))
info['productid'] = int(productid, 16)
info['manufacturer'] = self._readfile(os.path.join(dirname,
'manufacturer'))
info['product'] = self._readfile(os.path.join(dirname, 'product'))
info['busnum'] = int(self._readfile(os.path.join(dirname, 'busnum')))
info['devnum'] = int(self._readfile(os.path.join(dirname, 'devnum')))
info['devices'] = sorted(self._find_devices(dirname))
return info

def get_usb_devices(self):
'''Scan a well-known Linux hierarchy in /sys and try to find all of the
USB devices on a system. Return these as a dictionary indexed by the path.
'''
info = dict()
for entry in os.scandir(Temper.SYSPATH):
if entry.is_dir():
path = os.path.join(Temper.SYSPATH, entry.name)
device = self._get_usb_device(path)
if device is not None:
info[path] = device
return info

class USBRead(object):
'''Read temperature and/or humidity information from a specified USB device.
'''
def __init__(self, device, verbose=False):
self.device = device
self.verbose = verbose

def _parse_bytes(self, name, offset, divisor, bytes, info):
'''Data is returned from several devices in a similar format. In the first
8 bytes, the internal sensors are returned in bytes 2 and 3 (temperature)
and in bytes 4 and 5 (humidity). In the second 8 bytes, external sensor
information is returned. If there are only external sensors, then only 8
bytes are returned, and the caller is expected to use the correct 'name'.
The caller is also expected to detect the firmware version and provide the
appropriate divisor, which is usually 100 or 256.

There is no return value. Instead 'info[name]' is update directly, if a
value is found.
'''
try:
if bytes[offset] == 0x4e and bytes[offset+1] == 0x20:
return
except:
return
try:
info[name] = struct.unpack_from('>h', bytes, offset)[0] / divisor
except:
return

def _read_hidraw(self, device):
'''Using the Linux hidraw device, send the special commands and receive the
raw data. Then call '_parse_bytes' based on the firmware version to provide
temperature and humidity information.

A dictionary of temperature and humidity info is returned.
'''
path = os.path.join('/dev', device)
fd = os.open(path, os.O_RDWR)

# Get firmware identifier
os.write(fd, struct.pack('8B', 0x01, 0x86, 0xff, 0x01, 0, 0, 0, 0))
firmware = b''
while True:
r, _, _ = select.select([fd], [], [], 0.1)
if fd not in r:
break
data = os.read(fd, 8)
firmware += data

if firmware == b'':
os.close(fd)
return { 'error' : 'Cannot read firmware identifier from device' }
if self.verbose:
print('Firmware value: %s' % binascii.b2a_hex(firmware))

# Get temperature/humidity
os.write(fd, struct.pack('8B', 0x01, 0x80, 0x33, 0x01, 0, 0, 0, 0))
bytes = b''
while True:
r, _, _ = select.select([fd], [], [], 0.1)
if fd not in r:
break
data = os.read(fd, 8)
bytes += data

os.close(fd)
if self.verbose:
print('Data value: %s' % binascii.hexlify(bytes))

info = dict()
info['firmware'] = str(firmware, 'latin-1').strip()
info['hex_firmware'] = str(binascii.b2a_hex(firmware), 'latin-1')
info['hex_data'] = str(binascii.b2a_hex(bytes), 'latin-1')

if info['firmware'][:10] == 'TEMPerF1.4':
info['firmware'] = info['firmware'][:10]
self._parse_bytes('internal temperature', 2, 256.0, bytes, info)
return info

if info['firmware'][:15] == 'TEMPerGold_V3.1':
info['firmware'] = info['firmware'][:15]
self._parse_bytes('internal temperature', 2, 100.0, bytes, info)
return info

if info['firmware'][:12] in [ 'TEMPerX_V3.1', 'TEMPerX_V3.3' ]:
info['firmware'] = info['firmware'][:12]
self._parse_bytes('internal temperature', 2, 100.0, bytes, info)
self._parse_bytes('internal humidity', 4, 100.0, bytes, info)
self._parse_bytes('external temperature', 10, 100.0, bytes, info)
self._parse_bytes('external humidity', 12, 100.0, bytes, info)
return info

info['error'] = 'Unknown firmware %s: %s' % (info['firmware'],
binascii.hexlify(bytes))
return info

def _read_serial(self, device):
'''Using the Linux serial device, send the special commands and receive the
text data, which is parsed directly in this method.

A dictionary of device info (like that returned by USBList) combined with
temperature and humidity info is returned.
'''

path = os.path.join('/dev', device)
s = serial.Serial(path, 9600)
s.bytesize = serial.EIGHTBITS
s.parity = serial.PARITY_NONE
s.stopbits = serial.STOPBITS_ONE
s.timeout = 1
s.xonoff = False
s.rtscts = False
s.dsrdtr = False
s.writeTimeout = 0

# Send the "Version" command and save the reply.
s.write(b'Version')
firmware = str(s.readline(), 'latin-1').strip()

# Send the "ReadTemp" command and save the reply.
s.write(b'ReadTemp')
reply = str(s.readline(), 'latin-1').strip()
reply += str(s.readline(), 'latin-1').strip()
s.close()

info = dict()
info['firmware'] = firmware
m = re.search(r'Temp-Inner:([0-9.]*).*, ?([0-9.]*)', reply)
if m is not None:
info['internal temperature'] = float(m.group(1))
info['internal humidity'] = float(m.group(2))
m = re.search(r'Temp-Outer:([0-9.]*)', reply)
if m is not None:
try:
info['external temperature'] = float(m.group(1))
except:
pass
return info

def read(self):
'''Read the firmware version, temperature, and humidity from the device and
return a dictionary containing these data.
'''
# Use the last device found
if self.device.startswith('hidraw'):
return self._read_hidraw(self.device)
if self.device.startswith('tty'):
return self._read_serial(self.device)
return {'error': 'No usable hid/tty devices available'}

class Temper(object):
SYSPATH = '/sys/bus/usb/devices'

def __init__(self, verbose=False):
usblist = USBList()
self.usb_devices = usblist.get_usb_devices()
self.forced_vendor_id = None
self.forced_product_id = None
self.verbose = verbose

def _is_known_id(self, vendorid, productid):
'''Returns True if the vendorid and product id are valid.
'''
if self.forced_vendor_id is not None and \
self.forced_product_id is not None:
if self.forced_vendor_id == vendorid and \
self.forced_product_id == productid:
return True
return False

if vendorid == 0x0c45 and productid == 0x7401:
return True
if vendorid == 0x413d and productid == 0x2107:
return True
if vendorid == 0x1a86 and productid == 0x5523:
return True

# The id is not known to this program.
return False

def list(self, use_json=False):
'''Print out a list all of the USB devices on the system. If 'use_json' is
True, then JSON formatting will be used.
'''
if use_json:
print(json.dumps(self.usb_devices, indent=4))
return

for _, info in sorted(self.usb_devices.items(),
key=lambda x: x[1]['busnum'] * 1000 + \
x[1]['devnum']):
print('Bus %03d Dev %03d %04x:%04x %s %s %s' % (
info['busnum'],
info['devnum'],
info['vendorid'],
info['productid'],
'*' if self._is_known_id(info['vendorid'], info['productid']) else ' ',
info.get('product', '???'),
list(info['devices']) if len(info['devices']) > 0 else ''))

def read(self, verbose=False):
'''Read all of the known devices on the system and return a list of
dictionaries which contain the device information, firmware information,
and environmental information obtained. If there is an error, then the
'error' field in the dictionary will contain a string explaining the
error.
'''
results = []
for _, info in sorted(self.usb_devices.items(),
key=lambda x: x[1]['busnum'] * 1000 + \
x[1]['devnum']):
if not self._is_known_id(info['vendorid'], info['productid']):
continue
if len(info['devices']) == 0:
info['error'] = 'no hid/tty devices available'
results.append(info)
continue
usbread = USBRead(info['devices'][-1], verbose)
results.append({ **info, **usbread.read() })
return results

def _add_temperature(self, name, info):
'''Helper method to add the temperature to a string in both Celsius and
Fahrenheit. If no sensor data is available, then '- -' will be returned.
'''
if name not in info:
return '- -'
degC = info[name]
degF = degC * 1.8 + 32.0
return '%.1fC %.1fF' % (degC, degF)

def _add_humidity(self, name, info):
'''Helper method to add the humidity to a string. If no sensor data is
available, then '-' will be returned.
'''

if name not in info:
return '-'
return '%d%%' % int(info[name])

def print(self, results, use_json=False):
'''Print out a list of all of the known USB sensor devices on the system.
If 'use_json' is True, then JSON formatting will be used.
'''

if use_json:
print(json.dumps(results, indent=4))
return

for info in results:
s = 'Bus %03d Dev %03d %04x:%04x %s' % (info['busnum'],
info['devnum'],
info['vendorid'],
info['productid'],
info.get('firmware'))
if 'error' in info:
s += ' Error: %s' % info['error']
else:
s += ' ' + self._add_temperature('internal temperature', info)
s += ' ' + self._add_humidity('internal humidity', info)
s += ' ' + self._add_temperature('external temperature', info)
s += ' ' + self._add_humidity('external humidity', info)
print(s)

def main(self):
'''An example 'main' entry point that can be used to make temper.py a
standalone program.
'''

parser = argparse.ArgumentParser(description='temper')
parser.add_argument('-l', '--list', action='store_true',
help='List all USB devices')
parser.add_argument('--json', action='store_true',
help='Provide output as JSON')
parser.add_argument('--force', type=str,
help='Force the use of the hex id; ignore other ids',
metavar=('VENDOR_ID:PRODUCT_ID'))
parser.add_argument('--verbose', action='store_true',
help='Output binary data from thermometer')
args = parser.parse_args()
self.verbose = args.verbose

if args.list:
self.list(args.json)
return 0

if args.force:
ids = args.force.split(':')
if len(ids) != 2:
print('Cannot parse hexadecimal id: %s' % args.force)
return 1
try:
vendor_id = int(ids[0], 16)
product_id = int(ids[1], 16)
except:
print('Cannot parse hexadecimal id: %s' % args.force)
return 1
self.forced_vendor_id = vendor_id;
self.forced_product_id = product_id;

# By default, output the temperature and humidity for all known sensors.
results = self.read(args.verbose)
self.print(results, args.json)
return 0

if __name__ == "__main__":
temper = Temper()
sys.exit(temper.main())