Script to generate zigbee network backup json from a zigpy sqlite database

zigpy-net-backup.py

#!/usr/bin/env python

import argparse
import asyncio
import json
import logging
import pathlib

import zigpy.application
import zigpy.backups
import zigpy.util
from zigpy.config import (
    CONF_DATABASE,
    CONF_DEVICE,
    CONF_DEVICE_PATH,
    CONF_OTA,
    CONF_OTA_ENABLED,
)

LOGGER = logging.getLogger(__name__)


class App(zigpy.application.ControllerApplication):
    async def send_packet(self, packet):
        pass

    async def connect(self):
        pass

    async def disconnect(self):
        pass

    async def start_network(self):
        pass

    async def force_remove(self, dev):
        pass

    async def add_endpoint(self, descriptor):
        pass

    async def permit_ncp(self, time_s=60):
        pass

    async def permit_with_link_key(self, net, link_key, time_s=60):
        pass

    async def reset_network_info(self):
        pass

    async def write_network_info(self, *, network_info, node_info):
        pass

    async def load_network_info(self, *, load_devices=False):
        pass


def print_backup_list(backups):
    print(f"Found {len(backups)} network backups:")
    for i, backup in enumerate(backups):
        print(
            f"{i+1}\t{backup.backup_time} (Pan Id: {backup.network_info.pan_id}, Ext Pan Id: {backup.network_info.extended_pan_id}, Channel: {backup.network_info.channel})"
        )


async def do_backup(backups, number, output):
    if len(backups) == 0:
        print(f"No backups exist")
        return
    try:
        if number < 0:
            number = len(backups) - 1
        backup = backups[number]
    except IndexError:
        print(f"Backup with index {number + 1} does not exist")
        return

    with output.open("w") as io:
        print(f"Writing backup index {number + 1} json to {output}")
        backup_json = backup.as_open_coordinator_json()
        json.dump(backup_json, io, indent=2)


async def main(args):

    LOGGER.debug("Starting App")
    app = await App.new(
        {
            CONF_DATABASE: str(args.database),
            CONF_DEVICE: {CONF_DEVICE_PATH: "/dev/null"},
            CONF_OTA: {CONF_OTA_ENABLED: False},
        },
        auto_form=False,
        start_radio=False,
    )
    LOGGER.debug("App started")

    backups = app.backups.backups
    print_backup_list(backups)

    if args.output is not None:
        number = args.number
        if number > 0:
            number -= 1
        else:
            number = -1
        await do_backup(backups, number, args.output)

    LOGGER.debug("Shutting down App")
    await app.shutdown()
    LOGGER.debug("App shut down")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-d",
        "--database",
        metavar="FILE",
        type=pathlib.Path,
        default="data/zigbee.db",
        help="Path to sqlite zigbee.db file",
    )
    parser.add_argument(
        "-o",
        "--output",
        metavar="FILE",
        type=pathlib.Path,
        help="Path to write backup network data json",
    )
    parser.add_argument(
        "-n",
        "--number",
        metavar="NUMBER",
        type=int,
        default=-1,
        help="Index of which backup to save, default is most recent",
    )
    parser.add_argument(
        "-v", "--verbose", action="store_true", help="Verbose log output"
    )
    args = parser.parse_args()

    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)

    if not args.database.is_file():
        print(f"Database file {args.database} does not exist")
        if str(args.database) == parser.get_default("database"):
            parser.print_help()
        exit(1)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(args))