Skip to content

Instantly share code, notes, and snippets.

@Terrance
Created March 31, 2024 10:57
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Terrance/f9c00a02947c01a3fb7bf82854980893 to your computer and use it in GitHub Desktop.
Save Terrance/f9c00a02947c01a3fb7bf82854980893 to your computer and use it in GitHub Desktop.
Minimal, single-user μlogger server (https://github.com/bfabiszewski/ulogger-server) replacement, to accept location updates from μlogger for Android.
#!/usr/bin/env python3
"""
Credentials file (path in `ULOGGER_CREDS`):
```
<username>:<password>
```
Database schema (database name in `ULOGGER_DB`):
```
CREATE TABLE public.loc_track (
id integer NOT NULL,
name character varying(255) NOT NULL,
comment character varying(1024),
PRIMARY KEY (id)
);
CREATE TABLE public.loc_position (
id integer NOT NULL,
"time" timestamp(0) without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL,
track_id integer NOT NULL,
latitude double precision NOT NULL,
longitude double precision NOT NULL,
altitude double precision,
speed double precision,
bearing double precision,
accuracy integer,
provider character varying(100) DEFAULT NULL::character varying,
comment character varying(255) DEFAULT NULL::character varying,
PRIMARY KEY (id),
FOREIGN KEY (track_id) REFERENCES loc_track (id)
);
```
"""
from contextlib import contextmanager
import os
import sys
from flask import abort, Flask, request, session
from psycopg2.pool import SimpleConnectionPool
with open(os.environ["ULOGGER_CREDS"]) as f:
USER, PWD = f.read().strip().split(":", 1)
app = Flask(__name__)
app.secret_key = os.urandom(32)
pool = SimpleConnectionPool(1, 8, database=os.environ["ULOGGER_DB"])
@contextmanager
def database():
db = pool.getconn()
cur = db.cursor()
try:
yield db, cur
db.commit()
except:
db.rollback()
raise
finally:
cur.close()
pool.putconn(db)
@app.route("/client/index.php", methods=["POST"])
def client():
action = request.form["action"]
print("Action: %s" % action)
if action == "auth":
user = request.form["user"]
pwd = request.form["pass"]
if user == USER and pwd == PWD:
session["user"] = 1
return {"error": False}
else:
return {"error": True, "message": "Unauthorized"}, 401
elif "user" not in session:
return {"error": True, "message": "Unauthorized"}, 401
elif action == "addtrack":
try:
data = (request.form["track"], request.form.get("comment"))
except KeyError:
return {"error": True, "message": "Missing required parameter"}
with database() as (_, cur):
cur.execute("""
INSERT INTO loc_track
(name, comment)
VALUES
(%s, %s)
RETURNING id
""", data)
row = cur.fetchone()[0]
return {"error": False, "trackid": row}
elif action == "addpos":
try:
data = (request.form["time"],
request.form["trackid"],
request.form["lat"],
request.form["lon"],
request.form.get("altitude"),
request.form.get("speed"),
request.form.get("bearing"),
int(float(request.form["accuracy"])),
request.form.get("provider"),
request.form.get("comment"))
except KeyError:
return {"error": True, "message": "Missing required parameter"}
with database() as (_, cur):
cur.execute("""
INSERT INTO loc_position
(time, track_id, latitude, longitude, altitude, speed, bearing, accuracy, provider, comment)
VALUES
(to_timestamp(%s), %s, %s, %s, %s, %s, %s, %s, %s, %s)
""", data)
return {"error": False}
else:
abort(400)
if __name__ == "__main__":
try:
port = int(sys.argv[1])
except (IndexError, ValueError):
port = None
app.run(port=port, debug=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment