tibber2pvoutput

#!/usr/bin/env python3

# Copyright (c) 2020 Erik Johansson <erik@ejohansson.se>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA

import tibber
from pvoutput import PvOutput, DefaultOutput

from datetime import date, timedelta
from getpass import getpass

import argparse
import configparser
import logging

SHOULDER_HOURS_1 = slice(6, 8)
PEAK_HOURS = slice(8, 17)
HIGH_SHOULDER_HOURS = slice(17, 20)
SHOULDER_HOURS_2 = slice(20, 23)


def get_outputs_missing_import_export(pvoutput):
    yesterday = date.today() - timedelta(days=1)

    missing = []

    for output in pvoutput.get_output(date_to=yesterday, limit=35):
        if (
            output.import_peak is None
            or output.exported is None
            or (
                output.import_peak
                + output.import_off_peak
                + output.import_shoulder
                + output.import_high_shoulder
            )
            == 0
        ):
            missing.append(output)

    return missing


def main():
    logging.basicConfig(format="%(asctime)s %(levelname)s: %(message)s")

    parser = argparse.ArgumentParser(description="Connect Tibber to PVoutput.org")
    parser.add_argument("config", help="Configuration file to use")
    parser.add_argument(
        "-d", "--debug", help="Enable debug output", action="store_true"
    )
    parser.add_argument(
        "-n", "--dry-run", help="Don't send any data", action="store_true"
    )
    args = parser.parse_args()

    if args.debug:
        logging.getLogger().setLevel(logging.DEBUG)

    config = configparser.ConfigParser()
    config["tibber"] = {}
    config["pvoutput"] = {}
    config.read(args.config)

    modified = False
    if not config["tibber"].get("token"):
        config["tibber"]["token"] = input("Tibber access token: ")
        modified = True

    if not config["pvoutput"].get("apikey"):
        config["pvoutput"]["apikey"] = input("PVOutput API key: ")
        modified = True
    if not config["pvoutput"].get("systemid"):
        config["pvoutput"]["systemid"] = input("PVOutput System Id: ")
        modified = True

    if modified:
        with open(args.config, "w") as configfile:
            config.write(configfile)

    pvoutput = PvOutput(
        config["pvoutput"]["apikey"], config["pvoutput"]["systemid"], args.dry_run
    )

    missing = get_outputs_missing_import_export(pvoutput)
    logging.debug(
        "Missing export/import for: %s", [o.date.strftime("%Y-%m-%d") for o in missing]
    )
    if not missing:
        return

    client = tibber.Client(config["tibber"]["token"])
    home = client.get_homes()[0]

    first_date = min(o.date for o in missing)

    imports = {}
    exports = {}

    prev_date = None
    for consumed in home.consumption(tibber.EnergyResolution.HOURLY, reverse=True):
        day = consumed.date.date()
        if day < first_date:
            break
        data = imports.setdefault(day, list())
        if not data:
            data.extend(None for _ in range(24))
        if prev_date is not None and prev_date.utcoffset() != consumed.date.utcoffset():
            for hour in range(consumed.date.hour, prev_date.hour):
                data[hour] = 0
            data[consumed.date.hour] += consumed.consumption
        else:
            data[consumed.date.hour] = consumed.consumption
        prev_date = consumed.date

    for produced in home.production(tibber.EnergyResolution.DAILY, reverse=True):
        day = produced.date.date()
        exports[day] = produced.production
        if day <= first_date:
            break

    for output in missing:
        imp = imports.get(output.date)
        exp_sum = exports.get(output.date)

        if imp is None or None in imp or exp_sum is None:
            continue

        imp_sum = sum(imp)

        logging.debug(
            "Found export %d Wh and import %d Wh for %s",
            exp_sum,
            imp_sum,
            output.date.strftime("%Y-%m-%d"),
        )

        imp_peak = sum(imp[PEAK_HOURS])
        imp_shoulder = sum(imp[SHOULDER_HOURS_1]) + sum(imp[SHOULDER_HOURS_2])
        imp_high_shoulder = sum(imp[HIGH_SHOULDER_HOURS])
        imp_off_peak = imp_sum - sum([imp_peak, imp_shoulder, imp_high_shoulder])
        output = DefaultOutput._replace(
            date=output.date,
            exported=exp_sum,
            import_peak=imp_peak,
            import_off_peak=imp_off_peak,
            import_shoulder=imp_shoulder,
            import_high_shoulder=imp_high_shoulder,
            consumption=imp_sum + (output.generated - exp_sum),
        )
        pvoutput.add_output(output)


if __name__ == "__main__":
    main()