feat: Telegram Servicer

README.md

@@ -10,11 +10,11 @@ This project is a home security system that uses a Raspberry Pi and a camera, wh
 ### Installation
 
 ```bash
-$ sudo apt install -y python3-picamera2
+$ sudo apt install -y python3-picamera2 libsystemd-dev
 $ virtualenv venv
 $ source venv/bin/activate
 $ pip install -r requirements.txt
-$ python main.py
+$ python hss.py
 ```
 
 Create a `.config.json` file in the root directory with the following content:

main.py → hss.py

@@ -19,7 +19,7 @@
 from core.utils.fileio_adaptor import upload_to_fileio
 
 
-def read_configurations() -> dict[str, Any]:
+def read_configurations() -> tuple[dict[str, Any], dict[str, Any]]:
     """
     This method reads the configurations from the .config.json file.
     """

requirements.txt

@@ -30,6 +30,7 @@ launchpadlib==1.11.0
 lazr.restfulclient==0.14.5
 lazr.uri==1.0.6
 lgpio==0.2.2.0
+lxml==5.2.1
 magic-filter==1.0.12
 mccabe==0.7.0
 more-itertools==8.10.0
@@ -44,6 +45,7 @@ piexif==1.1.3
 pigpio==1.78
 Pillow==9.4.0
 platformdirs==4.2.0
+psutil==5.9.8
 pycodestyle==2.11.1
 pydantic==2.5.3
 pydantic_core==2.14.6
@@ -55,6 +57,7 @@ PyOpenGL==3.1.6
 pyparsing==3.0.9
 PyQt5==5.15.9
 PyQt5-sip==12.11.1
+pystemd==0.13.2
 pyTelegramBotAPI==4.17.0
 python-apt==2.6.0
 python-dotenv==1.0.1

servicer.py

@@ -0,0 +1,174 @@
+"""
+This module provides an Telegram bot API to interact with the user
+without a direct access to Home Security Service.
+
+Main responsibilities of the servicer as follows:
+    - Bot provides if hardware and the bot itself is alive. (/alive)
+    - Bot provides if the service is dead or alive. (/health hss.service)
+    - Bot restarts the service if the command is sent. (/restart hss.service)
+    - Bot provides the latest N logs if wanted. (/logs hss.service:N)
+    - Bot provides if protectors are in house, and whose. (/inhouse)
+    - Bot provides an image-shot if wanted. (/imageshot)
+    - Bot schedules a reboot for the hardware. (/reboot)
+    - Bot provides a shell access to the hardware. (/shell)
+"""
+import asyncio
+import json
+from typing import Any
+
+import cv2
+from pystemd import systemd1 as systemd
+from telebot.async_telebot import AsyncTeleBot
+
+from core.strategies.eye.picamera_strategy import PiCameraStrategy
+from core.strategies.wifi.admin_panel_strategy import AdminPanelStrategy
+
+
+def read_configurations() -> tuple[dict[str, Any], dict[str, Any]]:
+    """
+    This method reads the configurations from the .config.json file.
+    """
+    with open(".config.json", "r", encoding="utf-8") as file:
+        _config = json.load(file)
+    main_settings = _config['main_settings']
+    strategy_settings = _config['strategy_settings']
+    return main_settings, strategy_settings
+
+
+# Definitations
+MAIN_CONIGS, STRATEGY_CONFIGS = read_configurations()
+SERVICER_BOT = AsyncTeleBot(token=STRATEGY_CONFIGS["telegram_strategy"]["bot_key"])
+KNOWN_LOG_LOCATIONS: dict[str, str] = {
+    "hss.service": "/home/raspberry/.home-security-system/logs/hss.log"
+}
+
+
+@SERVICER_BOT.message_handler(commands=["info", "help", "hi"])
+async def info(message):
+    """
+    This method is called when the /info, /help or /hi command is sent.
+    """
+    await SERVICER_BOT.reply_to(message,
+                                "Hi, I am the Home Security System Servicer Bot.\n\n"
+                                "Here are the commands you can use:\n"
+                                "/alive - provides if hardware and the bot itself is alive.\n"
+                                "/health hss.service - provides if the service is dead or alive.\n"
+                                "/restart hss.service - restarts the given service.\n"
+                                "/logs hss.service:N - provides the latest N logs.\n"
+                                "/inhouse - provides if protectors are in house, and whose.\n"
+                                "/imageshot - captures an image and sends.\n"
+                                "/reboot - reboots the hardware.\n"
+                                "/shell echo 'test'- provides a shell access to the hardware.\n"
+                                "/info, /help, /hi - this help text.\n")
+
+
+@SERVICER_BOT.message_handler(commands=['alive'])
+async def alive(message):
+    """
+    This method is called when the /alive command is sent.
+    """
+    await SERVICER_BOT.reply_to(message, "I am alive.")
+
+
+@SERVICER_BOT.message_handler(commands=['health'])
+async def health(message):
+    """
+    This method is called when the /health command is sent.
+    """
+    parameters = message.text[len('/health'):]
+    service_name = parameters.strip().split(' ')[0]
+    with systemd.Unit(service_name.encode("utf-8")) as service:
+        active_state: str = service.Unit.ActiveState.decode("utf-8")
+        sub_state: str = service.Unit.SubState.decode("utf-8")
+        service_name: str = service.Unit.Description.decode("utf-8")
+        main_pid: str = service.Service.MainPID
+        await SERVICER_BOT.reply_to(message,
+                                    f"Service: {service_name}\n"
+                                    f"Active State: {active_state}\n"
+                                    f"Sub State: {sub_state}\n"
+                                    f"Main PID: {main_pid}")
+
+
+@SERVICER_BOT.message_handler(commands=['restart'])
+async def restart(message):
+    """
+    This method is called when the /restart command is sent.
+    """
+    parameters = message.text[len('/restart'):]
+    service_name = parameters.strip().split(' ')[0]
+    with systemd.Unit(service_name.encode("utf-8")) as service:
+        service.Unit.Restart("fail")
+        await SERVICER_BOT.reply_to(message, f"{service_name} is restarted.")
+
+
+@SERVICER_BOT.message_handler(commands=['logs'])
+async def logs(message):
+    """
+    This method is called when the /logs command is sent.
+    """
+    first_parameter = message.text[len('/logs'):].strip().split(' ')[0]
+    service_name, last_n_lines = first_parameter.split(":")
+    if service_name not in KNOWN_LOG_LOCATIONS:
+        await SERVICER_BOT.reply_to(message, f"Unknown service: {service_name}")
+    with open(KNOWN_LOG_LOCATIONS[service_name], "r") as log_file:
+        logs = log_file.readlines()[-int(last_n_lines):]
+        await SERVICER_BOT.reply_to(message, "".join(logs))
+
+
+@SERVICER_BOT.message_handler(commands=['inhouse'])
+async def in_house(message):
+    """
+    This method is called when the /in-house command is sent.
+    """
+    protectors_list = MAIN_CONIGS["protectors"]
+    strategy = AdminPanelStrategy(STRATEGY_CONFIGS["admin_panel_strategy"])
+    connected_macs = strategy._get_all_connected()
+    connected_protectors = "\n\t- " + "\n\t- ".join([
+        protector['name'] for protector in protectors_list
+        if protector['address'] in [device.address for device in connected_macs]
+    ])
+    response = f"Connected MACs: {[device.address for device in connected_macs]}\n\n\n" \
+               f"Protectors in house: {connected_protectors}"
+    await SERVICER_BOT.reply_to(message, response)
+
+
+@SERVICER_BOT.message_handler(commands=['imageshot'])
+async def image_shot(message):
+    """
+    This method is called when the /image-shot command is sent.
+    """
+    camera = PiCameraStrategy()
+    frame = camera.get_frame()
+    success, encoded_frame = cv2.imencode('.png', frame)
+    if not success:
+        await SERVICER_BOT.reply_to(message, "Failed to capture the image.")
+        return
+    await SERVICER_BOT.send_photo(message.chat.id, encoded_frame.tobytes())
+    del frame, encoded_frame
+
+
+@SERVICER_BOT.message_handler(commands=['reboot'])
+async def reboot(message):
+    """
+    This method is called when the /reboot command is sent.
+    """
+    await SERVICER_BOT.reply_to(message, "Rebooting the hardware.")
+    with systemd.Manager() as manager:
+        manager.Reboot()
+    await SERVICER_BOT.reply_to(message, "Hardware is rebooted.")
+
+
+@SERVICER_BOT.message_handler(commands=["shell"])
+async def shell_run(message):
+    """
+    This method is called when the /shell command is sent.
+    """
+    command = message.text[len("/shell"):].strip()
+    process = await asyncio.create_subprocess_shell(command,
+                                                    stdout=asyncio.subprocess.PIPE,
+                                                    stderr=asyncio.subprocess.PIPE)
+    stdout, stderr = await process.communicate()
+    await SERVICER_BOT.reply_to(message, f"stdout: {stdout.decode()}\nstderr: {stderr.decode()}")
+
+if __name__ == "__main__":
+    asyncio.run(SERVICER_BOT.polling())

system/servicefile → system/hss.service.template

@@ -3,7 +3,7 @@ Description=Home Security System
 After=network.target
 
 [Service]
-ExecStart=/home/raspberry/home-security-system/.hss_venv/bin/python /home/raspberry/home-security-system/main.py
+ExecStart=/home/raspberry/home-security-system/.hss_venv/bin/python /home/raspberry/home-security-system/hss.py
 Restart=always
 User=raspberry
 WorkingDirectory=/home/raspberry/home-security-system/

system/telegram-servicer.service.template

@@ -0,0 +1,13 @@
+[Unit]
+Description=Telegram Servicer
+After=network.target
+
+[Service]
+ExecStart=/home/raspberry/home-security-system/.hss_venv/bin/python /home/raspberry/home-security-system/servicer.py
+Restart=always
+User=root
+WorkingDirectory=/home/raspberry/home-security-system/
+RestartSec=30
+
+[Install]
+WantedBy=default.target