Quick and dirty script for setting weather forecast and time on X12 smartband

x12_sync.py

#!/usr/bin/python3

import sys
import time
import datetime
import subprocess
import logging
import requests

icon_codes = {
    1:  0,  # clear sky
    2:  0,  # few clouds
    3:  0,  # scattered clouds
    4:  1,  # broken clouds
    9:  2,  # shower rain
    10: 2,  # rain
    11: 2,  # thunderstorm
    13: 3,  # snow
    50: 1,  # mist
}

icon_names = {
    1:  'weather-clear',
    2:  'weather-few-clouds',
    3:  'weather-few-clouds',
    4:  'weather-few-clouds',
    9:  'weather-showers-scattered',
    10: 'weather-showers',
    11: 'weather-storm',
    13: 'weather-snow',
    50: 'weather-fog',
}

mac = sys.argv[1]
city = sys.argv[2]
api_key = sys.argv[3]

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s',
                    datefmt='%m-%d %H:%M:%S',
                    filemode='w')

weather_req = 'http://api.openweathermap.org/data/2.5/forecast?q={}&APPID={}&units=metric'.format(
    city, api_key)


def to_signed_byte(v):
    v = 2 * round(v)
    if (v > 127) or (v < -128):
        raise ValueError('The value must be betweeen -128 and 127')
    a = v
    if v < 0:
        a += 2**8
    return a


def decode_forecast(f):
    temp_fl = f['main']['feels_like']
    temp_low = f['main']['temp_min']
    temp = f['main']['temp']
    temp_high = f['main']['temp_max']
    descr = f['weather'][0]['description']
    icon = int(f['weather'][0]['icon'][:-1])
    dt_txt = f['dt_txt']

    forecast_msg = """Weather forecast for {}
    date: {}
    temperatures: {}C/{}C/{}C ({}C)
    forecast: {}"""\
        .format(city, dt_txt, temp_low, temp, temp_high, temp_fl, descr)

    for l in forecast_msg.splitlines():
        logging.debug(l)

    data = [to_signed_byte(temp), 0, icon_codes[icon]]
    return forecast_msg, "".join("%02X" % i for i in data)


retries = 5
while True:
    tmout = 10
    r = requests.get(weather_req).json()
    if r['cod'] == '200':
        break
    logging.warn('Weather service returned status: {}'.format(r['cod']))
    logging.info('Trying again in {} seconds'.format(tmout))
    time.sleep(tmout)
    retries = retries - 1
    if retries == 0:
        raise ConnectionRefusedError

# find forecast for at least hour ahead
nd = datetime.datetime.now()
mid = nd.replace(hour=0, minute=0, second=0, microsecond=0)
for forecast_1 in r['list']:
    fd = datetime.datetime.utcfromtimestamp(forecast_1['dt'])
    if (fd - nd) > datetime.timedelta(hours=1):
        break

for forecast_2 in r['list']:
    fd = datetime.datetime.utcfromtimestamp(forecast_2['dt'])
    if fd == mid + datetime.timedelta(days=1, hours=6):
        break

fmsg1, fdata1 = decode_forecast(forecast_1)
fmsg2, fdata2 = decode_forecast(forecast_2)

s = subprocess.run(['gatttool', '-b', mac, '--char-write-req',
                    '-a', '0x0009', '-n',
                    "ba20000b004c00050300300006" + fdata1 + fdata2])
s.check_returncode()
time.sleep(1)

dt = datetime.datetime.now() + datetime.timedelta(seconds=5)
logging.debug('Setting date and time to {}'.format(dt))
data = [dt.year % 100, dt.month, dt.day,
        dt.hour, dt.minute, dt.second, 0]

s = subprocess.run(['gatttool', '-b', mac, '--char-write-req',
                    '-a', '0x0009', '-n',
                    "ba20000c001e00010200200007" + "".join("%02X" % i for i in data)])
s.check_returncode()

s = subprocess.run(["notify-send", "-u", "low",
                    "-t", "10000",
                    'Weather info has been updated',
                    'Today: ' + fmsg1 + '\n'
                    'Tomorrow: ' + fmsg2])
s.check_returncode()