rcvd-py/rcvd.py

#!/usr/bin/env python3
# rcvd.py: a port of RCVD.js library to Python 3
# RCVD is a small library for syncing time on BLE-enabled Casio watches
# This script also offers a standalone way of running the sync
# Tested on: GW-B5600BC, GMW-B5000D, OCW-T200S, GM-B2100BD, GA-B001G, DW-B5600G
# Created by Luxferre in 2022, ported to Python in 2024
# Released into public domain

import asyncio, time, ntplib
from bleak import BleakScanner, BleakClient
from bleak.exc import BleakDBusError

# time fetching part
# must return a dictionary with the following fields:
# year, month, date, hour, minute, second, day (0-6)
def fetchtime(params = {}):
  res = {}
  delta = 0.0 # in milliseconds
  if 'delta' in params:
    delta = float(params['delta']) / 1000
  if 'server' in params and params['server'] is not None: # NTP server set
    ntpver = 3
    if 'version' in params:
      ntpver = params['version']
    c = ntplib.NTPClient()
    resp = c.request(params['server'], version=ntpver)
    unixtm = resp.tx_time
  else: # use current system time by default
    unixtm = time.time()
  tm = time.localtime(unixtm + delta) # account for delta
  res['year'] = tm.tm_year
  res['month'] = tm.tm_mon
  res['date'] = tm.tm_mday
  res['hour'] = tm.tm_hour
  res['minute'] = tm.tm_min
  res['second'] = tm.tm_sec
  res['day'] = tm.tm_wday
  return res

# device interaction part

# generate full Casio-specific UUID from the short one
def fullid(s):
  return '26eb00' + s + '-b012-49a8-b1f8-394fb2032b0f'

# populate constants
CASIO_SYNC_SRV = fullid('0d') # for service discovery
READER_CHAR = fullid('2c')    # for parameter reading
WRITER_CHAR = fullid('2d')    # for parameter writing

# the queue for GATT notification data
notiqueue = asyncio.Queue()

# GATT notification handler
def readfiller(sender, data):
  global notiqueue
  notiqueue.put_nowait(data)

# execute a read command:
# first write the value without response to READER_CHAR
# then read the result from the WRITER_CHAR (!) notifier
async def exec_read(client, *cmdargs):
  global notiqueue
  await client.write_gatt_char(READER_CHAR, bytes(cmdargs), response=False)
  res = await notiqueue.get()
  return res

# execute a write command:
# write to WRITER_CHAR with an immediate response
async def exec_write(client, cmd: bytearray):
  res = await client.write_gatt_char(WRITER_CHAR, cmd, response=True)
  return res

# handshake sequence, returns the watch model
async def handshake(client):
  rawmodel = await exec_read(client, 0x23) # read the raw model
  model = rawmodel[1:].decode('utf-8').rstrip('\x00') # human-readable model
  features = await exec_read(client, 0x10) # read feature setting
  if features[8] < 2: # we entered full setting mode, sync app info first
    info = await exec_read(client, 0x22) # detect if BLE has been reset
    # it has been reset if it consists of 11 FF bytes and then 00
    if info[:12] == b'\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x00':
      restorecmd = b'\x22\x34\x88\xf4\xe5\xd5\xaf\xc8\x29\xe0\x6d\x02'
      await exec_write(client, restorecmd)
  return model

# presync property cycling required to init the time sync process
# these parameters need to be read from the watch and then written back
# the set of the parameters is model-specific
# this function returns the final precycle list based on the mode
def precycle_prop_list(model: str):
  # full property list for primary targets, GW-B/GMW-B models
  props = [[30,0], [30,1], [30,2], [30,3], [30,4], [30,5],
           [31,0], [31,1], [31,2], [31,3], [31,4], [31,5]]
  if model.find('OCW') > 5 or model.find('GWR-B1000') > 5 \
      or model.find('ABL-100') > 5 or model.find('GD-B500') > 5:
    props = [[30,0], [30,1]]
  elif (model.find('B2100') > 5 or model.find('B001') > 5 \
        or model.find('DW-B5600') > 5 or model.find('GST-B') > 5 \
        or model.find('MSG-B') > 5 or model.find('ECB') > 5):
    props = [[30,0], [30,1], [31,0], [31,1]]
  return [[29,0], [29,2], [29,4]] + props # concatenate the prop lists

# sync the watch from here
async def dosync(client, timeparams):
  global notiqueue
  print('Starting handshake...')
  await client.start_notify(WRITER_CHAR, readfiller) # set up notifications
  model = await handshake(client) # run the handshake and detect watch model
  print('Connected to %s, preparing to sync...' % model)
  precycle_props = precycle_prop_list(model)
  for prop in precycle_props: # just read and write again
    data = await exec_read(client, prop[0], prop[1])
    await exec_write(client, data)
  print('Fetching the time...')
  timeparts = fetchtime(timeparams)
  print('Syncing the time...')
  timelist = [9, # sync time command
    timeparts['year'] & 255, timeparts['year'] >> 8,
    timeparts['month'], timeparts['date'],
    timeparts['hour'], timeparts['minute'], timeparts['second'],
    timeparts['day'], 0, 0, 1]
  try: # some models disconnect prematurely as soon as sync OK
    await exec_write(client, bytes(timelist))
  except (EOFError, BleakDBusError):
    pass
  print('Time synced')


# the process starts from here
# accepts fetchtime parameters and BLE scan timeout
async def syncstart(timeparams):
  # list of IDs to scan
  devices = await BleakScanner.discover(timeout=timeparams['timeout'],
    return_adv=True,
    service_uuids=['00001804-0000-1000-8000-00805f9b34fb', CASIO_SYNC_SRV])
  for d in devices: # enumerate found watches
    async with BleakClient(str(d)) as client:
      await dosync(client, timeparams) # init the client by BLE MAC

if __name__ == '__main__': # application entry point
  from argparse import ArgumentParser
  parser = ArgumentParser(description='RCVD: an opensource time synchronizer for BLE-enabled Casio watches', epilog='(c) Luxferre 2024 --- No rights reserved <https://unlicense.org>')
  parser.add_argument('-t', '--timeout', type=int, default=4, help='BLE scanning timeout (in seconds, default 4)')
  parser.add_argument('-d', '--delta', type=int, default=500, help='Manual delta correction (in ms, must be determined individually, 500 ms by default)')
  parser.add_argument('-s', '--ntp-server', type=str, default=None, help='NTP server to sync from (if not specified then will sync from the local system time)')
  parser.add_argument('-n', '--ntp-version', type=int, default=4, help='NTP protocol version to use (default 4)')
  args = parser.parse_args()
  params = { # populate parameters from the command line
    'server': args.ntp_server, 'version': args.ntp_version,
    'delta': args.delta, 'timeout': args.timeout
  }
  if params['version'] < 1 or params['version'] > 4:
    params['version'] = 4
  print('RCVD by Luxferre\nPress sync button on the watch ASAP to connect')
  asyncio.run(syncstart(params)) # start the process