Phoscon sensor rename utility

README.md

Phoscon device rename utility

This is a mass device rename utility for Phoscon. You can set unique names for all sensors and lights not only per devices.

Usage

python3 rename_sensors.py 'http://phoston_app_url:2080' 'API_KEY'

1. The first run will dump the current config from the Phoscon to devices.json
2. Edit the names in the devices.json w/ your favorite editor
3. Run again the script

Every run ends with a fresh dump, so the devices.json contains the committed state!

Getting API key

curl \
  -H 'Content-Type: application/json' \
  -X POST \
  -d '{"devicetype": "foo_bar"}' \
  http://phoston_app_url:2080/api/

A generated and customized devices.json (for example)

{
  "00:15:8d:00:02:83:a3:ae": {
    "ZHAWater": "i1Kitchen_DishWasher_XiaWat"
  },
  "00:15:8d:00:02:b6:a2:aa": {
    "ZHAVibration": "o1Balcony_Bicycle_XiaVibr"
  },
  "00:15:8d:00:02:f9:a9:ad": {
    "ZHASwitch": "i1Wc_Light_XiaBtn"
  },
  "00:15:8d:00:03:2c:a3:ab": {
    "ZHAPresence": "i1Wc_XiaMv_M"
  },
  "00:15:8d:00:03:61:ae:a9": {
    "ZHAFire": "i1Storage_XiaSmo"
  },
  "00:15:8d:00:03:6b:a1:ac": {
    "ZHAHumidity": "i1Living_XiaMul_H",
    "ZHAPressure": "i1Living_XiaMul_P",
    "ZHATemperature": "i1Living_XiaMul_T"
  },
  "00:15:8d:00:04:14:ab:af": {
    "ZHALightLevel": "i2Wc_XiaMvLi_L",
    "ZHAPresence": "i2Wc_XiaMvLi_M"
  },
  "00:15:8d:00:04:1c:a5:af": {
    "ZHAOpenClose": "i1Kitchen_WinDining_XiaOp"
  },
  "04:cf:8c:df:3c:77:a6:a6": {
    "ZHALightLevel": "o1Living_WinBig_XiaLi"
  },
  "90:fd:9f:ff:fe:dc:a7:a9": {
    "ZHASwitch": "x01_IkeaBtn"
  }
}

rename_devices.py

#!/usr/bin/env python3

# Source: https://gist.github.com/andras-tim/469899e901d78e41b88c7e7e57639f4f

import json
import logging
import os
import sys
from collections import defaultdict
from urllib.parse import urljoin

import requests

CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'devices.json')

logger = logging.getLogger()


def main(app_url: str, api_key: str):
    logging.basicConfig(level=logging.INFO, format='%(message)s')
    phoscon = Phoscon(app_url, api_key)

    if os.path.exists(CONFIG_PATH):
        load_config(phoscon)

    save_config(phoscon)


def save_config(phoscon: 'Phoscon'):
    logger.info('Generating config')
    devices = defaultdict(dict)
    extend_with_parsed_devices(devices, phoscon.list_sensors())
    extend_with_parsed_devices(devices, phoscon.list_lights())

    logger.info('Saving config to {!r}'.format(CONFIG_PATH))
    with open(CONFIG_PATH, 'w') as fd:
        json.dump(devices, fd, indent=2, sort_keys=True)
        fd.write('\n')


def extend_with_parsed_devices(sensors_by_mac_and_type: dict, raw_devices: dict):
    for device in raw_devices.values():
        mac = Phoscon.get_mac(device)
        sensors_by_mac_and_type[mac][device['type']] = device['name']


def load_config(phoscon: 'Phoscon'):
    logger.info('Loading config from {!r}'.format(CONFIG_PATH))
    with open(CONFIG_PATH, 'r') as fd:
        config = json.load(fd)

    apply_dev_config(phoscon, config, phoscon.list_lights(), 'lights')
    apply_dev_config(phoscon, config, phoscon.list_sensors(), 'sensors')


def apply_dev_config(phoscon: 'Phoscon', config: dict, devices: dict, api_endpoint: str):
    logger.info('Applying config on {}'.format(api_endpoint))

    for index, device in devices.items():
        mac = Phoscon.get_mac(device)

        device_config = config.get(mac)
        if device_config is None:
            continue

        desired_name = device_config.get(device['type'])
        if desired_name is None:
            continue

        if device['name'] != desired_name:
            logger.info('Set {!r} device name from {!r} to {!r}'.format(index,  device['name'], desired_name))
            phoscon.set_name(api_endpoint, index, desired_name)


class Phoscon:
    def __init__(self, app_url: str, api_key: str):
        self.__base_url = urljoin(app_url, '/api/{}/'.format(api_key))
        self.__session = requests.Session()

    def __request(self, method, url_fragment, *args, **kwargs):
        response = self.__session.request(
            method,
            urljoin(self.__base_url, url_fragment),
            *args,
            **kwargs
        )
        response.raise_for_status()

        return response.json()

    def list_lights(self) -> dict:
        logger.info('Getting lights')
        return self.__request('GET', 'lights')

    def list_sensors(self) -> dict:
        logger.info('Getting sensors')
        return self.__request('GET', 'sensors')

    def set_name(self, api_endpoint: str, index: int, new_name: str):
        if len(new_name) > 32:
            raise ValueError('Name of {!r} sensor ({!r}) is longer than 32 characters'.format(index, new_name))

        return self.__request('PUT', '{}/{}'.format(api_endpoint, index), json={'name': new_name})

    @classmethod
    def get_mac(cls, sensor: dict) -> str:
        return sensor['uniqueid'][:23]


if __name__ == '__main__':
    main(*sys.argv[1:])