Merge pull request #16 from PinoutLTD/libp2p_reports

send report using libp2p, decrease log size, don't wait for adding to RWS, fix blocking call issue

custom_components/robonomics_report_service/__init__.py

@@ -15,6 +15,7 @@
 from .robonomics import Robonomics
 from .error_sources.error_source_manager import ErrorSourcesManager
 from .report_service import ReportService
+from .libp2p import LibP2P
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -25,9 +26,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         hass,
         entry.data[CONF_SENDER_SEED],
     )
+    libp2p = LibP2P(hass, robonomics.sender_seed)
     # async_register_frontend(hass)
-    await RWSRegistrationManager.register(hass, robonomics, entry.data[CONF_EMAIL])
-    ReportService(hass, robonomics).register()
+    await RWSRegistrationManager.register(hass, robonomics, libp2p)
+    await ReportService(hass, robonomics, libp2p).register()
     error_sources_manager = ErrorSourcesManager(hass)
     error_sources_manager.setup_sources()
     hass.data[DOMAIN][ERROR_SOURCES_MANAGER] = error_sources_manager

custom_components/robonomics_report_service/const.py

@@ -1,6 +1,8 @@
 DOMAIN = "robonomics_report_service"
 ADDRESS = "address"
 
+SERVICE_PAID = False
+
 STORAGE_PINATA_CREDS = "pinata_creds"
 
 CONF_EMAIL = "email"
@@ -20,11 +22,12 @@
 LOG_FILE_NAME = "home-assistant.log"
 TRACES_FILE_NAME = ".storage/trace.saved_traces"
 IPFS_PROBLEM_REPORT_FOLDER = "ha_problem_report"
-LOGS_MAX_LEN = 9*1024*1024
+LOGS_MAX_LEN = 3*1024*1024
 
 LIBP2P_WS_SERVER = "ws://127.0.0.1:8888"
 LIBP2P_LISTEN_PROTOCOL = "/pinataCreds"
-LIBP2P_SEND_PROTOCOL = "/initialization"
+LIBP2P_SEND_INITIALISATION_PROTOCOL = "/initialization"
+LIBP2P_SEND_REPORT_PROTOCOL = "/report"
 INTEGRATOR_PEER_ID = "12D3KooWBE2XrMkf1Z6P3AtKqYmvdD59aoD5xwKySrCgkmBqJNFh"
 PROBLEM_SERVICE_ROBONOMICS_ADDRESS = "4HifM6Cny7bHAdLb5jw3hHV2KabuzRZV8gmHG1eh4PxJakwi"
 

custom_components/robonomics_report_service/libp2p.py

@@ -9,12 +9,15 @@
 from .const import (
     LIBP2P_WS_SERVER,
     LIBP2P_LISTEN_PROTOCOL,
-    LIBP2P_SEND_PROTOCOL,
+    LIBP2P_SEND_INITIALISATION_PROTOCOL,
+    LIBP2P_SEND_REPORT_PROTOCOL,
     INTEGRATOR_PEER_ID,
     STORAGE_PINATA_CREDS,
     PROBLEM_SERVICE_ROBONOMICS_ADDRESS,
     CONF_PINATA_PUBLIC,
     CONF_PINATA_SECRET,
+    CONF_EMAIL,
+    DOMAIN,
 )
 from .utils import (
     async_save_to_store,
@@ -29,21 +32,29 @@
 
 
 class LibP2P:
-    def __init__(self, hass: HomeAssistant, sender_seed: str, email: str):
+    def __init__(self, hass: HomeAssistant, sender_seed: str):
         self.hass = hass
         self.libp2p_proxy = Libp2pProxyAPI(LIBP2P_WS_SERVER)
         self.sender_seed = sender_seed
         self.sender_address = get_address_for_seed(self.sender_seed)
-        self.email = email
+        self.email = hass.data[DOMAIN][CONF_EMAIL]
         self._pinata_creds_saved = False
         self._listen_protocol = f"{LIBP2P_LISTEN_PROTOCOL}/{self.sender_address}"
+        self._initialisation = False
+
+    async def send_report(self, report_hashes: tp.Dict, address: str) -> None:
+        # await self._subscribe_to_feedback_protocol()
+        message = self._format_report_message(report_hashes, address)
+        _LOGGER.debug(f"Sending report using libp2p: {message}")
+        await self.libp2p_proxy.send_msg_to_libp2p(
+            message, LIBP2P_SEND_REPORT_PROTOCOL, server_peer_id=INTEGRATOR_PEER_ID
+        )
 
     async def get_and_save_pinata_creds(self) -> bool:
-        _LOGGER.debug(f"Start getting Pinata creds")
+        _LOGGER.debug("Start getting Pinata creds")
+        self._initialisation = True
         self._pinata_creds_saved = False
-        await self.libp2p_proxy.subscribe_to_protocol_async(
-            LIBP2P_LISTEN_ERRORS_PROTOCOL, self._handle_libp2p_feedback, reconnect=True
-        )
+        await self._subscribe_to_feedback_protocol()
         await self.libp2p_proxy.subscribe_to_protocol_async(
             self._listen_protocol, self._save_pinata_creds, reconnect=True
         )
@@ -52,8 +63,14 @@ async def get_and_save_pinata_creds(self) -> bool:
             await asyncio.sleep(1)
         _LOGGER.debug("After got pinata creds")
         await self.libp2p_proxy.unsubscribe_from_all_protocols()
+        self._initialisation = False
         return True
 
+    async def _subscribe_to_feedback_protocol(self) -> None:
+        await self.libp2p_proxy.subscribe_to_protocol_async(
+            LIBP2P_LISTEN_ERRORS_PROTOCOL, self._handle_libp2p_feedback, reconnect=True
+        )
+
     async def _save_pinata_creds(self, received_data: tp.Union[str, dict]):
         if "public" in received_data and "private" in received_data:
             storage_data = {}
@@ -74,10 +91,13 @@ async def _save_pinata_creds(self, received_data: tp.Union[str, dict]):
             _LOGGER.error(f"Libp2p message in wrong format: {received_data}")
 
     async def _handle_libp2p_feedback(self, received_data: tp.Union[str, dict]):
-        _LOGGER.debug(f"Libp2p feedback: {received_data}")
+        _LOGGER.debug(f"Libp2p feedback on {'initialisation' if self._initialisation else 'report'}: {received_data}")
         if received_data['feedback'] != "ok":
-            await asyncio.sleep(5)
-            await self._send_init_request()
+            if self._initialisation:
+                await asyncio.sleep(5)
+                await self._send_init_request()
+        if not self._initialisation:
+            await self.libp2p_proxy.unsubscribe_from_all_protocols()
 
     def _decrypt_message(self, encrypted_data: str) -> str:
         return decrypt_message(
@@ -90,7 +110,7 @@ async def _send_init_request(self) -> None:
         data = self._format_data_for_init_request()
         _LOGGER.debug(f"Sendig initialisation request: {data}")
         await self.libp2p_proxy.send_msg_to_libp2p(
-            data, LIBP2P_SEND_PROTOCOL, server_peer_id=INTEGRATOR_PEER_ID
+            data, LIBP2P_SEND_INITIALISATION_PROTOCOL, server_peer_id=INTEGRATOR_PEER_ID
         )
 
     def _format_data_for_init_request(self) -> str:
@@ -100,6 +120,9 @@ def _format_data_for_init_request(self) -> str:
             "sender_address": self.sender_address,
         }
         return json.dumps(data)
+
+    def _format_report_message(self, report_hashes: tp.Dict, address: str) -> str:
+        return json.dumps({"report": report_hashes, "address": address})
 
     def _encrypt_message(self, data: str) -> str:
         return encrypt_message(

custom_components/robonomics_report_service/manifest.json

@@ -1,11 +1,11 @@
 {
-    "domain": "robonomics_report_service",
-    "name": "Robonomics Report Service",
-    "config_flow": true,
-    "codeowners": ["@pinoutcloud"],
-    "version": "0.2.0",
-    "dependencies": ["persistent_notification", "http", "frontend"],
-    "requirements": ["robonomics-interface~=1.6.0", "pinatapy-vourhey==0.1.9", "tenacity==8.2.2", "py-ws-libp2p-proxy==0.2.0"],
-    "documentation": "https://wiki.robonomics.network/",
-    "issue_tracker": "https://github.com/PinoutLTD/rrs-ha-integration/issues"
-  }
+  "domain": "robonomics_report_service",
+  "name": "Robonomics Report Service",
+  "config_flow": true,
+  "codeowners": ["@pinoutcloud"],
+  "version": "0.3.0",
+  "dependencies": ["persistent_notification", "http", "frontend"],
+  "requirements": ["robonomics-interface~=1.6.0", "pinatapy-vourhey==0.1.9", "tenacity==8.2.2", "py-ws-libp2p-proxy==0.2.1"],
+  "documentation": "https://wiki.robonomics.network/",
+  "issue_tracker": "https://github.com/PinoutLTD/rrs-ha-integration/issues"
+}

custom_components/robonomics_report_service/report_service.py

@@ -13,36 +13,46 @@
     PROBLEM_SERVICE_ROBONOMICS_ADDRESS,
     DOMAIN,
     PROBLEM_REPORT_SERVICE,
+    SERVICE_PAID,
 )
 from .ipfs import IPFS
 from .utils import (
     create_temp_dir_with_encrypted_files,
     create_encrypted_picture,
     encrypt_message,
     delete_temp_dir,
+    get_tempdir_filenames,
 )
 from .robonomics import Robonomics
+from .libp2p import LibP2P
 
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class ReportService:
-    def __init__(self, hass: HomeAssistant, robonomics: Robonomics):
+    def __init__(self, hass: HomeAssistant, robonomics: Robonomics, libp2p: LibP2P):
         self.hass = hass
         self.robonomics = robonomics
         self.ipfs = IPFS(hass)
+        self.libp2p = libp2p
 
-    def register(self) -> None:
+    async def register(self) -> None:
         self.hass.services.async_register(DOMAIN, PROBLEM_REPORT_SERVICE, self.send_problem_report)
+        await self._clear_tempdirs()
 
     async def send_problem_report(self, call: ServiceCall) -> None:
         _LOGGER.debug(f"send problem service: {call.data.get('description')}")
-        tempdir = await self._create_temp_dir_with_report_data(call)
-        data_to_send = await self.ipfs.pin_to_pinata(tempdir)
-        self._remove_tempdir(tempdir)
+        try:
+            tempdir = await self._create_temp_dir_with_report_data(call)
+            data_to_send = await self.ipfs.pin_to_pinata(tempdir)
+        finally:
+            await self._remove_tempdir(tempdir)
         if data_to_send is not None:
-            await self.robonomics.send_datalog(json.dumps(data_to_send))
+            if SERVICE_PAID:
+                await self.robonomics.send_datalog(json.dumps(data_to_send))
+            else:
+                await self.libp2p.send_report(data_to_send, self.robonomics.sender_address)
 
     async def _create_temp_dir_with_report_data(self, call: ServiceCall) -> str:
         if call.data.get("attach_logs"):
@@ -120,7 +130,13 @@ def _encrypt_json(self, data: dict) -> str:
             recipient_address=PROBLEM_SERVICE_ROBONOMICS_ADDRESS,
         )
 
-    def _remove_tempdir(self, tempdir: str) -> None:
+    async def _clear_tempdirs(self) -> None:
+        dirs_to_delete = get_tempdir_filenames(IPFS_PROBLEM_REPORT_FOLDER)
+        for dirname in dirs_to_delete:
+            await self._remove_tempdir(dirname)
+
+    async def _remove_tempdir(self, tempdir: str) -> None:
         if os.path.exists(tempdir):
-            delete_temp_dir(tempdir)
+            await self.hass.async_add_executor_job(delete_temp_dir, tempdir)
+            _LOGGER.debug(f"Temp directory {tempdir} was deleted")
 

custom_components/robonomics_report_service/rws_registration.py

@@ -4,18 +4,19 @@
 from .robonomics import Robonomics
 from .libp2p import LibP2P
 from .utils import pinata_creds_exists, async_remove_store
-from .const import STORAGE_PINATA_CREDS
+from .const import STORAGE_PINATA_CREDS, SERVICE_PAID
 
 _LOGGER = logging.getLogger(__name__)
 
 class RWSRegistrationManager:
 
     @staticmethod
-    async def register(hass: HomeAssistant, robonomics: Robonomics, email: str) -> None:
-        libp2p = LibP2P(hass, robonomics.sender_seed, email)
+    async def register(hass: HomeAssistant, robonomics: Robonomics, libp2p: LibP2P) -> None:
+        libp2p = libp2p
         if not await pinata_creds_exists(hass):
             await libp2p.get_and_save_pinata_creds()
-        await robonomics.wait_for_rws()
+        if SERVICE_PAID:
+            await robonomics.wait_for_rws()
 
     @staticmethod
     async def delete(hass: HomeAssistant) -> None:

custom_components/robonomics_report_service/utils.py

@@ -6,6 +6,8 @@
 import typing as tp
 import json
 
+from os.path import isdir
+
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.json import JSONEncoder
 from homeassistant.helpers.storage import Store
@@ -224,3 +226,12 @@ def delete_temp_dir(dirpath: str) -> None:
     """
     shutil.rmtree(dirpath)
 
+def get_tempdir_filenames(dirname_prefix: str) -> tp.List[str]:
+    temp_dirname = tempfile.gettempdir()
+    filenames = os.listdir(temp_dirname)
+    temdir_names = []
+    for filename in filenames:
+        filepath = f"{temp_dirname}/{filename}"
+        if isdir(filepath) and filename.startswith(dirname_prefix):
+            temdir_names.append(filepath)
+    return temdir_names