text: import re
  from typing import TYPE_CHECKING, Union
  
  import click
  import gevent
  import requests
  import structlog
  from eth_utils import to_hex
  from gevent.event import AsyncResult
  from packaging.version import LegacyVersion, Version, parse as parse_version
  from web3 import Web3
  from web3.types import BlockData
  
  from raiden.api.objects import Notification
  from raiden.constants import (
      BLOCK_ID_LATEST,
      CHECK_CHAIN_ID_INTERVAL,
      CHECK_GAS_RESERVE_INTERVAL,
      CHECK_RDN_MIN_DEPOSIT_INTERVAL,
      CHECK_VERSION_INTERVAL,
      LATEST,
      RELEASE_PAGE,
      SECURITY_EXPRESSION,
      NotificationIDs,
  )
  from raiden.network.proxies.proxy_manager import ProxyManager
  from raiden.network.proxies.user_deposit import UserDeposit
  from raiden.settings import MIN_REI_THRESHOLD
  from raiden.utils import gas_reserve
  from raiden.utils.formatting import to_checksum_address
  from raiden.utils.runnable import Runnable
  from raiden.utils.transfers import to_rdn
  from raiden.utils.typing import Any, BlockNumber, Callable, ChainID, List, Optional
  
  if TYPE_CHECKING:
      from raiden.raiden_service import RaidenService
  
  REMOVE_CALLBACK = object()
  log = structlog.get_logger(__name__)
  
  
  def _do_check_version(
      current_version: Union[Version, LegacyVersion], raiden: "RaidenService"
  ) -> bool:
      content = requests.get(LATEST).json()
      if "tag_name" not in content:
          # probably API rate limit exceeded
          click.secho(
              "Error while contacting github for latest version. API rate limit exceeded?", fg="red"
          )
          return False
      # getting the latest release version
      latest_release = parse_version(content["tag_name"])
      security_message = re.search(SECURITY_EXPRESSION, content["body"])
      if security_message:
          notification = Notification(
              id=NotificationIDs.VERSION_SECURITY_WARNING.value,
              summary="Security Warning",
              body=security_message.group(0),
              urgency="high",
          )
          raiden.add_notification(notification, click_opts={"fg": "red"})
  
          # comparing it to the user's application
      if current_version < latest_release:
          msg = (
              f"You're running version {current_version}. The latest version is {latest_release}"
              f"It's time to update! Releases: {RELEASE_PAGE}"
          )
          notification = Notification(
              id=NotificationIDs.VERSION_OUTDATED.value,
              summary="Your version is outdated",
              body=msg,
              urgency="normal",
          )
          raiden.add_notification(notification, click_opts={"fg": "red"})
          return False
      return True
  
  
  def check_version(current_version: str, raiden: "RaidenService") -> None:  # pragma: no unittest
      """Check periodically for a new release"""
      app_version = parse_version(current_version)
      while True:
          try:
              _do_check_version(app_version, raiden)
          except (requests.exceptions.HTTPError, ValueError) as err:
              click.secho("Error while checking for version", fg="red")
              print(err)
  
          # repeat the process once every 3h
          gevent.sleep(CHECK_VERSION_INTERVAL)
  
  
  def check_gas_reserve(raiden: "RaidenService") -> None:  # pragma: no unittest
      """Check periodically for gas reserve in the account"""
      while True:
          has_enough_balance, estimated_required_balance = gas_reserve.has_enough_gas_reserve(
              raiden, channels_to_open=1
          )
          estimated_required_balance_eth = Web3.fromWei(estimated_required_balance, "ether")
  
          if not has_enough_balance:
              notification_body = (
                  "WARNING\n"
                  "Your account's balance is below the estimated gas reserve of "
                  f"{estimated_required_balance_eth} eth. This may lead to a loss of "
                  "of funds because your account will be unable to perform on-chain "
                  "transactions. Please add funds to your account as soon as possible."
              )
              notification = Notification(
                  id=NotificationIDs.MISSING_GAS_RESERVE.value,
                  summary="Missing gas reserve",
                  body=notification_body,
                  urgency="normal",
              )
              raiden.add_notification(
                  notification,
                  log_opts={"required_wei": estimated_required_balance},
                  click_opts={"fg": "red"},
              )
  
          gevent.sleep(CHECK_GAS_RESERVE_INTERVAL)
  
  
  def check_rdn_deposits(
      raiden: "RaidenService", user_deposit_proxy: UserDeposit
  ) -> None:  # pragma: no unittest
      """Check periodically for RDN deposits in the user-deposits contract"""
      while True:
          rei_balance = user_deposit_proxy.effective_balance(raiden.address, BLOCK_ID_LATEST)
          rdn_balance = to_rdn(rei_balance)
          if rei_balance < MIN_REI_THRESHOLD:
              notification_body = (
                  f"WARNING\n"
                  f"Your account's RDN balance deposited in the UserDepositContract of "
                  f"{rdn_balance} is below the minimum threshold {to_rdn(MIN_REI_THRESHOLD)}. "
                  f"Provided that you have either a monitoring service or a path "
                  f"finding service activated, your node is not going to be able to "
                  f"pay those services which may lead to denial of service or loss of funds."
              )
              notification = Notification(
                  id=NotificationIDs.LOW_RDN.value,
                  summary="RDN balance too low",
                  body=notification_body,
                  urgency="normal",
              )
              raiden.add_notification(notification, click_opts={"fg": "red"})
  
          gevent.sleep(CHECK_RDN_MIN_DEPOSIT_INTERVAL)
  
  
  def check_chain_id(chain_id: ChainID, web3: Web3) -> None:  # pragma: no unittest
      """Check periodically if the underlying ethereum client's network id has changed"""
      while True:
          try:
              current_id = web3.eth.chain_id
          except requests.exceptions.ConnectionError:
              raise RuntimeError(
                  "Could not reach ethereum RPC. "
                  "Please check that your ethereum node is running and accessible."
              )
          if chain_id != current_id:
              raise RuntimeError(
                  f"Raiden was running on network with id {chain_id} and it detected "
                  f"that the underlying ethereum client network id changed to {current_id}."
                  f" Changing the underlying blockchain while the Raiden node is running "
                  f"is not supported."
              )
          gevent.sleep(CHECK_CHAIN_ID_INTERVAL)
  
  
  class AlarmTask(Runnable):
      """Task to notify when a block is mined."""
  
      def __init__(self, proxy_manager: ProxyManager, sleep_time: float) -> None:
          super().__init__()
  
          self.callbacks: List[Callable] = []
          self.proxy_manager = proxy_manager
          self.rpc_client = proxy_manager.client
  
          self.known_block_number: Optional[BlockNumber] = None
          self._stop_event: Optional[AsyncResult] = None
  
          # TODO: Start with a larger sleep_time and decrease it as the
          # probability of a new block increases.
          self.sleep_time = sleep_time
  
      def __repr__(self) -> str:
          return (
              f"<{self.__class__.__name__} node:" f"{to_checksum_address(self.rpc_client.address)}>"
          )
  
      def start(self) -> None:
          log.debug("Alarm task started", node=to_checksum_address(self.rpc_client.address))
          self._stop_event = AsyncResult()
          super().start()
  
      def _run(self, *args: Any, **kwargs: Any) -> None:  # pylint: disable=method-hidden
          self.greenlet.name = f"AlarmTask._run node:{to_checksum_address(self.rpc_client.address)}"
          try:
              self.loop_until_stop()
          finally:
              self.callbacks = []
  
      def register_callback(self, callback: Callable) -> None:
          """Register a new callback.
  
          Note:
              The callback will be executed in the AlarmTask context and for
              this reason it should not block, otherwise we can miss block
              changes.
          """
          if not callable(callback):
              raise ValueError("callback is not a callable")
  
          self.callbacks.append(callback)
  
      def remove_callback(self, callback: Callable) -> None:
          """Remove callback from the list of callbacks if it exists"""
          if callback in self.callbacks:
              self.callbacks.remove(callback)
  
      def loop_until_stop(self) -> None:
          sleep_time = self.sleep_time
          while self._stop_event and self._stop_event.wait(sleep_time) is not True:
              latest_block = self.rpc_client.get_block(block_identifier=BLOCK_ID_LATEST)
  
              self._maybe_run_callbacks(latest_block)
  
      def _maybe_run_callbacks(self, latest_block: BlockData) -> None:
          """Run the callbacks if there is at least one new block.
  
          The callbacks are executed only if there is a new block, otherwise the
          filters may try to poll for an inexisting block number and the Ethereum
          client can return an JSON-RPC error.
          """
          latest_block_number = latest_block["number"]
  
          # First run, set the block and run the callbacks
          if self.known_block_number is None:
              self.known_block_number = latest_block_number
              missed_blocks = 1
          else:
              missed_blocks = latest_block_number - self.known_block_number
  
          if missed_blocks < 0:
              log.critical(
                  "Block number decreased",
                  chain_id=self.rpc_client.chain_id,
                  known_block_number=self.known_block_number,
                  old_block_number=latest_block["number"],
                  old_gas_limit=latest_block["gasLimit"],
                  old_block_hash=to_hex(latest_block["hash"]),
                  node=to_checksum_address(self.rpc_client.address),
              )
          elif missed_blocks > 0:
              log_details = dict(
                  known_block_number=self.known_block_number,
                  latest_block_number=latest_block_number,
                  latest_block_hash=to_hex(latest_block["hash"]),
                  latest_block_gas_limit=latest_block["gasLimit"],
                  node=to_checksum_address(self.rpc_client.address),
              )
              if missed_blocks > 1:
                  log_details["num_missed_blocks"] = missed_blocks - 1
  
              log.debug("Received new block", **log_details)
  
              remove = []
              for callback in self.callbacks:
                  result = callback(latest_block)
                  if result is REMOVE_CALLBACK:
                      remove.append(callback)
  
              for callback in remove:
                  self.callbacks.remove(callback)
  
              self.known_block_number = latest_block_number
  
      def stop(self) -> Any:
          if self._stop_event:
              self._stop_event.set(True)
          log.debug("Alarm task stopped", node=to_checksum_address(self.rpc_client.address))
          result = self.greenlet.join()
          # Callbacks should be cleaned after join
          self.callbacks = []
          return result
  