api_whattheshop.py

# Copyright 2023 Le Filament (https://le-filament.com)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)

import json
import logging
from datetime import datetime

from dateutil.relativedelta import relativedelta

from odoo import _, api, models
from odoo.exceptions import UserError

from odoo.addons.api_connector.tools.date_utils import utc_to_local

_logger = logging.getLogger(__name__)


class ApiWhatTheShop(models.AbstractModel):
    _name = "api.whattheshop"
    _inherit = ["api.connector"]
    _description = "API What The Shop"
    _backend_name = "backend.whattheshop"

    # ------------------------------------------------------
    # Auth
    # ------------------------------------------------------
    def _get_token(self):
        backend = self._get_backend()
        try:
            response = self.api_post(
                url=backend.url + "/1/user/auth",
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                verify=backend.ssl,
                timeout=backend.timeout,
                data=dict(
                    login=backend.login,
                    password=backend.password,
                    isEncryptedPassword=0,
                ),
            )
            token = response.get("token")
            if token:
                backend.token = token
                backend.token_expiration = datetime.now() + relativedelta(hours=1)
                return token
        except Exception as e:
            raise e

    # ------------------------------------------------------
    # Common functions
    # ------------------------------------------------------
    def _get_wts(self, url, params=None):
        backend = self._get_backend()
        if (
            not backend.token
            or not backend.token_expiration
            or datetime.now() > backend.token_expiration
        ):
            self._get_token()
        try:
            response = self.api_get(
                url=backend.url + url,
                headers={"wts_token": backend.token},
                verify=backend.ssl,
                timeout=backend.timeout,
                params=params,
            )
            if response.get("error"):
                raise UserError(_("API Error: \n%s") % (response.get("error"),))
            return response.get("result")
        except Exception as e:
            raise e

    # ------------------------------------------------------
    # API User functions
    # ------------------------------------------------------
    def _get_zones(self):
        return self._get_wts(url="/1/user/zones")

    def _get_zone_infos(self, zone_id):
        """
        :param zone_id: integer unique identifier of zone
        """
        return self._get_wts(
            url="/1/user/zoneinfos",
            params={"zoneId": zone_id},
        )

    def _get_devices(self):
        return self._get_wts(url="/1/user/devices")

    # ------------------------------------------------------
    # API Sensor functions
    # ------------------------------------------------------
    def _get_wifi_datas(
        self,
        url,
        zone_ids,
        start_date,
        end_date,
        opening_time=False,
        timezone="Europe/Paris",
    ):
        """
        :param string url : url endpoint
        :param list zone_ids : liste des zones de la requête
        :param datetime start_date : UTC start period
        :param datetime end_date : UTC end period
        :param boolean opening_time : use opening times period
        """
        start_date = utc_to_local(start_date, timezone)
        end_date = utc_to_local(end_date, timezone)
        params = {
            "zones": json.dumps(zone_ids),
            "startDate": start_date.strftime("%Y-%m-%d %H:%M:%S"),
            "endDate": end_date.strftime("%Y-%m-%d %H:%M:%S"),
            "days": self._set_days_interval(start_date, end_date),
            "openingTimes": 1 if opening_time else 0,
        }
        return self._get_wts(url="/1/analytic" + url, params=params)

    @api.model
    def _get_sensor_datas(
        self,
        url,
        sensor_id,
        start_date,
        end_date,
        period=False,
        mode="all",
        use_opening=True,
        timezone="Europe/Paris",
    ):
        """
        :param string url : url endpoint
        :param integer sensor_id : WTS sensor id
        :param datetime start_date : UTC start period
        :param datetime end_date : UTC end period
        :param string mode : in/out/all
        :param boolean use_opening : use opening times period
        """
        start_date = utc_to_local(start_date, timezone)
        end_date = utc_to_local(end_date, timezone)
        params = {
            "sensorId": sensor_id,
            "startDate": start_date.strftime("%Y-%m-%d %H:%M:%S"),
            "endDate": end_date.strftime("%Y-%m-%d %H:%M:%S"),
            "mode": mode,
            "useOpenings": 1 if use_opening else 0,
        }
        return self._get_wts(url="/1/analytic" + url, params=params)

    # ------------------------------------------------------
    # Common functions
    # ------------------------------------------------------
    def _set_days_interval(self, start_date, end_date):
        delta = (end_date - start_date).days
        if delta >= 7:
            days = list(range(1, 8))
        else:
            if end_date.weekday() >= start_date.weekday():
                days = list(range(start_date.weekday() + 2, end_date.weekday() + 3))
            else:
                days = list(range(1, end_date.weekday() + 2)) + list(
                    range(start_date.weekday(), 8)
                )
        return json.dumps(days)