Merge pull request #85 from airalab/dev

add crust

connectivity/src/feeders/datalog_feeder.py

@@ -3,6 +3,7 @@
 Every `dump_interval` (from the config) the buffer writes to the file which pins to IPFS.
 IPFS hash of the file sends to Robonomics Datalog.
 """
+from crustinterface import Mainnet
 import json
 import logging.config
 import os
@@ -56,8 +57,8 @@ def _sort_payload(data: dict) -> dict:
     return ordered
 
 
-def _get_multihash(buf: set, db: object, endpoint: str = "/ip4/127.0.0.1/tcp/5001/http") -> tp.Dict[str, str]:
-    """Write sorted measurements to the temp file, add file to IPFS and add 
+def _get_multihash(buf: set, db: object, endpoint: str = "/ip4/127.0.0.1/tcp/5001/http") -> tuple:
+    """Write sorted measurements to the temp file, add file to IPFS and add
     measurements and hash in the database with 'not sent' status.
 
     :param buf: Set of measurements from all sensors.
@@ -97,7 +98,7 @@ def _get_multihash(buf: set, db: object, endpoint: str = "/ip4/127.0.0.1/tcp/500
     with ipfshttpclient2.connect(endpoint) as client:
         response = client.add(temp.name)
         db.add_data("not sent", response["Hash"], time.time(), json.dumps(payload))
-        return (response["Hash"], temp.name)
+        return (response["Hash"], temp.name, response["Size"])
 
 
 def _pin_to_pinata(file_path: str, config: dict) -> None:
@@ -121,6 +122,34 @@ def _pin_to_pinata(file_path: str, config: dict) -> None:
             logger.warning(f"DatalogFeeder: Failed while pining file to Pinata. Error: {e}")
 
 
+def _upload_to_crust(hash: str, file_size: int, seed: str) -> None:
+    mainnet = Mainnet(seed=seed)
+    try:
+        # Check balance
+        balance = mainnet.get_balance()
+        logger.debug(f"DatalogFeeder: Actual balance in crust network - {balance}")
+
+        # Check price in Main net. Price in pCRUs
+        price = mainnet.get_appx_store_price(file_size)
+        logger.debug(f"DatalogFeeder: Approximate cost to store the file - {price}")
+
+    except Exception as e:
+        logger.warning(f"DatalogFeeder: Error while getting account balance - {e}")
+        return None
+
+    if price >= balance:
+        logger.warning(f"DatalogFeeder: Not enough account balance to store the file in Crust Network")
+        return None
+
+    try:
+        logger.info(f"DatalogFeeder: Start adding {hash} to crust with size {file_size}")
+        file_stored = mainnet.store_file(hash, file_size)
+        logger.info(f"DatalogFeeder: File stored in Crust. Extrinsic data is  {file_stored}")
+    except Exception as e:
+        logger.warning(f"error while uploading file to crust - {e}")
+        return None
+
+
 class DatalogFeeder(IFeeder):
     """
     The feeder is responsible for collecting measurements and
@@ -148,8 +177,8 @@ def __init__(self, config) -> None:
         self.db.create_table()
 
     def feed(self, data: tp.List[dict]) -> None:
-        """Main function of the feeder and it is called in `main.py`. It collects 
-        data into buffer and, every `interval` from config, adds it to IPFS and sends the hash 
+        """Main function of the feeder and it is called in `main.py`. It collects
+        data into buffer and, every `interval` from config, adds it to IPFS and sends the hash
         to Robonomics Datalog.
 
         :param data: Data from the stations.
@@ -165,9 +194,10 @@ def feed(self, data: tp.List[dict]) -> None:
                     if self.buffer:
                         logger.debug("Datalog Feeder: About to publish collected data...")
                         logger.debug(f"Datalog Feeder: Buffer is {self.buffer}")
-                        ipfs_hash, file_path = _get_multihash(self.buffer, self.db, self.ipfs_endpoint)
+                        ipfs_hash, file_path, file_size = _get_multihash(self.buffer, self.db, self.ipfs_endpoint)
                         self._pin_to_temporal(file_path)
                         _pin_to_pinata(file_path, self.config)
+                        _upload_to_crust(ipfs_hash, int(file_size), self.config["datalog"]["suri"])
                         os.unlink(file_path)
                         self.to_datalog(ipfs_hash)
                     else:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "sensors_connectivity"
-version = "1.5.4"
+version = "1.6.0"
 description = "Robonomics package to read data from sensors and publish to different output channels"
 authors = [
     "Vadim Manaenko <vadim.razorq@gmail.com>",
@@ -34,6 +34,7 @@ paho-mqtt = "^1.6.1"
 prometheus-client = "^0.13.1"
 py-sr25519-bindings = "^0.2.0"
 IPFS-Toolkit = "^0.4.4"
+crust-interface-patara = "^0.1.1"
 
 [tool.poetry.scripts]
 sensors_connectivity = "connectivity.main:run"