#!/usr/bin/python3

# Copyright (c) 2022-2025, Digi International, Inc.
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
# AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
# LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
# OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
# PERFORMANCE OF THIS SOFTWARE.

import argparse
import base64
import cgi
import errno
import hashlib
import http.server
import json
import logging
import os
import platform
import re
import shutil
import signal
import socketserver
import stat
import subprocess
import time
from json import JSONDecodeError

from digi.apix.bluetooth import BluetoothException, BluetoothDevice, BluetoothProfile
from digi.apix.exceptions import DigiAPIXException
from digi.apix.network import IPMode, NetworkException, NetStatus, NetworkInterface, NetworkProfile
from digi.apix.wifi import SecurityMode, WifiException, WifiInterface, WifiProfile
from logging.handlers import SysLogHandler
from threading import Thread, Lock

from digi.ccble.exceptions import BluetoothNotSupportedException, ConnectCoreBLEException
from digi.ccble.service import BLEInterface, ConnectCoreBLEService

# Constants.
APP_NAME = "Demo server"

PORT = 9090

EMMC_SIZE_FILE = "/sys/class/mmc_host/mmc0/mmc0:0001/block/mmcblk0/size"
NAND_SIZE_FILE = "/proc/mtd"

SIZE_KB = "KB"
SIZE_MB = "MB"
SIZE_GB = "GB"

DIVIDERS = {SIZE_KB: 1, SIZE_MB: 2, SIZE_GB: 3}

ZERO_MAC = "00:00:00:00:00:00"
ZERO_IP = "0.0.0.0"
NOT_AVAILABLE = "N/A"

ELEMENT_BLUETOOTH = "bluetooth"
ELEMENT_ETHERNET = "ethernet"
ELEMENT_INFO = "info"
ELEMENT_PING = "ping"
ELEMENT_RESTART_BLUETOOTH = "restart-bluetooth"
ELEMENT_TEST_CONNECTIVITY = "test-connectivity"
ELEMENT_WIFI = "wifi"

PREFIX_ETHERNET = "eth"
PREFIX_WIFI = "wlan"

BT_PASSWORD_FILE = "/etc/bt.pwd"
BT_PASSWORD_DEFAULT = "1234"
BT_PASSWORD_LENGTH = 16
BT_PASSWORD_SERIAL_LENGTH = 6

BT_ADVERT_NAME = "CONNECTCORE-%s"

BT_TAG_DATA = "data"
BT_TAG_DESC = "desc"
BT_TAG_ELEMENT = "element"
BT_TAG_IFACE = "iface"
BT_TAG_INDEX = "index"
BT_TAG_NAME = "name"
BT_TAG_OP = "operation"
BT_TAG_PASSWORD = "password"
BT_TAG_SETTINGS = "settings"
BT_TAG_STATUS = "status"
BT_TAG_TOTAL = "total"
BT_TAG_VALUE = "value"

BT_OP_READ = "R"
BT_OP_WRITE = "W"

BT_PAYLOAD_LIMIT = 140

BT_REQUEST_TIMEOUT = 5000

COMMAND_PING = "ping -q -w 1 -c 1 -I %s 8.8.8.8"
COMMAND_READ_SN = "fw_printenv -n serial#"

NPU_DEMOS_FILE = "static/assets/npu_demos.json"

PROGRAM_RUN_CHECK_TIMES = 3
PROGRAM_RUN_CHECK_INTERVAL = 0.2

# Variables.
log = logging.getLogger(APP_NAME)
last_cpu_work = 0
last_cpu_total = 0
led_status = {}
fw_process = None
npu_demos = None


class RequestHandler(http.server.SimpleHTTPRequestHandler):
    """
    Custom HTTP request handler class.
    """

    def do_OPTIONS(self):
        """
        Override.
        """
        self.send_response(200, "ok")
        self.end_headers()

    def do_GET(self):
        """
        Override.
        """
        if re.search("/ping", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Send the JSON value.
            self.wfile.write("{}".encode(encoding="utf_8"))
        else:
            # Forbidden.
            self._set_headers(403)

    def do_POST(self):
        """
        Override.
        """
        if re.search("/ajax/get_device_type", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            log.debug("Get device info")

            info = {
                "device_type": get_platform_id()
            }

            # Send the JSON value.
            self.wfile.write(json.dumps(info).encode(encoding="utf_8"))
        elif re.search("/ajax/get_device_info", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            log.debug("Get device info")

            # Generate the JSON information.
            mem_info = get_mem_info()
            mca_versions = get_mca_version()
            info = {
                "uboot_version": read_proc_file("/proc/device-tree/digi,uboot,version"),
                "kernel_version": "%s %s %s %s %s GNU/Linux" % (platform.system(),
                                                                platform.node(),
                                                                platform.release(),
                                                                platform.version(),
                                                                platform.machine()),
                "dey_version": "DEY-%s-%s" % (get_dey_version(), read_file("/etc/version")),
                "serial_number": get_serial_number(),
                "device_type": get_platform_id(),
                "module_variant": read_proc_file("/proc/device-tree/digi,hwid,variant"),
                "board_variant": read_proc_file("/proc/device-tree/digi,carrierboard,version"),
                "board_id": read_proc_file("/proc/device-tree/digi,carrierboard,id"),
                "mca_hw_version": mca_versions[1],
                "mca_fw_version": mca_versions[0],
                "memory_total": mem_info.get("MemTotal", NOT_AVAILABLE) if mem_info else NOT_AVAILABLE,
                "flash_size": get_storage_size(),
                "video_resolution": get_video_resolution(),
                "fw_store_path": get_fw_store_path(),
                "bluetooth_mac": get_bluetooth_mac(),
                "wifi_mac": ZERO_MAC,
                "wifi_ip": ZERO_IP,
                "ethernet0_mac": ZERO_MAC,
                "ethernet0_ip": ZERO_IP,
                "ethernet1_mac": ZERO_MAC,
                "ethernet1_ip": ZERO_IP,
                "ethernet2_mac": ZERO_MAC,
                "ethernet2_ip": ZERO_IP,
            }
            # Fill ethernet interfaces data.
            try:
                interfaces = NetworkInterface.list_interfaces()
                for iface in interfaces:
                    if iface not in ("eth0", "eth1", "eth2"):
                        continue
                    try:
                        net_iface = NetworkInterface.get(iface)
                        index = iface[len("eth"):]
                        info["ethernet%s_mac" % index] = mac_to_human_string(net_iface.mac)
                        info["ethernet%s_ip" % index] = str(net_iface.ipv4)
                    except NetworkException as exc2:
                        log.error("Error reading interface '%s' data: %s", iface, str(exc2))
            except DigiAPIXException as exc:
                log.error("Error listing network interfaces: %s", str(exc))

            # Fill WiFi interfaces data.
            try:
                interfaces = WifiInterface.list_interfaces()
                if interfaces and "wlan0" in interfaces:
                    try:
                        net_iface = WifiInterface.get("wlan0")
                        info["wifi_mac"] = mac_to_human_string(net_iface.mac)
                        info["wifi_ip"] = str(net_iface.ipv4)
                    except WifiException as exc2:
                        log.error("Error reading interface 'wlan0' data: %s", str(exc2))
            except DigiAPIXException as exc:
                log.error("Error listing network interfaces: %s", str(exc))

            # Send the JSON value.
            self.wfile.write(json.dumps(info).encode(encoding="utf_8"))
        elif re.search("/ajax/get_device_status", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            log.debug("Get device status")

            mem_info = get_mem_info()
            mem_total = int(mem_info.get("MemTotal", "-1")) if mem_info else -1
            mem_free = int(mem_info.get("MemFree", "-1")) if mem_info else -1
            mem_used = (mem_total - mem_free) if (mem_total > 0 and mem_free > 0) else -1

            status = {
                "system_monitor/cpu_load": get_cpu_load(),
                "system_monitor/cpu_temperature": get_cpu_temp(),
                "system_monitor/frequency": read_file("/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_cur_freq").strip(),
                "system_monitor/uptime": get_uptime(),
                "system_monitor/free_memory": mem_free,
                "system_monitor/used_memory": mem_used,
                "system_monitor/led_status": get_led_status("USER_LED")
            }

            # Fill bluetooth devices data.
            try:
                bt_devices = BluetoothDevice.list_devices() or []
                for device in bt_devices:
                    try:
                        bt_device = BluetoothDevice.get(device)
                        statistics = bt_device.get_statistics()
                        status[f"system_monitor/{device}/state"] = 1 if \
                            bt_device.is_enabled else 0
                        status[f"system_monitor/{device}/rx_bytes"] = statistics.rx_bytes
                        status[f"system_monitor/{device}/tx_bytes"] = statistics.tx_bytes
                    except NetworkException as exc2:
                        log.error("Error reading bluetooth device '%s' data: %s", device, str(exc2))
            except DigiAPIXException as exc:
                log.error("Error listing bluetooth devices: %s", str(exc))

            # Fill ethernet interfaces data.
            try:
                interfaces = NetworkInterface.list_interfaces() or []
                for iface in interfaces:
                    try:
                        net_iface = NetworkInterface.get(iface)
                        statistics = net_iface.get_statistics()
                        status[f"system_monitor/{iface}/state"] = 1 if \
                            net_iface.status == NetStatus.CONNECTED else 0
                        status[f"system_monitor/{iface}/rx_bytes"] = statistics.rx_bytes
                        status[f"system_monitor/{iface}/tx_bytes"] = statistics.tx_bytes
                    except NetworkException as exc2:
                        log.error("Error reading interface '%s' data: %s", iface, str(exc2))
            except DigiAPIXException as exc:
                log.error("Error listing network interfaces: %s", str(exc))

            # Fill WiFi interfaces data.
            try:
                interfaces = WifiInterface.list_interfaces() or []
                for iface in interfaces:
                    try:
                        wifi_iface = WifiInterface.get(iface)
                        statistics = wifi_iface.get_statistics()
                        status[f"system_monitor/{iface}/state"] = 1 if \
                            wifi_iface.status == NetStatus.CONNECTED else 0
                        status[f"system_monitor/{iface}/rx_bytes"] = statistics.rx_bytes
                        status[f"system_monitor/{iface}/tx_bytes"] = statistics.tx_bytes
                    except WifiException as exc2:
                        log.error("Error reading interface '%s' data: %s", iface, str(exc2))
            except DigiAPIXException as exc:
                log.error("Error listing WiFi interfaces: %s", str(exc))

            # Send the JSON value.
            self.wfile.write(json.dumps(status).encode(encoding="utf_8"))
        elif re.search("/ajax/set_led_value", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            led = json.loads(data.decode("utf-8")).get("led_name", None)
            value = json.loads(data.decode("utf-8")).get("value", None)

            log.debug("Set LED %s to %s", led, value)

            if not led or value is None:
                error = "Invalid LED name" if not led else "Invalid LED value"
                log.error("Error setting LED '%s': %s", led, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            error = set_led_status(led, value)

            # Send the JSON value.
            if error:
                log.error("Error setting LED '%s': %s", led, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            self.wfile.write("{}".encode(encoding="utf_8"))
        elif re.search("/ajax/play_music", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            play = json.loads(data.decode("utf-8")).get("play", None)
            music_file = json.loads(data.decode("utf-8")).get("music_file", None)

            log.debug("Play music: %s", play)
            if music_file:
                log.debug("Music file: %s", music_file)

            play_music(play, music_file)

            # Send the JSON value.
            self.wfile.write(json.dumps({"play": play}).encode(encoding="utf_8"))
        elif re.search("/ajax/set_audio_volume", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            value = json.loads(data.decode("utf-8")).get("value", None)

            log.debug("Set audio volume to %s", value)

            if value is None or value > 100 or value < 0:
                error = "Invalid volume value"
                log.error("Error setting audio volume to '%s%%': %s", value, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            error = set_audio_volume(value)

            # Send the JSON value.
            if error:
                log.error("Error setting audio volume to '%s%%': %s", value, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            # Send the JSON value.
            self.wfile.write("{}".encode(encoding="utf_8"))
        elif re.search("/ajax/fs_list_directory", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            path = json.loads(data.decode("utf-8")).get("directory", None)
            filters = json.loads(data.decode("utf-8")).get("filters", None)

            log.debug("List directory: %s (filters %s)", path, filters)

            if not path or not os.path.exists(path):
                error = "Invalid path" if not path else "No such file or directory"
                log.error("Error listing directory '%s': %s", path, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            work_path = os.path.realpath(path)
            tmp_list = []

            for item in os.listdir(work_path):
                item_path = os.path.join(path, item)
                item_work_path = os.path.realpath(item_path)
                st = os.stat(item_work_path)
                if not os.path.isdir(item_work_path):
                    if not filter_by_extension(item_path, filters):
                        continue
                item = {
                    "type": "dir" if os.path.isdir(item_work_path) else "file",
                    "name": os.path.join(path, item),
                    "last_modified": st[stat.ST_MTIME],
                }
                if os.path.isfile(item_work_path):
                    item["size"] = st[stat.ST_SIZE]
                tmp_list.append(item)

            f_list = sorted(tmp_list, key=lambda item: (item["type"], os.path.basename(item["name"])))

            # Add item '..'
            if os.path.dirname(work_path) != work_path:
                f_list.insert(0, {
                    "type": "dir",
                    "name": os.path.join(path, ".."),
                    "last_modified": os.stat(os.path.join(path, "..")),
                })

            # Send the JSON value.
            self.wfile.write(json.dumps({"current_dir": path, "files": f_list}).encode(encoding="utf_8"))
        elif re.search("/ajax/fs_remove_file", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            path = json.loads(data.decode("utf-8")).get("path", None)

            log.debug("Remove file: %s", path)

            if not path or not os.path.exists(path):
                error = "Invalid path" if not path else "No such file or directory"
                log.error("Error removing file '%s': %s", path, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            try:
                os.remove(path)
                self.wfile.write("{}".encode(encoding="utf_8"))
            except OSError as e:
                self.wfile.write(json.dumps({"error": e.strerror}).encode(encoding="utf_8"))
                log.error("Error removing file '%s': %s", e.filename, e.strerror)
                return

        elif re.search("/ajax/fs_create_dir", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            path = json.loads(data.decode("utf-8")).get("directory", None)

            log.debug("Create directory: %s", path)

            if not path:
                log.error("Error creating directory '%s': Invalid directory", path)
                self.wfile.write(json.dumps({"error": "Invalid directory"}).encode(encoding="utf_8"))
                return

            try:
                os.makedirs(path, exist_ok=True)
            except OSError as e:
                self.wfile.write(json.dumps({"error": e.strerror}).encode(encoding="utf_8"))
                log.error("Error creating directory '%s': %s", path, e.strerror)
                return

            self.wfile.write("{}".encode(encoding="utf_8"))
        elif re.search("/ajax/fs_download_file", self.path) is not None:
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            path = json.loads(data.decode("utf-8")).get("path", None)

            log.debug("Download file: %s", path)

            if not path:
                self._set_headers(200)
                log.error("Error downloading file '%s': Invalid path", path)
                self.wfile.write(json.dumps({"error": "Invalid path"}).encode(encoding="utf_8"))
                return

            work_path = os.path.realpath(path)
            try:
                with open(work_path, "rb") as f:
                    contents = f.read()
            except OSError as e:
                self._set_headers(200)
                log.error("Error downloading file '%s': %s", path, e.strerror)
                self.wfile.write(json.dumps({"error": e.strerror}).encode(encoding="utf_8"))
                return

            self.send_response(200)
            self.send_header("Content-type", "application/octet-stream")
            self.end_headers()
            self.wfile.write(contents)
        elif re.search("/ajax/fs_upload_file", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the data.
            ctype, pdict = cgi.parse_header(self.headers['content-type'])
            if "boundary" in pdict:
                pdict["boundary"] = pdict["boundary"].encode()
            data = cgi.parse_multipart(self.rfile, pdict)
            path = data.get("path", [None])[0]
            f_data = data.get("file", [bytes()])[0]
            overwrite = data.get("overwrite", [False])[0]
            error = None

            log.debug("Upload file: %s (overwrite %s)", path, overwrite)

            if not path:
                error = "Invalid path"
            elif os.path.exists(path):
                if not overwrite:
                    error = "File already exists"
                else:
                    try:
                        os.remove(path)
                    except OSError as e:
                        error = e.strerror

            if error:
                log.error("Error uploading file '%s': %s", path, error)
                self.wfile.write(json.dumps({"error": error}).encode(encoding="utf_8"))
                return

            work_path = os.path.realpath(path)
            try:
                with open(work_path, "xb") as f:
                    f.write(f_data)
                self.wfile.write("{}".encode(encoding="utf_8"))
            except OSError as e:
                log.error("Error uploading file '%s': %s", path, e.strerror)
                self.wfile.write(json.dumps({"error": e.strerror}).encode(encoding="utf_8"))
        elif re.search("/ajax/reboot_device", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            log.debug("Reboot device")

            # Send the JSON value.
            self.wfile.write("{}".encode(encoding="utf_8"))

            exec_cmd("reboot")
        elif re.search("/ajax/update_firmware", self.path) is not None:
            global fw_process

            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            path = json.loads(data.decode("utf-8")).get("file", None)

            log.debug("Update firmware with file %s", path)
            if not is_dual_system() and not path.startswith(get_fw_store_path())\
                    and shutil.disk_usage(get_fw_store_path())[2] >= os.path.getsize(path):
                # Move the package to /mnt/update to avoid permission problems
                update_path = os.path.join(get_fw_store_path(), os.path.basename(path))
                if os.path.exists(update_path):
                    os.remove(update_path)
                shutil.move(path, update_path)
                path = update_path
            cmd = "update-firmware %s" % path
            log.debug("Update cmd: %s", cmd)

            try:
                fw_process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                              stderr=subprocess.STDOUT,
                                              shell=True, text=True)
                self.wfile.write("{}".encode(encoding="utf_8"))
            except OSError as e:
                log.error("Error updating firmware '%s': %s", path, e.strerror)
                self.wfile.write(json.dumps({"error": e.strerror}).encode(encoding="utf_8"))
                fw_process = None
            except subprocess.SubprocessError as e:
                log.error("Error updating firmware '%s': %s", path, e.stdout)
                self.wfile.write(json.dumps({"error": e.stdout}).encode(encoding="utf_8"))
                fw_process = None

        elif re.search("/ajax/check_firmware_update_running", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            is_running = (fw_process.poll() is not None) if fw_process else False

            log.debug("Update firmware is running %s", is_running)

            self.wfile.write(json.dumps({"update-running": is_running}).encode(encoding="utf_8"))
        elif re.search("/ajax/check_firmware_update_status", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)

            status = "successful"
            if fw_process:
                ret = fw_process.poll()
                if ret is None:
                    status = "active"
                elif ret != 0:
                    status = "failed"

            log.debug("Update firmware status %s", status)

            self.wfile.write(json.dumps({"status": status}).encode(encoding="utf_8"))
        elif re.search("/ajax/check_firmware_update_progress", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            progress = 100
            if fw_process and fw_process.poll() is None:
                progress = "?"
            self.wfile.write(json.dumps({"progress": progress, "message": "Updating firmware"}).encode(encoding="utf_8"))
        elif re.search("/ajax/get_config", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            elements = json.loads(data.decode("utf-8")).get("elements", None)
            # Fill configuration data.
            config_data = get_elements_configuration(elements)
            # Send the answer.
            if "error" in config_data:
                self.wfile.write(json.dumps(config_data).encode(encoding="utf_8"))
            else:
                self.wfile.write(json.dumps(
                    {"data": json.dumps(config_data)}).encode(encoding="utf_8"))
        elif re.search("/ajax/set_config", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            configuration = json.loads(data.decode("utf-8")).get("configuration", None)
            # Apply configuration.
            result = set_configuration(configuration)
            # Send the answer.
            if "error" in result:
                self.wfile.write(json.dumps(result).encode(encoding="utf_8"))
            else:
                self.wfile.write(json.dumps({"data": json.dumps(result)}).encode(encoding="utf_8"))
        elif re.search("/ajax/get_npu_info", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Fill NPU info.
            info = {
                "has-npu-demos": str(has_npu_demos()).lower()
            }
            # Send the JSON value.
            self.wfile.write(json.dumps(info).encode(encoding="utf_8"))
        elif re.search("/ajax/get_npu_demos", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get NPU demos data.
            platform_id = get_platform_id()
            demos = get_npu_demos_for_platform(platform_id)
            self.wfile.write(json.dumps({"data": json.dumps(demos)}).encode(encoding="utf_8"))
        elif re.search("/ajax/run_npu_demo", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            demo_id = json.loads(data.decode("utf-8")).get("demo_id", None)
            # Get the demo with the given ID.
            demo = get_npu_demo(demo_id)
            if demo:
                # Get the command to execute.
                script = None
                platform_id = get_platform_id()
                for comp_platform in demo.get("compatible_platforms", []):
                    if comp_platform.get("platform", None) == platform_id:
                        script = comp_platform.get("launch_script", None)
                        break
                if script:
                    # Execute the demo.
                    exec_cmd_nowait(script)
                    self.wfile.write(json.dumps({"id": demo_id}).encode(encoding="utf_8"))
                else:
                    self.wfile.write(json.dumps({"error": "NPU demo launch script not found."}).encode(encoding="utf_8"))
            else:
                self.wfile.write(json.dumps({"error": "NPU demo not found."}).encode(encoding="utf_8"))
        elif re.search("/ajax/is_npu_demo_running", self.path) is not None:
            # Set the response headers.
            self._set_headers(200)
            # Get the JSON data.
            data = self.rfile.read(int(self.headers["Content-Length"]))
            # Get the demo ID.
            demo_id = json.loads(data.decode("utf-8")).get("demo_id", None)
            # Return whether demo is running or not.
            self.wfile.write(json.dumps({"id": demo_id, "is_running": is_npu_demo_running(demo_id)}).encode(encoding="utf_8"))
        else:
            # Forbidden.
            self._set_headers(403)

    def end_headers(self):
        """
        Override.
        """
        self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "X-Requested-With")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Access-Control-Allow-Origin", "*")
        http.server.SimpleHTTPRequestHandler.end_headers(self)

    def _set_headers(self, status):
        """
        Sets the status and heders of the response.

        Args:
            status (Integer): The status code of the response.
        """
        self.send_response(status)
        self.send_header("Content-type", "application/json")
        self.end_headers()


class BluetoothService:
    def __init__(self):
        self._last_request_chunk_time = 0
        self._request_chunks = []
        self._ble_service = None
        self._ble_thread = None
        self._lock = Lock()

    def start(self):
        """
        Start Bluetooth service.
        """
        with self._lock:
            try:
                self._ble_service = ConnectCoreBLEService.get_instance(ble_interface=BLEInterface.ADAPTER)
                self._ble_service.add_connect_callback(self._connection_cb)
                self._ble_service.add_data_received_callback(self._data_received_cb)
                self._ble_service.set_password(get_bt_password())
                self._ble_service.set_advertising_name(get_bt_advertising_name())
                self._ble_thread = Thread(target=self._ble_service.start, daemon=True)
                self._ble_thread.start()
                return True
            except BluetoothNotSupportedException:
                log.warning("The system does not support Bluetooth")
                return False

    def stop(self):
        """
        Stop Bluetooth service.
        """
        with self._lock:
            if self._ble_service:
                self._ble_service.remove_data_received_callback(self._data_received_cb)
                self._ble_service.remove_connect_callback(self._connection_cb)
                self._ble_service.stop()
            if self._ble_thread:
                self._ble_thread.join()
                self._ble_thread = None

    def restart(self):
        """
        Restarts the Bluetooth service.
        """
        time.sleep(0.5)
        self.stop()
        time.sleep(1)
        self.start()

    def _connection_cb(self):
        """
        Callback to be notified when a new Bluetooth connection is established.
        """
        log.info("New connection (type: '%s')", self._ble_service.get_interface_type().title)

    def _data_received_cb(self, data):
        """
        Callback to be notified when new Bluetooth data is received.
            - For a read request, send back the configuration data.
            - For a write request, update the configuration.

        Args:
            data (Bytearray): received (JSON) data.
        """
        # Check if request chunks list must be reset.
        if time.time() * 1000 > self._last_request_chunk_time + BT_REQUEST_TIMEOUT:
            self._request_chunks = []
        self._last_request_chunk_time = time.time() * 1000
        # Process request chunk.
        try:
            request_chunk = json.loads(data.decode("utf-8"))
        except JSONDecodeError as exc:
            self._send_error(None, f"Invalid JSON data from Bluetooth: {str(exc)}")
            return
        if request_chunk[BT_TAG_INDEX] is None or request_chunk[BT_TAG_TOTAL] is None \
                or request_chunk[BT_TAG_DATA] is None:
            self._send_error(None, "Invalid JSON data struct from Bluetooth")
            return
        # Extract request chunk data.
        self._request_chunks.insert(request_chunk[BT_TAG_INDEX] - 1, request_chunk[BT_TAG_DATA])
        # Check if request is complete.
        if len(self._request_chunks) == request_chunk[BT_TAG_TOTAL]:
            # Build request.
            encoded_data = "".join(chunk for chunk in self._request_chunks)
            self._request_chunks = []
            request = self._get_request(base64.b64decode(encoded_data))
            if not request:
                return
            self._process_request(request)

    def _get_request(self, data):
        """
        Checks if a request follows the format:

        {
            "element"  : "<info, wifi, ethernet>", required: optional, default=info
            "iface"    : "<iface_name>",           required: element=wifi or element=ethernet
            "operation": "R" / "W",                required: optional, default=R
            "settings"                             required: always
            [
                {
                    "name"  : "<setting_name>",    required: always
                    "value" : "<setting_value>"    required: Operation=W
                },
                . . .
            ]
        }

        Args:
            data (Bytearray): Received (JSON) data.

        Returns:
            Dictionary: A dictionary, `None` if request is invalid.
        """
        try:
            req = json.loads(data.decode("utf-8"))
        except JSONDecodeError as exc:
            self._send_error(None, f"Invalid JSON data from Bluetooth: {str(exc)}")
            return None

        if (req.get(BT_TAG_ELEMENT, ELEMENT_INFO) == ELEMENT_INFO
                and req.get(BT_TAG_OP, BT_OP_READ) != BT_OP_READ):
            self._send_error(req,
                             f"Invalid request format: '{req.get(BT_TAG_OP, BT_OP_READ)}' "
                             f"not allowed for '{ELEMENT_INFO}'")
            return None

        invalid_tag = None
        if (req.get(BT_TAG_ELEMENT, ELEMENT_INFO) in (ELEMENT_ETHERNET, ELEMENT_WIFI)
                and not req.get(BT_TAG_IFACE, None)):
            invalid_tag = BT_TAG_IFACE
        elif (req.get(BT_TAG_ELEMENT, ELEMENT_INFO) not in (ELEMENT_TEST_CONNECTIVITY,
                                                            ELEMENT_RESTART_BLUETOOTH,
                                                            ELEMENT_PING)
              and BT_TAG_SETTINGS not in req):
            invalid_tag = BT_TAG_SETTINGS
        elif (req.get(BT_TAG_OP, BT_OP_READ) == BT_OP_WRITE
              and not all(BT_TAG_VALUE in setting for setting in req.get(BT_TAG_SETTINGS, None))):
            invalid_tag = BT_TAG_VALUE

        if invalid_tag:
            self._send_error(req, f"Invalid request format: '{invalid_tag}' not specified")
            req = None

        return req

    def _process_request(self, request):
        """
        Processes the request.

        Args:
            request (Dictionary): JSON string to be processed.
        """
        print(f"REQUEST: {request}")
        element = request.get(BT_TAG_ELEMENT, ELEMENT_INFO)

        if element == ELEMENT_INFO:
            self._process_info_request(request)
        elif element in (ELEMENT_ETHERNET, ELEMENT_WIFI):
            self._process_net_request(request)
        elif element == ELEMENT_BLUETOOTH:
            self._process_bluetooth_request(request)
        elif element == ELEMENT_TEST_CONNECTIVITY:
            self._process_test_connection_request(request)
        elif element == ELEMENT_PING:
            self._process_ping_request()
        elif element == ELEMENT_RESTART_BLUETOOTH:
            self._process_restart_bluetooth()
        else:
            self._send_error(request, f"Element '{element}' not found")
            return

    def _process_info_request(self, request):
        """
        Processes an 'info' request.

        Args:
            request (Dictionary): The info request.
        """
        settings = request.get(BT_TAG_SETTINGS, None)
        for setting in settings:
            setting_name = setting[BT_TAG_NAME]
            if setting_name == "uboot_version":
                setting_value = read_proc_file("/proc/device-tree/digi,uboot,version")
            elif setting_name == "kernel_version":
                setting_value = f"{platform.system()} " \
                                f"{platform.node()} " \
                                f"{platform.release()} " \
                                f"{platform.version()} " \
                                f"{platform.machine()} GNU/Linux"
            elif setting_name == "dey_version":
                setting_value = f"DEY-{get_dey_version()}-{read_file('/etc/version')}"
            elif setting_name == "device_type":
                setting_value = get_platform_id()
            elif setting_name == "ethernet0_mac":
                setting_value = get_ethernet_mac()
            elif setting_name == "n_eth":
                setting_value = 0
                net_ifaces = NetworkInterface.list_interfaces()
                for net_iface in net_ifaces:
                    setting_value = setting_value + 1 \
                        if net_iface.startswith("eth")else setting_value
            elif setting_name == "variant":
                setting_value = read_proc_file("/proc/device-tree/digi,hwid,variant")
            elif setting_name == "serial":
                setting_value = get_serial_number()
            else:
                self._send_error(request, f"Invalid '{ELEMENT_INFO}' setting: '{setting_name}'")
                return
            setting[BT_TAG_VALUE] = setting_value

        resp = {BT_TAG_STATUS: 0,
                BT_TAG_ELEMENT: ELEMENT_INFO,
                BT_TAG_SETTINGS: settings}

        self._send_data(json.dumps(resp).encode('utf-8'))

    def _process_net_request(self, request):
        """
        Processes a 'ethernet' or 'wifi' request.

        Args:
            request (Dictionary): The request.
        """
        element = request.get(BT_TAG_ELEMENT, ELEMENT_ETHERNET)
        iface_name = request.get(BT_TAG_IFACE, None)
        settings = request.get(BT_TAG_SETTINGS, None)
        operation = request.get(BT_TAG_OP, BT_OP_READ)

        if operation == BT_OP_READ:
            try:
                res = get_network_iface_configuration(iface_name)
            except DigiAPIXException as exc:
                self._send_error(request, f"Error getting interface configuration: {str(exc)}")
                return

            for setting in settings:
                setting[BT_TAG_VALUE] = res.get(setting[BT_TAG_NAME], None)
                if setting[BT_TAG_VALUE] is None:
                    self._send_error(request, f"Invalid '{iface_name}' setting: "
                                              f"'{setting[BT_TAG_NAME]}'")
                    return

            resp = {BT_TAG_STATUS: 0,
                    BT_TAG_ELEMENT: element,
                    BT_TAG_IFACE: iface_name,
                    BT_TAG_SETTINGS: settings}
        else:
            cfg = {}
            for setting in settings:
                cfg[setting[BT_TAG_NAME]] = setting[BT_TAG_VALUE]
            res = set_network_iface_configuration(iface_name, cfg)
            if res.get("status", None) == 1:
                self._send_error(request, f"Error setting '{iface_name}' "
                                          f"configuration: {res.get('desc', '')}")
                return
            resp = {BT_TAG_STATUS: 0,
                    BT_TAG_ELEMENT: element,
                    BT_TAG_IFACE: iface_name,
                    BT_TAG_SETTINGS: settings}

        self._send_data(json.dumps(resp).encode('utf-8'))

    def _process_bluetooth_request(self, request):
        """
        Processes a 'bluetooth' request.

        Args:
            request (Dictionary): The request.
        """
        settings = request.get(BT_TAG_SETTINGS, None)
        operation = request.get(BT_TAG_OP, BT_OP_READ)

        if operation == BT_OP_READ:
            try:
                res = get_bluetooth_device_configuration(0)
            except DigiAPIXException as exc:
                self._send_error(request, f"Error getting bluetooth configuration: {str(exc)}")
                return

            for setting in settings:
                setting[BT_TAG_VALUE] = res.get(setting[BT_TAG_NAME], None)
                if setting[BT_TAG_VALUE] is None:
                    self._send_error(request, f"Invalid 'bluetooth' setting: "
                                              f"'{setting[BT_TAG_NAME]}'")
                    return

            resp = {BT_TAG_STATUS: 0,
                    BT_TAG_ELEMENT: ELEMENT_BLUETOOTH,
                    BT_TAG_SETTINGS: settings}
        else:
            # For the moment, attend only 'password' setting.
            for setting in settings:
                if setting[BT_TAG_NAME] == BT_TAG_PASSWORD:
                    if not save_bt_password(setting[BT_TAG_VALUE]):
                        self._send_error(request, "Error saving Bluetooth password")
                        return

            resp = {BT_TAG_STATUS: 0,
                    BT_TAG_ELEMENT: ELEMENT_BLUETOOTH,
                    BT_TAG_SETTINGS: settings}

        self._send_data(json.dumps(resp).encode('utf-8'))

    def _process_test_connection_request(self, request):
        """
        Processes a 'test-connectivity' request.

        Args:
            request (Dictionary): The request.
        """
        interface = request.get(BT_TAG_IFACE, None)

        if not test_interface_connection(interface):
            self._send_error(request, f"Connectivity error in '{interface}'")
            return

        resp = {BT_TAG_STATUS: 0,
                BT_TAG_ELEMENT: ELEMENT_TEST_CONNECTIVITY}

        self._send_data(json.dumps(resp).encode('utf-8'))

    def _process_restart_bluetooth(self):
        """
        Processes a 'restart-bluetooth' request.
        """
        # No answer can be sent as connection will be closed.
        restart_thread = Thread(target=self.restart)
        restart_thread.start()

    def _process_ping_request(self):
        """
        Processes a 'ping' request.
        """
        resp = {BT_TAG_STATUS: 0, BT_TAG_ELEMENT: ELEMENT_PING}

        self._send_data(json.dumps(resp).encode('utf-8'))

    def _send_data(self, data):
        """
        Sends data to the Bluetooth connected device.

        Args:
            data (Bytearray): Data to be sent.
        """
        print(f"REQUEST ANSWER: {data}")
        try:
            encoded_data = base64.b64encode(data).decode('utf-8')
            total_chunks = (int(len(encoded_data) / BT_PAYLOAD_LIMIT)) + \
                           (1 if len(encoded_data) % BT_PAYLOAD_LIMIT != 0 else 0)
            chunk_index = 0
            while chunk_index < total_chunks:
                chunk_data = {
                    BT_TAG_INDEX: chunk_index + 1,
                    BT_TAG_TOTAL: total_chunks,
                    BT_TAG_DATA: encoded_data[chunk_index * BT_PAYLOAD_LIMIT: chunk_index *
                                              BT_PAYLOAD_LIMIT + BT_PAYLOAD_LIMIT]
                }
                self._ble_service.send_data(json.dumps(chunk_data).encode('utf-8'))
                chunk_index += 1
        except ConnectCoreBLEException as exc:
            log.error("Unable send data to connected device: %s", str(exc))

    def _send_error(self, request, msg):
        """
        Sends an error message to the client.

        Args:
            request (Dictionary): Original request.
            msg (String): Error message to send.
        """
        log.error(msg)
        resp = {BT_TAG_STATUS: 1}
        if request:
            element = request.get(BT_TAG_ELEMENT, ELEMENT_INFO)
            iface = request.get(BT_TAG_IFACE, None)
            resp.update({BT_TAG_ELEMENT: element})
            if iface is not None:
                resp.update({BT_TAG_IFACE: iface})

        resp.update({BT_TAG_DESC: msg})
        self._send_data(json.dumps(resp).encode('utf-8'))


def filter_by_extension(name, filters):
    """
    Returns whether the provided name ends with one of the provided filters.

    Args:
        name (String): Name of the file to check.
        filters (String): Comma-separated extensions.

    Returns:
        Boolean: True if the file ends with one of the filters, False otherwise.
    """
    if not filters:
        return True

    for ext in filters.split(","):
        if name.endswith(ext):
            return True

    return False


def get_uptime():
    """
    Gets the system uptime in seconds.

    Returns:
        Float: The system uptime, -1 if fails.
    """
    uptime = read_proc_file("/proc/uptime").strip()
    if uptime == NOT_AVAILABLE:
        return -1

    uptime = uptime.split()
    try:
        val = float(uptime[0]) if len(uptime) > 0 else -1
        return int(val)
    except ValueError:
        return -1


def get_cpu_temp():
    """
    Gets the CPU temperature in Celsius.

    Returns:
        Float: The CPU temperature, -1 if fails.
    """
    temp = read_file("/sys/class/thermal/thermal_zone0/temp").strip()
    if temp == NOT_AVAILABLE:
        return -1
    try:
        return int(int(temp) / 1000)
    except ValueError:
        return -1


def get_cpu_load():
    """
    Gets the CPU load of the system in percentage.

    Returns:
        Float: The CPU load, -1 if fails.
    """
    global last_cpu_work
    global last_cpu_total

    stat_info = read_proc_file("/proc/stat")
    if stat_info == NOT_AVAILABLE:
        return -1

    lines = stat_info.splitlines()
    if not lines:
        return -1

    fields = lines[0].split()
    if not fields or len(fields) < 5:
        return -1

    work = 0
    for i in range(1, 4):
        work += int(fields[i])
    total = 0
    for i in range(1, len(fields) - 1):  # Remove first
        total += int(fields[i])

    if not last_cpu_work and not last_cpu_total:
        usage = 0
    else:
        usage = (work - last_cpu_work) * 100.0 / (total - last_cpu_total)

    last_cpu_total = total
    last_cpu_work = work

    return usage


def get_storage_size():
    """
    Gets the internal storage size in kB.

    Returns:
        Integer: The internal storage size, -1 if fails.
    """
    if os.path.exists(EMMC_SIZE_FILE):
        return get_emmc_size()
    if os.path.exists(NAND_SIZE_FILE):
        return get_nand_size()
    return -1


def get_emmc_size():
    """
    Gets the internal eMMC storage size in KB.

    Returns:
        Integer: The internal eMMC storage size, -1 if fails.
    """
    size = read_file(EMMC_SIZE_FILE)
    if size == NOT_AVAILABLE:
        return -1
    try:
        return int(resize_to(int(size) * 512, SIZE_KB))
    except ValueError:
        return -1


def get_nand_size():
    """
    Gets the internal NAND storage size in KB.

    Returns:
        Integer: The internal NAND storage size, -1 if fails.
    """
    total_size = 0
    mtd_contents = read_file(NAND_SIZE_FILE)
    if mtd_contents == NOT_AVAILABLE:
        return -1
    for line in mtd_contents.splitlines():
        if line.startswith("mtd"):
            fields = line.split()
            if len(fields) < 4:
                continue
            try:
                total_size += int(fields[1], 16)
            except ValueError:
                return -1

    return total_size / 1024  # kB


def get_video_resolution():
    """
    Gets the video resolution.

    Returns:
        String: Video resolution.
    """
    res = read_file("/sys/class/drm/card0/card0-HDMI-A-1/modes")
    if res == NOT_AVAILABLE:
        res = read_file("/sys/class/drm/card0/card0-DPI-1/modes")
    if res == NOT_AVAILABLE:
        res = read_file("/sys/class/graphics/fb0/modes")
    if res == NOT_AVAILABLE or not res:
        return "No video device found"

    line = res.splitlines()[0]
    if ":" in line:
        line = line.split(":")[1]
    return line.strip()


def is_dual_system():
    """
    Returns wherther is a dual system.

    Returns:
        Boolean: True for dual systems, False otherwise.
    """
    res = exec_cmd("fw_printenv -n dualboot")
    return res[0] == 0 and res[1].strip() == "yes"


def get_fw_store_path():
    """
    Returns the path to store an image before an update.

    Returns:
        String: Absolute path to store a firmware image.
    """
    if is_dual_system():
        return "/root/"

    return "/mnt/update/"


def get_mem_info():
    """
    Gets a dictionary with memory info.

    Returns:
        Dictionary: A dictionary with memory information, None if fails.
    """
    mem_info = read_file("/proc/meminfo")
    if mem_info == NOT_AVAILABLE:
        return None

    return dict((line.split()[0].rstrip(':'), int(line.split()[1])) for line in mem_info.splitlines())


def get_dey_version():
    """
    Gets the DEY version.

    Returns:
        String: DEY version, "N/A" if it fails.
    """
    build_info = read_file("/etc/buildinfo")
    if build_info == NOT_AVAILABLE:
        return NOT_AVAILABLE
    for line in build_info.splitlines():
        m = re.search("DISTRO_VERSION = (.*)", line)
        if m:
            return m.group(1)

    return NOT_AVAILABLE


def get_mca_version():
    """
    Gets MCA firmware and hardware versions.

    Returns:
        Tuple(String, String): Tuple with firmware and hardware versions.
    """
    fw_version = None
    hw_version = None
    for dir_path, _, file_names in os.walk("/sys/devices"):
        for name in file_names:
            if name == "fw_version":
                fw_version = read_file(os.path.join(dir_path, name))
            elif name == "hw_version":
                hw_version = read_file(os.path.join(dir_path, name))
            if fw_version and hw_version:
                return fw_version, hw_version

    return NOT_AVAILABLE, NOT_AVAILABLE


def get_serial_number():
    """
    Gets the device serial number.

    Returns:
        String: The device serial number.
    """
    code, serial = exec_cmd(COMMAND_READ_SN)
    if code == 0:
        return serial.strip()

    return read_proc_file("/proc/device-tree/digi,hwid,sn")


def get_bluetooth_mac():
    """
    Gets the device Bluetooth MAC address.

    Returns:
        String: The Bluetooth MAC address.
    """
    bt_mac = ZERO_MAC
    try:
        bt_devices = BluetoothDevice.list_devices()
        if bt_devices and "hci0" in bt_devices:
            try:
                bt_device = BluetoothDevice.get("hci0")
                bt_mac = mac_to_human_string(bt_device.mac)
            except BluetoothException as exc2:
                log.error("Error reading interface 'hci0' data: %s", str(exc2))
    except DigiAPIXException as exc:
        log.error("Error listing bluetooth devices: %s", str(exc))

    return bt_mac


def get_ethernet_mac():
    """
    Gets the device Bluetooth MAC address.

    Returns:
        String: The Bluetooth MAC address.
    """
    eth_mac = ZERO_MAC
    try:
        net_iface = NetworkInterface.get("eth0")
        eth_mac = str(net_iface.mac)
    except NetworkException as exc:
        log.error("Error getting eth0 MAC: %s", str(exc))

    return eth_mac


def get_bt_advertising_name():
    """
    Gets the Bluetooth advertising name.

    Returns:
        String: The Bluetooth advertising name.
    """
    # The advertising name must be a value well known by both sides: the device and the mobile app.
    # The mobile app only has access to the scanned label values, which includes the ETH MAC and
    # uses it to calculate the device BT MAC address. For devices with only one ethernet interface,
    # the BT MAC is the ETH MAC + 2, but for devices with 2 ethernet interfaces, the BT MAC is the
    # ETH MAC + 3. Since with the scanned label it is impossible to determine for the mobile app
    # whether the device has 1 or 2 ethernet interfaces, we will be using the ETH0 MAC address as
    # the advertised name, which is well known in both sides instead of trying to determine the
    # real BT MAC.
    mac = get_ethernet_mac().replace(":", "")
    mac_tail = f"{mac[-4:]}".upper()

    return BT_ADVERT_NAME % mac_tail


def get_bt_password():
    """
    Gets the Bluetooth password.

    Returns:
        String: The Bluetooth password.
    """
    # Check first if password is saved in file.
    password = read_file(BT_PASSWORD_FILE)
    if not password or password == NOT_AVAILABLE:
        # Generate password using Serial number.
        password = get_serial_number()
        if not password or password == NOT_AVAILABLE:
            # Use default password
            password = BT_PASSWORD_DEFAULT
        else:
            # Use only the last 6 positions of the serial.
            password = password[-1 * BT_PASSWORD_SERIAL_LENGTH:]
        # Hash the plain password and return 16 last characters.
        password_hash = hashlib.sha256(password.encode())
        password = password_hash.hexdigest().lower()

    return password[-1 * BT_PASSWORD_LENGTH:]


def save_bt_password(password):
    """
    Saves the Bluetooth password.

    Params:
        password (String): The password to save.

    Returns:
        Boolean: 'True' if success, 'False' otherwise.
    """
    # Hash the plain password and save the 16 last characters.
    password_hash = hashlib.sha256(password.encode())
    password = password_hash.hexdigest().lower()

    return write_file(BT_PASSWORD_FILE, password)


def test_interface_connection(interface):
    """
    Tests the given interface connectivity.

    Returns:
        Boolean: 'True' if success, 'False' otherwise.
    """
    code, _ = exec_cmd(COMMAND_PING % interface)

    return code == 0


def get_led_status(name):
    """
    Get the LED value.

    Args:
        name (String): LED name in '/etc/libdigiapix.conf', if not the format
            must be '<chip> <led>', i.e, 'mca-gpio 12'.

    Returns:
        Boolean: True if LED is on, False otherwise.
    """
    value = led_status.get(name, None)
    if value is None:
        value = led_status.get(name.lower(), False)
    return value


def set_led_status(name, value):
    """
    Set the LED value.

    Args:
        name (String): LED name in '/etc/libdigiapix.conf', if not the format
            must be '<chip> <led>', i.e, 'mca-gpio 12'.
        value (Boolean): True to switch on the LED, False otherwise.

    Returns:
        String: An empty string if success, the error description if fails.
    """
    chip, led = get_led_by_alias(name)
    if not chip or not led:
        cmd = "gpioset %s=%s" % (name, "1" if value else "0")
    else:
        cmd = "gpioset %s %s=%s" % (chip, led, "1" if value else "0")

    res = exec_cmd(cmd)
    if res[0] != 0:
        return res[1]

    led_status[name] = value

    return ""


def get_led_by_alias(alias):
    """
    Get LED chip and line.

    Args:
        alias (String): LED name in '/etc/libdigiapix.conf'.

    Returns:
        Tuple (String, String): LED chip name and line. If not found, both are None.
    """
    apix_info = read_file("/etc/libdigiapix.conf")
    if apix_info == NOT_AVAILABLE:
        return None, None

    apix_values = {}
    for line in apix_info.splitlines():
        key, val = line.partition("=")[::2]
        apix_values[key.strip()] = val.strip()

    led_loc = apix_values.get(alias, None)
    if not led_loc:
        led_loc = apix_values.get(alias.upper(), None)

    if not led_loc:
        log.debug("LED alias '%s' not found", alias)
        return None, None
    else:
        return led_loc.split(",")


def play_music(play, music_file):
    """
    Sets the play music value.

    Args:
        play (Boolean): `True` to play music, `False` to stop it.
        music_file (String): Path of the music file to play.
    """
    exec_cmd("pkill -KILL -f mpg123")
    if play:
        exec_cmd_nowait(f"mpg123 {music_file}")


def set_audio_volume(value):
    """
    Configures the audio volume.

    Args:
        value (Integer): Volume to set in percentage.

    Returns:
        String: Error string if fails.
    """
    cmd = f"amixer set 'Speaker' {value}% && amixer set 'Headphone' {value}%"\
        if get_platform_id() not in ("ccimx6sbc", "ccimx6qpsbc")\
        else f"amixer set 'Master' {value}%"
    res = exec_cmd(cmd)
    if res[0] != 0:
        return res[1]
    return None


def get_elements_configuration(elements):
    """
    Gets the configuration for the given elements.

    Args:
        elements (List): Elements for which to get the configuration.

    Returns:
        Dictionary: The elements configuration data.
    """
    data = {}
    for element in elements:
        if element == ELEMENT_BLUETOOTH:
            data[element] = get_bluetooth_element_configuration()
        elif element in (ELEMENT_ETHERNET, ELEMENT_WIFI):
            data[element] = get_network_element_configuration(element)
        else:
            data["error"] = f"Unknown element '{element}'"

    return data


def get_network_element_configuration(element):
    """
    Gets the configuration for the given network element.

    Args:
        element (String): Network element for which to get the configuration.

    Returns:
        Dictionary: The network element configuration data.
    """
    data = {}
    prefix = PREFIX_ETHERNET
    if element == ELEMENT_WIFI:
        prefix = PREFIX_WIFI
    try:
        ifaces = NetworkInterface.list_interfaces() or []
        for iface in ifaces:
            if prefix in iface:
                data[iface] = get_network_iface_configuration(iface)
    except DigiAPIXException as exc:
        data["error"] = f"Error listing network interfaces: '{str(exc)}'"

    return data


def get_network_iface_configuration(interface):
    """
    Returns the network configuration of the given interface.

    Args:
        interface (String): Network interface to get its configuration.

    Returns:
        Dictionary: dictionary with the interface configuration.
    """
    data = {}
    try:
        net_iface = NetworkInterface.get(interface)
        data["enable"] = 1 if net_iface.status == NetStatus.CONNECTED else 0
        data["mac"] = mac_to_human_string(net_iface.mac)
        data["type"] = 0 if net_iface.ip_mode == IPMode.STATIC else 1
        data["ip"] = str(net_iface.ipv4)
        data["netmask"] = str(net_iface.netmask)
        data["dns1"] = str(net_iface.dns1)
        data["dns2"] = str(net_iface.dns2)
        data["gateway"] = str(net_iface.gateway)
        if PREFIX_WIFI in interface:
            try:
                wifi_iface = WifiInterface.get(interface)
                data["ssid"] = wifi_iface.ssid
                data["frequency"] = wifi_iface.frequency
                data["channel"] = wifi_iface.channel
                data["sec_mode"] = 0 if wifi_iface.sec_mode == SecurityMode.UNKNOWN \
                    else wifi_iface.sec_mode.code
            except WifiException as exc:
                raise NetworkException({str(exc)}) from exc
    except NetworkException as exc2:
        data["error"] = f"Error reading interface data: {str(exc2)}"

    return data


def set_configuration(config_data):
    """
    Applies the given configuration.

    Args:
        config_data (Dictionary): Configuration to apply.

    Returns:
        Dictionary: The operation result.
    """
    data = {}
    for element in config_data:
        if element == ELEMENT_BLUETOOTH:
            data[element] = set_bluetooth_element_configuration(config_data[element])
        elif element in (ELEMENT_ETHERNET, ELEMENT_WIFI):
            data[element] = set_network_element_configuration(config_data[element])
        else:
            data["error"] = f"Unknown element '{element}'"

    return data


def set_network_element_configuration(config_data):
    """
    Applies the given network configuration.

    Args:
        config_data (Dictionary): Network configuration to apply.

    Returns:
        Dictionary: The operation result.
    """
    data = {}
    for iface in config_data:
        data[iface] = set_network_iface_configuration(iface, config_data[iface])

    return data


def set_network_iface_configuration(interface, config_data):
    """
    Applies the given network configuration to the given interface.

    Args:
        interface (String): Network interface to apply configuration to.
        config_data (Dictionary): Network configuration to apply.

    Returns:
        Dictionary: Operation result.
    """
    data = {}
    profile = NetworkProfile()
    if PREFIX_WIFI in interface:
        profile = WifiProfile()
    profile.connect = config_data.get("enable", None)
    profile.mode = IPMode.get(config_data["type"]) if "type" in config_data else None
    profile.ipv4 = config_data.get("ip", None)
    profile.netmask = config_data.get("netmask", None)
    profile.gateway = config_data.get("gateway", None)
    profile.dns1 = config_data.get("dns1", None)
    profile.dns2 = config_data.get("dns2", None)
    if PREFIX_WIFI in interface:
        profile.ssid = config_data.get("ssid", None)
        profile.sec_mode = (SecurityMode.get(config_data["sec_mode"])
                            if "sec_mode" in config_data else None)
        profile.psk = config_data.get("psk", None)
    try:
        if PREFIX_ETHERNET in interface:
            net_iface = NetworkInterface.get(interface)
        elif PREFIX_WIFI in interface:
            net_iface = WifiInterface.get(interface)
        else:
            raise DigiAPIXException(f"Unknown interface '{interface}'")
        net_iface.configure(profile)
        data["status"] = 0
    except DigiAPIXException as exc2:
        data["status"] = 1
        data["desc"] = f"Error configuring interface: {str(exc2)}"

    return data


def get_bluetooth_element_configuration():
    """
    Gets the configuration for the bluetooth element.

    Returns:
        Dictionary: The bluetooth element configuration data.
    """
    data = {}
    try:
        bt_devices = BluetoothDevice.list_devices() or []
        for bt_device in bt_devices:
            data[bt_device] = get_bluetooth_device_configuration(bt_device)
    except DigiAPIXException as exc:
        data["error"] = f"Error listing bluetooth devices: '{str(exc)}'"

    return data


def get_bluetooth_device_configuration(device):
    """
    Returns the configuration of the given bluetooth device.

    Args:
        device (String): Bluetooth device to get its configuration.

    Returns:
        Dictionary: dictionary with the device configuration.
    """
    data = {}
    try:
        bt_device = BluetoothDevice.get(device)
        data["device_id"] = bt_device.device_id
        data["advert_name"] = str(bt_device.advert_name)
        data["mac"] = mac_to_human_string(bt_device.mac)
        data["enable"] = 1 if bt_device.is_enabled else 0
        data["running"] = 1 if bt_device.is_running else 0
    except BluetoothException as exc:
        data["error"] = f"Error reading device data: {str(exc)}"

    return data


def set_bluetooth_element_configuration(config_data):
    """
    Applies the given bluetooth configuration.

    Args:
        config_data (Dictionary): Bluetooth configuration to apply.

    Returns:
        Dictionary: The operation result.
    """
    data = {}
    for device in config_data:
        data[device] = set_bluetooth_device_configuration(device, config_data[device])

    return data


def set_bluetooth_device_configuration(device, config_data):
    """
    Applies the given bluetooth configuration to the given device.

    Args:
        device (String): Bluetooth device to apply configuration to.
        config_data (Dictionary): Bluetooth configuration to apply.

    Returns:
        Dictionary: Operation result.
    """
    data = {}
    try:
        bt_device = BluetoothDevice.get(device)
        profile = BluetoothProfile()
        profile.enable = config_data.get("enable", None)
        profile.advert_name = config_data.get("advert_name", None)
        bt_device.configure(profile)
        data["status"] = 0
    except DigiAPIXException as exc2:
        data["status"] = 1
        data["desc"] = f"Error configuring interface: {str(exc2)}"

    return data


def has_npu_demos():
    """
    Returns whether the device has NPU demos available or not.

    Returns:
        Boolean: True if device has NPU demos available, False otherwise.
    """
    return len(get_npu_demos_for_platform(get_platform_id())) > 0


def load_npu_demos():
    """
    Returns the loaded NPU demos or parses the JSON file with NPU demos
    metadata.

    Returns:
        list: A list of dictionaries, each representing an NPU demo,
              `None` if error
    """
    global npu_demos

    # Return the demos if they are already loaded.
    if npu_demos:
        return npu_demos

    # Initialize the demos var.
    npu_demos = None

    # Build demos file path.
    npu_demos_file_path = os.path.join(os.path.dirname(__file__), NPU_DEMOS_FILE)
    try:
        with open(npu_demos_file_path, 'r') as file:
            content = json.load(file)
            if content:
                npu_demos = content.get("npu_demos", [])
    except FileNotFoundError:
        log.error("Error reading NPU demos: File '%s' not found." % NPU_DEMOS_FILE)
    except json.JSONDecodeError:
        log.error("Error reading NPU demos: File '%s' contains invalid JSON." % NPU_DEMOS_FILE)
    except OSError as exc:
        log.error("Error reading NPU demos: %s" % str(exc))

    return npu_demos


def get_npu_demo(demo_id):
    """
    Returns the NPU demo matching the given ID.

    Args:
        demo_id (str): ID of the NPU demo to get.

    Returns:
        Dictionary: The NPU demo for the given ID, None if the demo is not
                    found.
    """
    demos = load_npu_demos()
    if not demos:
        return None

    return next((demo for demo in demos if demo.get("id", None) == demo_id), None)


def get_npu_demos_for_platform(platform_name):
    """
    Returns a list of NPU demos that are compatible with the given platform.

    Args:
        platform_name (str): The name of the platform to filter NPU demos by.

    Returns:
        list: A list of dictionaries, each representing an NPU demo compatible
              with the given platform.
    """
    compatible_demos = []
    demos = load_npu_demos()

    if demos:
        for demo in demos:
            for platform in demo.get('compatible_platforms', []):
                if platform.get('platform') == platform_name and file_exists(platform.get('launch_script')):
                    compatible_demos.append(demo)
                    break

    return compatible_demos


def is_npu_demo_running(demo_id):
    """
    Check if the given NPU demo is currently running in the system or not.
    
    Args:
        demo_id (str): ID of the demo to check.
        
    Returns:
        bool: True if the demo is running, False otherwise.
    """
    demo = get_npu_demo(demo_id)
    if demo:
        # Get the demo execution script.
        script = None
        platform_id = get_platform_id()
        for comp_platform in demo.get("compatible_platforms", []):
            if comp_platform.get("platform", None) == platform_id:
                script = comp_platform.get("launch_script", None)
                break
        if script:
            return is_program_running(script)
        
    return False


def is_program_running(program):
    """
    Check if a given program is currently running in the system or not.
    
    Args:
        program (str): The program/command to check.
        
    Returns:
        bool: True if the program is running, False otherwise.
    """
    try:
        for i in range(PROGRAM_RUN_CHECK_TIMES):
            # Run the 'ps' command and capture the output
            command = f"ps w | grep -v 'grep' | grep -q '{program}'"
            result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            
            # Check if the return code is 0 (program is running)
            if result.returncode == 0:
                return True
            
            # Keep checking to discard false negatives.
            time.sleep(PROGRAM_RUN_CHECK_INTERVAL)
    except Exception as e:
        print(f"An error occurred: {e}")
    
    return False


def get_platform_id():
    """
    Returns the running platform ID.

    Returns:
        String: The running platform ID.
    """
    return read_proc_file("/proc/device-tree/digi,machine,name")


def mac_to_human_string(mac, num_bytes=6):
    """
    Transforms the given MAC address into a human readable string.

    Args:
        mac (:class:`.MacAddress`): The MAC address to transform.
        num_bytes (Integer): The number of bytes to use in the MAC Address.

    Return:
        String: The given MAC as human readable string.
    """
    return ":".join(["%02X" % i for i in mac][8-num_bytes:]).upper()


def read_proc_file(path):
    """
    Gets contents of a proc file.
    Reads the file and eliminates the last NULL character.

    Args:
        path (String): Absolute path of the file to read.

    Returns:
        String: The U-Boot version.
    """
    data = read_file(path)
    if data == NOT_AVAILABLE:
        return data

    if data[len(data) - 1] == '\0':
        data = data[:len(data) - 1]

    return data


def exec_cmd(cmd, timeout=None):
    """
    Executes the provided command and waits for it.

    Args:
        cmd (String, List): The command to execute.
        timeout (Float, optional, default=None): Timeout to wait for termination.

    Returns:
        Tuple (Integer, String): A tuple with the return code and the result
            output or error.
    """
    try:
        result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                timeout=timeout, shell=True, check=True, text=True)
        return result.returncode, result.stdout
    except subprocess.TimeoutExpired as e:
        return -1, e.stdout
    except subprocess.CalledProcessError as e:
        return e.returncode, e.stdout


def exec_cmd_nowait(command, *args):
    """
    Executes the provided command without waiting to finish.

    Args:
        command (String): The command to execute.
        args (List): The list of arguments.
    """
    arguments = []
    for arg in args:
        arguments.extend(arg)
    subprocess.Popen([command] + arguments, shell=True, stdout=subprocess.PIPE,
                     stderr=subprocess.STDOUT, text=True)


def read_file(path):
    """
    Reads the provided file path.

    Args:
        path (String): Absolute path of the file to read.

    Returns:
        String: Contents of the file, 'N/A' if any error occurs.
    """
    try:
        with open(path, encoding="utf-8") as file:
            return file.read()
    except OSError as exc:
        if exc.errno == errno.ENOENT:
            log.error("File '%s' does not exist (errno: %d)", path, exc.errno)
        else:
            log.error("Cannot open file '%s' (errno: %d)", path, exc.errno)

    return NOT_AVAILABLE


def write_file(path, value):
    """
    Writes the provided file path with the given value.

    Args:
        path (String): Absolute path of the file to write.
        value (String): Value to write.

    Returns:
        Boolean: 'True' if success, 'False' otherwise.
    """
    try:
        with open(path, 'w', encoding="utf-8") as file:
            file.write(value)
    except OSError as exc:
        log.error("Cannot write file '%s' (errno: %d)", path, exc.errno)
        return False

    return True


def file_exists(file_path):
    """
    Determines whether the given file path exists or not.

    Args:
        file_path (String): Absolute path of the file to check.

    Returns:
        Boolean: 'True' if the file exists, 'False' otherwise.
    """
    return os.path.isfile(file_path)


def resize_to(value, to, divider=1024):
    """
    Resizes the given value.

    Args:
        value (Integer): The value to resize.
        to (Integer): The divider scale.
        divider (Integer, Optional): The base divider.

    Returns:
        Float: The resized value.
    """
    r = float(value)
    return r / (divider ** DIVIDERS[to])


def init_log(log_level, log_console):
    """
    Initialize console and syslog loggers.

    Args:
        log_level (String): The file logger level.
        log_console (Boolean): `True` to log to console, `False` otherwise.
    """
    log_format = "%(name)s: %(message)s"
    level = logging.INFO
    handlers = []

    if log_level == "D":
        level = logging.DEBUG
    elif log_level == "I":
        level = logging.INFO
    elif log_level == "W":
        level = logging.WARNING
    elif log_level == "E":
        level = logging.ERROR

    def configure_handler(log_handler, name, fmt, handlers_list):
        log_handler.name = name
        log_handler.setFormatter(
            logging.Formatter(fmt=fmt, datefmt='%Y-%m-%d,%H:%M:%S'))
        log_handler.setLevel(level)
        handlers_list.append(log_handler)

    if log_console:
        log_console_format = "%(asctime)s %(name)s: %(message)s"
        configure_handler(logging.StreamHandler(), "%s console handler",
                          log_console_format, handlers)
    configure_handler(SysLogHandler(address='/dev/log'), "%s syslog handler",
                      log_format, handlers)

    log.disabled = False
    log.setLevel(level)
    for handler in handlers:
        log.addHandler(handler)


def disable_logger():
    """
    Disables the logger and removes all handlers.
    """
    log.disabled = True

    for hdl in log.handlers:
        hdl.close()
        log.removeHandler(hdl)


def signal_handler(signal_number, _frame):
    """
    Signal handler function.

    Args:
        signal_number (Integer): Received signal.
        _frame: Current stack frame.
    """
    log.debug("Signal received %d", signal_number)


def main():
    """
    Main function to listen to XBee status changes.
    """
    parser = argparse.ArgumentParser(description='Device demo server',
                                     add_help=True,
                                     formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("-p", "--port", metavar="<port>", nargs=1, default=PORT,
                        type=int, help="Server port")
    parser.add_argument("--log", metavar="<D, I, W, E>", default='I',
                        choices=['D', 'I', 'W', 'E'],
                        help="Log level: debug, info, warning, error")
    parser.add_argument("--log-console", action='store_true',
                        dest="log_console",
                        help="Enable log to standard output")

    args = parser.parse_args()

    init_log(args.log, args.log_console)

    # Create the HTTP server and start it.
    socketserver.ThreadingTCPServer.allow_reuse_address = True
    server = socketserver.ThreadingTCPServer(("", args.port), RequestHandler)

    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    log.info("Serving at port %s" % args.port)

    server_thread = Thread(target=server.serve_forever)
    server_thread.deamon = True
    server_thread.start()

    bt_service = BluetoothService()
    bt_service.start()

    # Wait for termination/interrupt signal.
    signal.sigwait([signal.SIGTERM, signal.SIGINT])

    bt_service.stop()
    server.shutdown()
    server.server_close()
    log.info("Sever stopped")
    disable_logger()


if __name__ == "__main__":
    main()