huawei_lte_exporter/huwawei_lte_exporter

#!/usr/bin/env python3

import os
import pprint
import http.server as SimpleHTTPServer
import socketserver as SocketServer
from huawei_lte_api.Client import Client
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
from huawei_lte_api.Connection import Connection

def prom_exporter():
	MANDATORY_ENV_VARS = ["ROUTER_ADDRESS", "ROUTER_USER", "ROUTER_PASS", "PROM_PORT"]
	for var in MANDATORY_ENV_VARS:
	    if var not in os.environ:
	        raise EnvironmentError("Failed because {} is not set.".format(var))

	# Auth to router
	conn ='http://'+os.environ.get('ROUTER_USER')+':'+os.environ.get('ROUTER_PASS')+'@'+os.environ.get('ROUTER_ADDRESS')+'/'
	#print('Connecting to '+conn)
	connection = AuthorizedConnection(conn)

	# Initiatise client
	client = Client(connection)

	pprint.pprint(client.device.signal())  # Can be accessed without authorization
	#pprint.pprint(client.device.information())  # Needs valid authorization, will throw exception if invalid credentials are passed in URL

	signal = {	'band' : client.device.signal().get('band'), 
				'rsrp' : client.device.signal().get('rsrp').replace("dBm", ""),
				'rsrq' : client.device.signal().get('rsrq').replace("dB", ""),
				'rssi' : client.device.signal().get('rssi').replace("dBm", ""),
				'sinr' : client.device.signal().get('sinr').replace("dB", "")
				}

	dev = 'deviceName="'+ client.device.information().get('DeviceName')+'",iccid="'+client.device.information().get('Iccid')+'"'
	devband=dev
	if signal.get('band') is not None:
		devband = dev+',band="'+signal.get('band')+'"'

	# HELP ifInDiscards The number of inbound packets which were chosen to be discarded even though no errors had been detected to prevent their being deliverable to a higher-layer protocol - 1.3.6.1.2.1.2.2.1.13
	# TYPE ifInDiscards counter
	#ifInDiscards{ifAlias="",ifDescr="atm0",ifIndex="7",ifName=""} 81

	response=[]
	# Band
	if signal.get('band') is not None:
		response.append('#HELP band The signal band the LTE connection is using')
		response.append('#TYPE band gauge')
		response.append('band{'+dev+'} '+signal.get('band'))
	# rsrp
	if signal.get('rsrp') is not None:
		response.append('#HELP rsrp The average power received from a single Reference signal, and Its typical range is around -44dbm (good) to -140dbm(bad)')
		response.append('#TYPE rsrp gauge')
		response.append('rsrp{'+devband+'} '+signal.get('rsrp'))
	# rsrq
	if signal.get('rsrq') is not None:
		response.append('#HELP rsrq Indicates quality of the received signal, and its range is typically -19.5dB(bad) to -3dB (good)')
		response.append('#TYPE rsrq gauge')
		response.append('rsrq{'+devband+'} '+signal.get('rsrq'))
	# rssi
	if signal.get('rssi') is not None:
		response.append('#HELP rssi Represents the entire received power including the wanted power from the serving cell as well as all co-channel power and other sources of noise in dBm')
		response.append('#TYPE rssi gauge')
		response.append('rssi{'+devband+'} '+signal.get('rssi'))
	# sinr
	if signal.get('sinr') is not None:
		response.append('#HELP sinr The signal-to-noise ratio of the given signal in dB')
		response.append('#TYPE sinr gauge')
		response.append('sinr{'+devband+'} '+signal.get('sinr'))

	s='\n'
	return s.join(response)


class GetHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
	
    timeout=5
    def do_GET(self):
        self.send_response(200)
        self.end_headers()
        self.wfile.write(prom_exporter().encode())



Handler = GetHandler
httpd = SocketServer.TCPServer(("", int(os.environ.get('PROM_PORT'))), Handler)

httpd.serve_forever()