domoticz-Auroras/plugin.py

# Signal Level python plugin for Domoticz
#
# Author: fjumelle
#
#pylint: disable=line-too-long
"""
<plugin key="Auroras" name="Auroras" author="fjumelle" version="1.0.0" wikilink="" externallink="">
    <description>
        <h2>Auroras</h2><br/>
        Powered by Auroras.live
    </description>
    <params>
        <param field="Mode1" label="Latitude" width="100px" default="45"/>
        <param field="Mode2" label="Longitude" width="100px" default="5"/>
        <param field="Mode5" label="Pooling (min)" width="200px" default="60"/>
        <param field="Mode6" label="Logging Level" width="200px">
            <options>
                <option label="Normal" value="0" default="true"/>
                <option label="Verbose" value="1"/>
            </options>
        </param>
    </params>
</plugin>
"""

import math
import requests
from requests.adapters import HTTPAdapter, Retry

import Domoticz # type: ignore

if None is not None: #Fake statement to remove warning on global Domoticz variables  #NOSONAR
    Parameters = Parameters # type: ignore #NOSONAR  #pylint: disable=undefined-variable,self-assigning-variable
    Images = Images # type: ignore #NOSONAR  #pylint: disable=undefined-variable,self-assigning-variable
    Devices = Devices # type: ignore #NOSONAR  #pylint: disable=undefined-variable,self-assigning-variable

class BasePlugin(object):
    """plugin"""
    _DEFAULT_POOLING = 60*60 #1hour

    #Device ID
    _UNIT_BZ = 1
    _UNIT_DENSITY = 2
    _UNIT_SPEED = 3
    _UNIT_KP = 4
    _UNIT_LOCAL_VALUE = 10

    _UNITS = {
        # See https://www.domoticz.com/wiki/Developing_a_Python_plugin#Available_Device_Types
        # Unit, Name, Type, Subtype, Switchtype, Options, Icon, Used
        _UNIT_BZ: ["Bz", 243, 31, 0, {"Custom": "0;nt Bz"}, None, 1,],
        _UNIT_DENSITY: ["Density", 243,31, 0, {"Custom": "0;p/cm3"}, None, 1,],
        _UNIT_SPEED: ["Speed", 243, 31, 0, {"Custom": "0;km/s"}, None, 1,],
        _UNIT_KP: ["KP", 243, 31, 0, None, None, 1],
        _UNIT_LOCAL_VALUE: ["Local Value", 243, 6, 0, None, None, 1],
    }

    def __init__(self):
        self._pooling = 30
        self._pooling_steps = 1
        self._pooling_current_step = 1

    def on_start(self):
        """At statup"""
        # setup the appropriate logging level
        debuglevel = int(Parameters["Mode6"])
        if debuglevel != 0:
            Domoticz.Debugging(debuglevel)
        else:
            Domoticz.Debugging(0)

        # Polling interval = X sec
        try:
            pooling = int(Parameters["Mode5"])*60
        except ValueError:
            pooling = self._DEFAULT_POOLING
        self._pooling_steps = math.ceil(pooling/30)
        self._pooling_current_step = self._pooling_steps #To heartbeat at the first beat
        self._pooling = pooling // self._pooling_steps
        Domoticz.Heartbeat(self._pooling)

        # Create devices
        for u, unit in self._UNITS.items():
            if u not in Devices:
                Domoticz.Log(f"Create device #{u}: '{unit[0]}', type '{unit[1]}', subtype '{unit[2]}', switchtype '{unit[3]}', options '{unit[4]}', image '{unit[5]}'")

               #Image
                if unit[5] is None:
                    image = 0
                elif isinstance(unit[5], int):
                    image = unit[5]
                else:
                    if unit[5] not in Images:
                        Domoticz.DomoticzImage(unit[5]+".zip").Create()
                    image = Images[unit[5]].ID

                #Create device
                Domoticz.Device(Unit=u,
                                Name=unit[0],
                                Type=unit[1],
                                Subtype=unit[2],
                                Switchtype=unit[3],
                                Options=unit[4],
                                Image=image,
                                Used=unit[6]
                                ).Create()

    def on_stop(self):  #NOSONAR
        """Stop the plugin"""

    def on_heartbeat(self):
        """Time to heartbeat :)"""
        if self._pooling_current_step >= self._pooling_steps:
            Domoticz.Log(f"onHeartbeat {self._pooling_steps*self._pooling//60} min")
            query = {
                'lat': Parameters["Mode1"],
                'long': Parameters["Mode2"],
                'type': "all",
                'forecast': "false",
                'threeday': "false",
            }
            headers = {
                'User-Agent': 'domoticz'
            }

            location_session = requests.Session()
            retries = Retry(total=3, backoff_factor=0.1, status_forcelist=[500], raise_on_status=False)
            location_session.mount('https://', HTTPAdapter(max_retries=retries))
            try:
                response = location_session.get('https://api.auroras.live/v1', params=query, headers=headers, timeout=5)
                if response.status_code == requests.codes['ok']:
                    res = response.json()
                    bz = float(res["ace"]["bz"])
                    density = float(res["ace"]["density"])
                    speed = float(res["ace"]["speed"])
                    kp = float(res["ace"]["kp"])
                    local_value = int(res["probability"]["value"])

                    Devices[self._UNIT_BZ].Update(nValue=int(bz), sValue=str(bz))
                    Devices[self._UNIT_DENSITY].Update(nValue=int(density), sValue=str(density))
                    Devices[self._UNIT_SPEED].Update(nValue=int(speed), sValue=str(speed))
                    Devices[self._UNIT_KP].Update(nValue=int(kp), sValue=str(kp))
                    Devices[self._UNIT_LOCAL_VALUE].Update(nValue=int(local_value), sValue=str(local_value))
                else:
                    Domoticz.Error(f"HTTP Error: {response.status_code}")
            except requests.exceptions.RetryError as exc:
                Domoticz.Error("HTTP Error - " + str(exc))
            except Exception as exc:
                Domoticz.Error(str(exc))

            self._pooling_current_step = 1
        else:
            self._pooling_current_step = self._pooling_current_step + 1


_plugin = BasePlugin()

def onStart():  #NOSONAR #pylint: disable=invalid-name
    """OnStart"""
    _plugin.on_start()

def onStop():  #NOSONAR #pylint: disable=invalid-name
    """OnStop"""
    _plugin.on_stop()

def onHeartbeat():  #NOSONAR #pylint: disable=invalid-name
    """onHeartbeat"""
    _plugin.on_heartbeat()