ungleich-public/lorawan
6
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Shift into library, add debug

Browse source
This commit is contained in:
Nico Schottelius 2016-11-02 17:42:39 +01:00
parent 8e0aa77715
commit b5bfcd3657
2 changed files with 11 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
receiver/lorautil.py
View file

@ -5,11 +5,17 @@
import psycopg2
import json
import logging
logging.basicConfig(format=FORMAT)
log = logging.getLogger("lorautil")
log.setLevel(logging.DEBUG)
dbname="lorawan"
def db_notify(provider, payload='', deveui=''):
notify="{}:{}".format(deveui, payload)
log.debug("Notify: {} {}".format(provider, notify))
try:
conn = psycopg2.connect("dbname={}".format(dbname))
cursor = conn.cursor()
@ -17,9 +23,10 @@ def db_notify(provider, payload='', deveui=''):
cursor.execute("select pg_notify (%s, %s)", (provider, notify))
cursor.connection.commit()
except Exception as e:
print("DB Notify failed: %s" % e)
log.error("DB Notify failed: %s" % e)
def db_insert_json(provider, data, payload='', deveui=''):
try:
conn = psycopg2.connect("dbname={}".format(dbname))
cursor = conn.cursor()
@ -28,7 +35,7 @@ def db_insert_json(provider, data, payload='', deveui=''):
conn.close()
except Exception as e:
print("DB Insert failed: %s" % e)
log.error("DB Insert failed: %s" % e)
def jsonToDict(self, data):
return json.loads(data)

16
receiver/swisscom-receiver.py
View file

@ -31,9 +31,6 @@ class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
lorautil.db_insert_json("swisscom", post_data, payload, deveui)
lorautil.db_notify("swisscom", payload, deveui)
def jsonToDict(self, data):
return json.loads(data)
def dictToPayload(self, thedict):
return thedict['DevEUI_uplink']['payload_hex']
@ -41,28 +38,19 @@ class testHTTPServer_RequestHandler(BaseHTTPRequestHandler):
return bytes.fromhex(myhex).decode('utf-8')
def dataToString(self, data):
mydict = self.jsonToDict(data)
mydict = lorautil.jsonToDict(data)
payload = self.dictToPayload(mydict)
return self.hexToString(payload)
def dataToDevEUI(self, data):
mydict = self.jsonToDict(data)
mydict = lorautil.jsonToDict(data)
eui = mydict['DevEUI_uplink']['DevEUI']
return eui
def devEUI(self, data):
root = ET.fromstring(data)
return root[1].text
def payload_hex(self, data):
root = ET.fromstring(data)
return root[7].text
def payload(self, data):
myhex = self.payload_hex(data)
return bytes.fromhex(myhex).decode('utf-8')
if __name__ == '__main__':
server_address = ('0.0.0.0', 8000)
httpd = HTTPServer(server_address, testHTTPServer_RequestHandler)