NTPd GPSd OLED display script

ntp-gps-oled.py

#!/usr/bin/python3

import time
from luma.core.interface.serial import i2c
from luma.core.render import canvas
from luma.oled.device import ssd1306
from PIL import ImageFont
import threading
import ntpq
import gps

serial = i2c(port=1, address=0x3C)
device = ssd1306(serial, 128, 32)
device.contrast(1)

(_, _, device_width, device_height) = device.bounding_box

font_default = ImageFont.load_default()
font_small = ImageFont.load("slkscr.pil")

def text_regular(draw, pos, text, anchor="l"):
    bbox = font_default.getbbox(text)

    if anchor == "m":
        dpos = (pos[0] - bbox[2]/2, pos[1])
    elif anchor == "r":
        dpos = (pos[0] - bbox[2], pos[1])
    else:
        dpos = pos

    draw.text(dpos, text, fill="white", font=font_default)
    return (bbox[0] + dpos[0], bbox[1] + dpos[1], bbox[2] + dpos[0], bbox[3] + dpos[1] - 1)

def text_small(draw, pos, text):
    bbox = font_small.getbbox(text)
    draw.text(pos, text, fill="white", font=font_small)
    return (bbox[0] + pos[0], bbox[1] + pos[1], bbox[2] + pos[0], bbox[3] + pos[1] - 2)

def bbox_get(bbox, anchor_or_xp, yp = None):
    if type(anchor_or_xp) is str:
        if len(anchor_or_xp) != 2 or anchor_or_xp[0] not in ['t','m','b'] or anchor_or_xp[1] not in ['l','c','r']:
            raise ValueError('Invalid anchor: ' + anchor_or_xp)

        anchor_mapping = {'t': 0, 'm': 1/2, 'b': 1, 'l': 0, 'c': 1/2, 'r': 1}
        yp = anchor_mapping[anchor_or_xp[0]]
        xp = anchor_mapping[anchor_or_xp[1]]
    elif type(anchor_or_xp) in [float, int] and type(yp) in [float, int]:
        yp = yp
        xp = anchor_or_xp
    else:
        raise ValueError('Invalid arguments')

    (x1, y1, x2, y2) = bbox
    ret = (x1 + (x2 - x1) * xp, y1 + (y2 - y1) * yp)
    return ret

def update_display(draw, info):
    pos = (device_width/2, -2) # Start in middle
    pos = bbox_get(text_regular(draw, pos, info["time"] + " " + info["date"], "m"), "bl")
    pos = (0, pos[1])
    subs_y = pos[1] # Track bottom of top line
    pos = bbox_get(text_small(draw, pos, "Sats:" + info["sats"]), "bl")
    pos = bbox_get(text_small(draw, pos, "Fix:" + info["fix"]), "bl")
    pos = bbox_get(text_small(draw, pos, "Rem:" + info["remote"]), "br")
    room_for_reach = pos[0] < device_width/2

    pos = (device_width/2, subs_y) # Move to second column, below top line
    pos = bbox_get(text_small(draw, pos, "Offs:" + info["offset"]), "bl")
    pos = bbox_get(text_small(draw, pos, "Jitt:" + info["jitter"]), "bl")
    if room_for_reach:
        text_small(draw, pos, "Reach:" + info["reach"])

ntp_shm2 = None
def ntp_thread():
    global ntp_shm2

    while True:
        print("NTP updating")
        ntp_shm2 = ntpq.get(select='*')

        if ntp_shm2["offset"][0] != "-":
            ntp_shm2["offset"] = " " + ntp_shm2["offset"]

        print("NTP thread sleep")
        time.sleep(5)

gps_session = gps.gps(mode=gps.WATCH_ENABLE|gps.WATCH_NEWSTYLE)
def gps_thread():
    global gps_device

    while True:
        data = gps_session.next()

def run():
    t1 = threading.Thread(target=ntp_thread, daemon=True)
    t1.start()

    t2 = threading.Thread(target=gps_thread, daemon=True)
    t2.start()

    info = {
        "time": "?",
        "date": "?",
        "sats": "?",
        "fix": "?",
        "remote": "?",
        "offset": "?",
        "jitter": "?",
        "reach": "?"
    }

    while True:
        info["time"] = time.strftime("%H:%M:%S")
        info["date"] = time.strftime("%d-%m-%Y")

        if ntp_shm2:
            info["remote"] = ntp_shm2["remote"]
            info["offset"] = ntp_shm2["offset"]
            info["jitter"] = ntp_shm2["jitter"]
            info["reach"] = ntp_shm2["reach"]

        if (gps.MODE_SET & gps_session.valid):
            modes = ["?", "NO", "2D", "3D"]
            info["fix"] = modes[gps_session.fix.mode]
            info["sats"] = str(gps_session.satellites_used)

        if info["remote"] == 'SHM(1)':
            info["remote"] = "GPS"
        elif info["remote"] == 'SHM(2)':
            info["remote"] = "GPS:PPS"

        with canvas(device) as draw:
            update_display(draw, info)

        #print(time.time())
        time.sleep(.5)

if __name__ == "__main__":
    try:
        run()
    except KeyboardInterrupt:
        print('')

    gps_session.close()
    exit(0)