Set scheduled charges on a single ZCell, to generally keep the battery full except for peak electricity usage times, and maintenance cycle days

sched.py

#!/usr/bin/env python3
#
# Set scheduled charges on a single ZCell, to generally keep the
# battery full except for peak electricity usage times, and maintenance
# cycle days.  Peak electricity usage times are 07:00-10:00 and
# 16:00-21:00 per Aurora Energy in Tasmania.  This will need adjusting
# by one hour if used during daylight savings time.
#
# This script is designed to be run on the GX from cron at 07:00 each
# day (just as peak electricity starts), because that makes the scheduled
# charges as simple as I could imagine.  We take ownership of charge
# schedules 4 and 5, configuring them to do The Right Thing(TM) for the
# following 24 hours, depending on whether today is a maintenance cycle
# day or not.  If for some reason there's an error querying the BMS,
# the scheduled charges are disabled, and this script will have no further
# effect until the user goes and manually re-enables those schedules.
# This also provides a convenient way to deactivate this script, without
# removing it from crontab - just disable either or both of scheduled
# charges 4 and 5, and this script will take its hands off.
#
# Run on the GX with:
#   TZ=$(dbus -y com.victronenergy.settings /Settings/System/TimeZone GetValue | tr -d "'") sched.py $BMS_IP
# to make sure datetime.now().weekday() is working off the local timezone,
# which is what the scheduler uses, vs. UTC, which is what the GX itself
# runs in.

import argparse
import dbus
import logging
import requests
from datetime import datetime

# Note that when log output lands in /var/log/messages via cron, lines are
# truncated at about 250 characters for some reason.
logging.basicConfig(level=logging.INFO, format='sched.py: %(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

HOURS = 60 * 60
DAYS = 60 * 60 * 24

def cgx_day(day):
    # CGX days are:
    #   0 Sunday
    #   1 Monday
    #   2 Tuesday
    #   3 Wednesday
    #   4 Thursday
    #   5 Friday
    #   6 Saturday
    #   7 Every Day
    #   8 Weekdays
    #   9 Weekends
    #
    # python's datetime.weekday() is Monday 0 - Sunday 6
    # So to get from python to CGX, we add 1 modulo 7
    return (day + 1) % 7

def next_day(day):
    # Yeah, I know this is the same code as the above function, but the
    # semantics are different ;-)
    return (day + 1) % 7

def is_maintenance_day():
    # If it's less than 24 hours until maintenance is due when
    # this script is run, we know today is a maintenance day
    status = requests.get(f'http://{args.bms_ip}:3000/rest/1.0/status').json()
    strip_pump_run_target = status['list'][0]['strip_pump_run_target']
    strip_pump_run_timer = status['list'][0]['strip_pump_run_timer']
    next_strip = strip_pump_run_target - strip_pump_run_timer
    return next_strip / DAYS < 1

def schedules_are_enabled():
    bus = dbus.SystemBus()
    sched_4_enabled = bus.get_object('com.victronenergy.settings',
        f'/Settings/CGwacs/BatteryLife/Schedule/Charge/3/Day').GetValue() >= 0
    sched_5_enabled = bus.get_object('com.victronenergy.settings',
        f'/Settings/CGwacs/BatteryLife/Schedule/Charge/4/Day').GetValue() >= 0
    return sched_4_enabled and sched_5_enabled

def set_charge_schedule(index, day, start, duration, soc):
    bus = dbus.SystemBus()
    # SetValue returns dbus.Int32(0) on success or dbus.Int32(-1) on failure,
    # which we are blissfully ignoring.
    o = bus.get_object('com.victronenergy.settings',
        f'/Settings/CGwacs/BatteryLife/Schedule/Charge/{index}/Day')
    o.SetValue(day)
    o = bus.get_object('com.victronenergy.settings',
        f'/Settings/CGwacs/BatteryLife/Schedule/Charge/{index}/Start')
    o.SetValue(start)
    o = bus.get_object('com.victronenergy.settings',
        f'/Settings/CGwacs/BatteryLife/Schedule/Charge/{index}/Duration')
    o.SetValue(duration)
    o = bus.get_object('com.victronenergy.settings',
        f'/Settings/CGwacs/BatteryLife/Schedule/Charge/{index}/Soc')
    o.SetValue(soc)

def set_scheduled_charges():
    if not schedules_are_enabled():
        logger.info("Schedules are disabled, will not make any changes")
        return

    today = cgx_day(datetime.now().weekday())
    tomorrow = next_day(today)
    try:
        if is_maintenance_day():
            logger.info("Setting maintenance day scheduled charges")
            # On maintenance days, set a scheduled charge from 13:00 for 3
            # hours with a 50% SoC limit...
            set_charge_schedule(3, today, 13 * HOURS, 3 * HOURS, 50)
            # ...and set an overnight charge starting at 03:00 for 4 hours
            # the next day (maintenance should be complete by then)
            set_charge_schedule(4, tomorrow, 3 * HOURS, 4 * HOURS, 100)
        else:
            logger.info("Setting NON-maintenance day scheduled charges")
            # On non-maintenance days, set scheduled charges from 10:00 for
            # 6 hours and 21:00 for 10 hours, so we're basically keeping the
            # battery charged all the time, except for peak electricity hours
            # (07:00-10:00 and 16:00-21:00).  Note that for simplicity this
            # assumes that all days have peak times.  That's not actually
            # true on Weekends, but it's just simpler to pretend it's true
            # rather than trying to set up schedules that account for the
            # difference between weekdays and weekends.
            set_charge_schedule(3, today, 10 * HOURS, 6 * HOURS, 100)
            set_charge_schedule(4, today, 21 * HOURS, 10 * HOURS, 100)
    except Exception as e:
        logger.error(e)
        logger.error("Disabling all scheduled charges")
        # If something breaks (e.g. can't talk to the BMS) fall back to
        # disabling our scheduled charges as the least worst course of
        # action
        set_charge_schedule(3, -today, 10 * HOURS, 6 * HOURS, 100)
        set_charge_schedule(4, -today, 21 * HOURS, 10 * HOURS, 100)

   
if __name__ == "__main__":
    parser = argparse.ArgumentParser("Set scheduled charges based on ZCell maintenance cycle")
    parser.add_argument("bms_ip", help="ZCell BMS IP address")
    args = parser.parse_args()
    set_scheduled_charges()