Customize logging colors

duties/cli/arguments.py

@@ -50,6 +50,32 @@ def __get_raw_arguments() -> Namespace:
         action="store_true",
         default=False,
     )
+    parser.add_argument(
+        "--log-color-warning",
+        type=parse.set_logging_color,
+        help=(
+            "The logging color as hex or rgb code for warning logs (default: '255,255,0' - yellow)"
+        ),
+        action="store",
+        default=[255, 255, 0],
+    )
+    parser.add_argument(
+        "--log-color-critical",
+        type=parse.set_logging_color,
+        help="The logging color as hex or rgb code for critical logs (default: '255, 0, 0' - red)",
+        action="store",
+        default=[255, 0, 0],
+    )
+    parser.add_argument(
+        "--log-color-proposing",
+        type=parse.set_logging_color,
+        help=(
+            "The logging color as hex or rgb code for proposing duty logs "
+            "(default: '0, 128, 0' - green)"
+        ),
+        action="store",
+        default=[0, 128, 0],
+    )
     parser.add_argument(
         "--log-time-warning",
         type=float,

duties/cli/parse.py

@@ -3,6 +3,8 @@
 
 from typing import List
 
+from constants.program import HEX_COLOR_STARTING_POSITIONS, HEX_TO_INT_BASE
+
 
 def set_validator_identifiers(validators: str) -> List[str]:
     """Parse provided validators for space and comma separation
@@ -42,3 +44,32 @@ def set_beacon_node_urls(beacon_node_urls: str) -> List[str]:
     if not beacon_node_urls.startswith(("http://", "https://")):
         raise ValueError()
     return [beacon_node_urls]
+
+
+def set_logging_color(logging_color: str) -> List[int]:
+    """Parse provided logging color
+
+    Args:
+        logging_color (str): Logging color in hex or RGB code
+
+    Raises:
+        ValueError: Error if RGB codes are not in range 0-255
+
+    Returns:
+        List[int]: RGB color code
+    """
+    if logging_color.startswith("#"):
+        logging_color = logging_color[1:]
+        rgb_color_codes = [
+            int(logging_color[position : position + 2], HEX_TO_INT_BASE)
+            for position in HEX_COLOR_STARTING_POSITIONS
+        ]
+    else:
+        string_rgb_color_codes = logging_color.split(",")
+        rgb_color_codes = [int(rgb_code) for rgb_code in string_rgb_color_codes]
+    filtered_rgb_color_codes = [
+        rgb_code for rgb_code in rgb_color_codes if rgb_code >= 0 if rgb_code <= 255
+    ]
+    if len(filtered_rgb_color_codes) != 3:
+        raise ValueError()
+    return filtered_rgb_color_codes

duties/constants/program.py

@@ -36,3 +36,6 @@
 SECONDS_UNTIL_BEACON_NODE_CALL_ERROR_IS_LOGGED_AGAIN = 5
 REST_RAW_DUTY_NO_BEACON_NODE_CONNECTION_TIMEOUT = 7
 REST_ANY_DUTY_NO_BEACON_NODE_CONNECTION_TIMEOUT = 10
+HEX_COLOR_STARTING_POSITIONS = (0, 2, 4)
+HEX_TO_INT_BASE = 16
+USED_STY_BACKGROUND_COLOR_NAMES = ["yellow", "red", "green"]

duties/fetcher/__init__.py

@@ -3,18 +3,20 @@
 
 import sys
 from logging import config as logging_config
-from os import path
+from os import path, system
 from typing import Any
 
 from cli.arguments import ARGUMENTS
-from colorama import init
+from constants.program import USED_STY_BACKGROUND_COLOR_NAMES
+from sty import RgbBg, Style, bg  # type: ignore[import]
 from yaml import safe_load
 
 
 def __initialize() -> None:
     """Initializes logger and colorama"""
+    system("")
     __initialize_logging(ARGUMENTS.log)
-    __initialize_colorama()
+    __set_colors()
 
 
 def __initialize_logging(log_level: str) -> None:
@@ -49,9 +51,26 @@ def __get_logging_configuration_path() -> str:
     return logging_configuration_path
 
 
-def __initialize_colorama() -> None:
-    """Initializes coloroma so that colorful logging works independent from OS"""
-    init()
+def __set_colors() -> None:
+    """Overrides used color attributes of sty package"""
+    log_color_arguments = [
+        argument
+        # pylint: disable-next=protected-access
+        for argument in ARGUMENTS._get_kwargs()
+        if argument[0].startswith("log_color")
+    ]
+    for index, color_name in enumerate(USED_STY_BACKGROUND_COLOR_NAMES):
+        setattr(
+            bg,
+            color_name,
+            Style(
+                RgbBg(
+                    log_color_arguments[index][1][0],
+                    log_color_arguments[index][1][1],
+                    log_color_arguments[index][1][2],
+                )
+            ),
+        )
 
 
 __initialize()

duties/fetcher/data_types.py

@@ -5,7 +5,7 @@
 from enum import Enum
 from typing import List
 
-from dataclass_wizard import JSONWizard
+from dataclass_wizard import JSONWizard  # type: ignore[import]
 
 
 class DutyType(Enum):

duties/fetcher/log.py

@@ -7,11 +7,11 @@
 from typing import List, Tuple
 
 from cli.arguments import ARGUMENTS
-from colorama import Back, Style
 from constants import logging, program
 from fetcher.data_types import DutyType, ValidatorDuty, ValidatorIdentifier
 from fetcher.identifier.core import read_validator_identifiers_from_shared_memory
 from protocol import ethereum
+from sty import bg, rs  # type: ignore[import]
 
 __validator_identifiers_with_alias = {"0": ValidatorIdentifier()}
 
@@ -71,7 +71,7 @@ def __create_logging_message(duty: ValidatorDuty) -> str:
             f"{__get_logging_color(duty.time_to_duty, duty)}"
             f"Validator {__get_validator_identifier_for_logging(duty)} "
             f"has next {duty.type.name} duty in: "
-            f"{time_to_next_duty} min. (slot: {duty.slot}){Style.RESET_ALL}"
+            f"{time_to_next_duty} min. (slot: {duty.slot}){rs.all}"
         )
     return logging_message
 
@@ -94,17 +94,17 @@ def __create_sync_committee_logging_message(sync_committee_duty: ValidatorDuty)
     )
     if sync_committee_duty.time_to_duty == 0:
         logging_message = (
-            f"{Back.RED}Validator {__get_validator_identifier_for_logging(sync_committee_duty)} "
+            f"{bg.red}Validator {__get_validator_identifier_for_logging(sync_committee_duty)} "
             f"is in current sync committee (next sync committee starts in "
             f"{time_to_next_sync_committee} / "
-            f"epoch: {current_sync_committee_epoch_boundaries[1] + 1}){Style.RESET_ALL}"
+            f"epoch: {current_sync_committee_epoch_boundaries[1] + 1}){rs.all}"
         )
     else:
         logging_message = (
-            f"{Back.YELLOW}Validator "
+            f"{bg.yellow}Validator "
             f"{__get_validator_identifier_for_logging(sync_committee_duty)} will be in next "
             f"sync committee which starts in {time_to_next_sync_committee} "
-            f"(epoch: {current_sync_committee_epoch_boundaries[1] + 1}){Style.RESET_ALL}"
+            f"(epoch: {current_sync_committee_epoch_boundaries[1] + 1}){rs.all}"
         )
     return logging_message
 
@@ -144,12 +144,12 @@ def __get_logging_color(seconds_to_next_duty: float, duty: ValidatorDuty) -> str
         str: ANSI codes for colorful logging
     """
     if ARGUMENTS.log_time_critical < seconds_to_next_duty <= ARGUMENTS.log_time_warning:
-        return Back.YELLOW
+        return bg.yellow
     if seconds_to_next_duty <= ARGUMENTS.log_time_critical:
-        return Back.RED
+        return bg.red
     if duty.type is DutyType.PROPOSING:
-        return Back.GREEN
-    return Style.RESET_ALL
+        return bg.green
+    return rs.all
 
 
 def __get_validator_identifier_for_logging(duty: ValidatorDuty) -> str:

pyproject.toml

@@ -8,13 +8,13 @@ repository = "https://github.com/TobiWo/eth-duties"
 version = "0.4.0"
 
 [tool.poetry.dependencies]
-colorama = "==0.4.6"
 dataclass-wizard = "==0.22.0"
 eth-typing = "==3.4.0"
 fastapi = "==0.103.0"
 python = ">=3.10.0,<3.11"
 pyyaml = "==6.0.1"
 requests = "==2.31.0"
+sty = "==1.0.4"
 uvicorn = "==0.23.2"
 
 [tool.poetry.group.dev.dependencies]
@@ -25,7 +25,6 @@ pyinstaller = "==4.10.0"
 pylint = "==2.17.5"
 rope = "==1.4.0"
 types-PyYAML = "==6.0.12.11"
-types-colorama = "==0.4.15.12"
 types-requests = "==2.31.0.2"
 
 [build-system]