Add config for filtering small wallet transactions (#336)

Co-authored-by: egeedot <[email protected]>

config-example.yaml

@@ -33,6 +33,28 @@ daily_stats:
   time_of_day: "21:00"
   frequency_hours: 24
 
+# In this section you can enable and configure the log handlers
+handlers:
+  # Checks for new coins received (wallet)
+  wallet_added_coin_handler:
+    enable: true
+    # Transactions with lower amount mojos will be ignored
+    # Use this to avoid notification spam during dust storms
+    min_mojos_amount:  5
+  # Checks for skipped signage points (full node)
+  finished_signage_point_handler:
+    enable: true
+  # Checks for new blocks found (full node)
+  block_handler:
+    enable: true
+  # Checks for found partials (farmer)
+  partial_handler:
+    enable: true
+  # Checks harvester activity & health (harvester)
+  harvester_activity_handler:
+    enable: true
+
+
 # We support a lot of notifiers, please check the README for more
 # information. You can delete the sections which you aren't using.
 # You can also enable more than one notifier and send different

main.py

@@ -71,7 +71,8 @@ def init(config:Config):
 
     # Link stuff up in the log handler
     # Pipeline: Consume -> Handle -> Notify
-    log_handler = LogHandler(log_consumer=log_consumer, notify_manager=notify_manager, stats_manager=stats_manager)
+    log_handler = LogHandler(config=config.get_handlers_config(), log_consumer=log_consumer, notify_manager=notify_manager,
+                             stats_manager=stats_manager)
 
     def interrupt(signal_number, frame):
         if signal_number == signal.SIGINT:

src/chia_log/handlers/__init__.py

@@ -9,15 +9,24 @@ class that analyses a specific part of the log
 # std
 from abc import ABC, abstractmethod
 from typing import List, Optional
+import logging
 
 # project
 from .daily_stats.stats_manager import StatsManager
 from src.notifier import Event
 
 
-class LogHandler(ABC):
+class LogHandlerInterface(ABC):
     """Common interface for log handlers"""
 
+    @staticmethod
+    @abstractmethod
+    def config_name() -> str:
+        pass
+
+    def __init__(self, config: Optional[dict] = None):
+        logging.info(f"Initializing handler: {self.config_name()}")
+
     @abstractmethod
     def handle(self, logs: str, stats_manager: Optional[StatsManager] = None) -> List[Event]:
         pass

src/chia_log/handlers/block_handler.py

@@ -3,21 +3,26 @@
 from typing import List, Optional
 
 # project
-from . import LogHandler
+from . import LogHandlerInterface
 from ..parsers.block_parser import BlockParser
 from .condition_checkers import BlockConditionChecker
 from .condition_checkers.found_blocks import FoundBlocks
 from .daily_stats.stats_manager import StatsManager
 from src.notifier import Event
 
 
-class BlockHandler(LogHandler):
+class BlockHandler(LogHandlerInterface):
     """This handler parses all logs indicating found block
     activity by the full node. It holds a list of condition checkers
     that are evaluated for each event.
     """
 
-    def __init__(self):
+    @staticmethod
+    def config_name() -> str:
+        return "block_handler"
+
+    def __init__(self, config: Optional[dict] = None):
+        super().__init__(config)
         self._parser = BlockParser()
         self._cond_checkers: List[BlockConditionChecker] = [FoundBlocks()]
 

src/chia_log/handlers/finished_signage_point_handler.py

@@ -3,21 +3,26 @@
 from typing import List, Optional
 
 # project
-from . import LogHandler
+from . import LogHandlerInterface
 from ..parsers.finished_signage_point_parser import FinishedSignagePointParser
 from .condition_checkers import FinishedSignageConditionChecker
 from .condition_checkers.non_skipped_signage_points import NonSkippedSignagePoints
 from .daily_stats.stats_manager import StatsManager
 from src.notifier import Event
 
 
-class FinishedSignagePointHandler(LogHandler):
+class FinishedSignagePointHandler(LogHandlerInterface):
     """This handler parses all logs indicating finished signage point
     activity by the full node. It holds a list of condition checkers
     that are evaluated for each event.
     """
 
-    def __init__(self):
+    @staticmethod
+    def config_name() -> str:
+        return "finished_signage_point_handler"
+
+    def __init__(self, config: Optional[dict] = None):
+        super().__init__(config)
         self._parser = FinishedSignagePointParser()
         self._cond_checkers: List[FinishedSignageConditionChecker] = [NonSkippedSignagePoints()]
 

src/chia_log/handlers/harvester_activity_handler.py

@@ -3,7 +3,7 @@
 from typing import List, Optional
 
 # project
-from . import LogHandler
+from . import LogHandlerInterface
 from ..parsers.harvester_activity_parser import HarvesterActivityParser
 from .condition_checkers import HarvesterConditionChecker
 from .condition_checkers.non_decreasing_plots import NonDecreasingPlots
@@ -13,14 +13,19 @@
 from src.notifier import Event, EventService, EventType, EventPriority
 
 
-class HarvesterActivityHandler(LogHandler):
+class HarvesterActivityHandler(LogHandlerInterface):
     """This handler parses all logs indicating harvester
     activity and participation in challenges. It holds a list
     of condition checkers that are evaluated for each event to
     ensure that farming is going smoothly.
     """
 
-    def __init__(self):
+    @staticmethod
+    def config_name() -> str:
+        return "harvester_activity_handler"
+
+    def __init__(self, config: Optional[dict] = None):
+        super().__init__(config)
         self._parser = HarvesterActivityParser()
         self._cond_checkers: List[HarvesterConditionChecker] = [
             TimeSinceLastFarmEvent(),

src/chia_log/handlers/partial_handler.py

@@ -2,20 +2,25 @@
 from typing import List, Optional
 
 # project
-from . import LogHandler
+from . import LogHandlerInterface
 from ..parsers.partial_parser import PartialParser
 from .condition_checkers import PartialConditionChecker
 from .daily_stats.stats_manager import StatsManager
 from src.notifier import Event
 
 
-class PartialHandler(LogHandler):
+class PartialHandler(LogHandlerInterface):
     """This handler parses all logs indicating found partials
     activity by the full node. It holds a list of condition checkers
     that are evaluated for each event.
     """
 
-    def __init__(self):
+    @staticmethod
+    def config_name() -> str:
+        return "partial_handler"
+
+    def __init__(self, config: Optional[dict] = None):
+        super().__init__(config)
         self._parser = PartialParser()
         self._cond_checkers: List[PartialConditionChecker] = []
 

src/chia_log/handlers/wallet_added_coin_handler.py

@@ -3,19 +3,27 @@
 from typing import List, Optional
 
 # project
-from . import LogHandler
+from . import LogHandlerInterface
 from ..parsers.wallet_added_coin_parser import WalletAddedCoinParser
 from .daily_stats.stats_manager import StatsManager
 from src.notifier import Event, EventService, EventType, EventPriority
 
 
-class WalletAddedCoinHandler(LogHandler):
+class WalletAddedCoinHandler(LogHandlerInterface):
     """This handler parses all logs that report wallet
     receiving XCH and creates user notifications.
     """
 
-    def __init__(self):
+    @staticmethod
+    def config_name() -> str:
+        return "wallet_added_coin_handler"
+
+    def __init__(self, config: Optional[dict] = None):
+        super().__init__(config)
         self._parser = WalletAddedCoinParser()
+        config = config or {}
+        self.min_mojos_amount = config.get("min_mojos_amount", 0)
+        logging.info(f"Filtering transaction with mojos less than {self.min_mojos_amount}")
 
     def handle(self, logs: str, stats_manager: Optional[StatsManager] = None) -> List[Event]:
         events = []
@@ -28,7 +36,7 @@ def handle(self, logs: str, stats_manager: Optional[StatsManager] = None) -> Lis
             logging.info(f"Cha-ching! Just received {coin_msg.amount_mojos} mojos.")
             total_mojos += coin_msg.amount_mojos
 
-        if total_mojos > 0:
+        if total_mojos > self.min_mojos_amount:
             chia_coins = total_mojos / 1e12
             xch_string = f"{chia_coins:.12f}".rstrip("0").rstrip(".")
             events.append(
@@ -39,5 +47,10 @@ def handle(self, logs: str, stats_manager: Optional[StatsManager] = None) -> Lis
                     message=f"Cha-ching! Just received {xch_string} XCH ☘️",
                 )
             )
+        else:
+            logging.debug(
+                f"Filtering out chia coin message since the amount ${total_mojos} received is less than"
+                f"the configured transaction_amount: ${self.min_mojos_amount}"
+            )
 
         return events

src/chia_log/log_handler.py

@@ -1,7 +1,9 @@
 # std
-from typing import Optional
+from typing import Optional, List, Type
+import logging
 
 # project
+from src.chia_log.handlers import LogHandlerInterface
 from src.chia_log.handlers.daily_stats.stats_manager import StatsManager
 from src.chia_log.handlers.harvester_activity_handler import HarvesterActivityHandler
 from src.chia_log.handlers.partial_handler import PartialHandler
@@ -12,6 +14,15 @@
 from src.notifier.notify_manager import NotifyManager
 
 
+def _check_handler_enabled(config: dict, handler_name: str) -> bool:
+    """Fallback to True for backwards compatability"""
+    try:
+        return config[handler_name].get("enable", True)
+    except KeyError as key:
+        logging.error(f"Invalid config.yaml. Missing key: {key}")
+    return True
+
+
 class LogHandler(LogConsumerSubscriber):
     """This class holds a list of handlers that analyze
     specific parts of the logs and generate events that
@@ -27,17 +38,30 @@ class LogHandler(LogConsumerSubscriber):
     """
 
     def __init__(
-        self, log_consumer: LogConsumer, notify_manager: NotifyManager, stats_manager: Optional[StatsManager] = None
+        self,
+        config: Optional[dict],
+        log_consumer: LogConsumer,
+        notify_manager: NotifyManager,
+        stats_manager: Optional[StatsManager] = None,
     ):
         self._notify_manager = notify_manager
         self._stats_manager = stats_manager
-        self._handlers = [
-            HarvesterActivityHandler(),
-            PartialHandler(),
-            BlockHandler(),
-            FinishedSignagePointHandler(),
-            WalletAddedCoinHandler(),
+
+        config = config or {}
+        available_handlers: List[Type[LogHandlerInterface]] = [
+            HarvesterActivityHandler,
+            PartialHandler,
+            BlockHandler,
+            FinishedSignagePointHandler,
+            WalletAddedCoinHandler,
         ]
+        self._handlers = []
+        for handler in available_handlers:
+            if _check_handler_enabled(config, handler.config_name()):
+                self._handlers.append(handler(config.get(handler.config_name())))
+            else:
+                logging.info(f"Disabled handler: {handler.config_name()}")
+
         log_consumer.subscribe(self)
 
     def consume_logs(self, logs: str):

src/config.py

@@ -34,6 +34,9 @@ def get_notifier_config(self):
     def get_chia_logs_config(self):
         return self._get_child_config("chia_logs")
 
+    def get_handlers_config(self):
+        return self._get_child_config("handlers", required=False)
+
     def get_log_level_config(self):
         return self._get_child_config("log_level")
 

src/notifier/ifttt_notifier.py

@@ -2,7 +2,6 @@
 import http.client
 import logging
 import json
-import urllib.parse
 from typing import List
 
 # project

src/notifier/pushcut_notifier.py

@@ -2,7 +2,6 @@
 import http.client
 import logging
 import json
-import urllib.parse
 from typing import List
 
 # project

tests/chia_log/handlers/test_wallet_added_coin_handler.py

@@ -1,17 +1,27 @@
 # std
 import unittest
 from pathlib import Path
+import copy
 
 # project
 from src.chia_log.handlers.wallet_added_coin_handler import WalletAddedCoinHandler
 from src.notifier import EventType, EventService, EventPriority
+from src.config import Config
 
 
 class TestWalledAddedCoinHandler(unittest.TestCase):
     def setUp(self) -> None:
-        self.handler = WalletAddedCoinHandler()
+        config_dir = Path(__file__).resolve().parents[3]
+        config = Config(config_dir / "config-example.yaml")
+        self.handler_config = config.get_handlers_config()[WalletAddedCoinHandler.config_name()]
+
+        self.handler = WalletAddedCoinHandler(config=None)
         self.example_logs_path = Path(__file__).resolve().parents[1] / "logs/wallet_added_coin"
 
+    def testConfig(self):
+        self.assertEqual(self.handler_config["enable"], True)
+        self.assertEqual(self.handler_config["min_mojos_amount"], 5)
+
     def testNominal(self):
         with open(self.example_logs_path / "nominal.txt", encoding="UTF-8") as f:
             logs = f.readlines()
@@ -34,6 +44,20 @@ def testFloatPrecision(self):
         self.assertEqual(events[0].service, EventService.WALLET, "Unexpected service")
         self.assertEqual(events[0].message, "Cha-ching! Just received 0.000000000001 XCH ☘️")
 
+    def testTransactionAmountFilter(self):
+        default_config = self.handler_config
+        no_filter_config = copy.deepcopy(default_config)
+        no_filter_config["min_mojos_amount"] = 0
+
+        filter_handler = WalletAddedCoinHandler(default_config)
+        no_filter_handler = WalletAddedCoinHandler(no_filter_config)
+        with open(self.example_logs_path / "small_values.txt", encoding="UTF-8") as f:
+            logs = f.readlines()
+        filter_events = filter_handler.handle("".join(logs))
+        self.assertEqual(0, len(filter_events))
+        no_filter_events = no_filter_handler.handle("".join(logs))
+        self.assertEqual(1, len(no_filter_events))
+
 
 if __name__ == "__main__":
     unittest.main()