[Flow] add update_automation_configuration keywords

Author: GuillaumeDSM

octobot/commands.py

@@ -39,8 +39,6 @@
 import octobot.community.tentacles_packages as community_tentacles_packages
 import octobot.configuration_manager as configuration_manager
 
-import octobot.storage.process_bot_state_dumper as process_bot_state_dumper
-
 COMMANDS_LOGGER_NAME = "Commands"
 IGNORED_COMMAND_WHEN_RESTART = ["-u", "--update"]
 
@@ -326,12 +324,6 @@ async def start_bot(bot, logger, catch=False):
             await bot.initialize()
         except asyncio.CancelledError:
             logger.info("Core engine tasks cancelled.")
-        else:
-            if bot.dump_state_path:
-
-                bot._process_bot_state_dump_task = asyncio.create_task(
-                    process_bot_state_dumper.run_periodic_dump_loop(bot.dump_state_path, logger, bot)
-                )
 
     except Exception as e:
         logger.exception(e)

octobot/constants.py

@@ -215,9 +215,6 @@
 DEFAULT_LOGS_FOLDER = "logs"
 LOGS_FOLDER = os.getenv("LOGS_FOLDER", DEFAULT_LOGS_FOLDER)
 
-# Web automation: child process sets OCTOBOT_WEB_API_KEY
-ENV_WEB_API_KEY = "OCTOBOT_WEB_API_KEY"
-WEB_API_KEY_HEADER = "X-Octobot-Api-Key"
 # Process bot state JSON next to user config (--dump-state); liveness for run_octobot_process
 PROCESS_BOT_STATE_FILE_NAME = "process_bot_state.json"
 ENV_PROCESS_BOT_STATE_DUMP_INTERVAL_SECONDS = "OCTOBOT_PROCESS_BOT_STATE_DUMP_INTERVAL_SECONDS"

octobot/octobot.py

@@ -107,7 +107,6 @@ def __init__(self, config: configuration.Configuration, community_authenticator=
         self._init_metadata_run_task = None
         # optional path for periodic ProcessBotState JSON (see cli --dump-state)
         self.dump_state_path = None
-        self._process_bot_state_dump_task = None
 
         # Producers
         self.exchange_producer = None
@@ -214,9 +213,6 @@ async def stop(self):
             self.logger.debug("Stopping ...")
             if self._init_metadata_run_task is not None and not self._init_metadata_run_task.done():
                 self._init_metadata_run_task.cancel()
-            if self._process_bot_state_dump_task is not None and not self._process_bot_state_dump_task.done():
-                self._process_bot_state_dump_task.cancel()
-            self._process_bot_state_dump_task = None
             signals.SignalPublisher.instance().stop()
             if self.evaluator_producer is not None:
                 await self.evaluator_producer.stop()

octobot/storage/process_bot_state_dumper.py

@@ -21,16 +21,20 @@
 
 import aiofiles
 import octobot_commons.json_util as json_util
+import octobot_commons.logging as logging
 import octobot_trading.api as trading_api
 
 import octobot.constants as octobot_app_constants
 import octobot_flow.entities.accounts.exchange_account_elements as exchange_account_elements_import
 import octobot_flow.entities.accounts.process_bot_state as process_bot_state_import
 
 
+if typing.TYPE_CHECKING:
+    import octobot.octobot
+
+
 def _synced_exchange_account_elements_for_first_trading_exchange(
-    octobot: typing.Any,
-    logger: typing.Any,
+    octobot: "octobot.octobot.OctoBot",
 ) -> exchange_account_elements_import.ExchangeAccountElements:
     """
     Build one snapshot for the first trading exchange only. If several are trading, log an error
@@ -51,7 +55,7 @@ def _synced_exchange_account_elements_for_first_trading_exchange(
     elements.name = trading_api.get_exchange_name(first_exchange_manager)
     elements.sync_from_exchange_manager(first_exchange_manager, [])
     for skipped_exchange_manager in trading_managers[1:]:
-        logger.error(
+        _get_logger().error(
             "process bot state dump includes only the first trading exchange; dumping %s (%s). "
             "Skipping additional trading exchange %s (%s).",
             trading_api.get_exchange_name(first_exchange_manager),
@@ -65,14 +69,13 @@ def _synced_exchange_account_elements_for_first_trading_exchange(
 async def _write_state_file_async(
     state_file_path: str,
     interval: float,
-    bot: typing.Any,
-    logger: typing.Any,
+    bot: "octobot.octobot.OctoBot",
 ) -> None:
     now = time.time()
     state = process_bot_state_import.ProcessBotState(
         metadata=process_bot_state_import.Metadata(updated_at=now, next_updated_at=now + interval),
         exchange_account_elements=_synced_exchange_account_elements_for_first_trading_exchange(
-            bot, logger
+            bot,
         ),
     )
     content = state.to_dict(include_default_values=False)
@@ -95,19 +98,23 @@ async def _write_state_file_async(
         raise
 
 
-async def run_periodic_dump_loop(state_file_path: str, logger, bot: typing.Any) -> None:
+async def run_periodic_dump_loop(state_file_path: str, bot: "octobot.octobot.OctoBot") -> None:
     """
     Periodically write ProcessBotState next to the user config. Cancel the task to stop.
     """
     interval = octobot_app_constants.PROCESS_BOT_STATE_DUMP_INTERVAL_SECONDS
     while True:
         try:
-            await _write_state_file_async(state_file_path, interval, bot, logger)
+            await _write_state_file_async(state_file_path, interval, bot)
         except asyncio.CancelledError:
             raise
         except Exception as err:  # pylint: disable=broad-except
-            logger.exception("process bot state dump failed: %s", err)
+            _get_logger().exception(err, True, "process bot state dump failed: %s", err)
         try:
             await asyncio.sleep(interval)
         except asyncio.CancelledError:
             break
+
+
+def _get_logger() -> logging.BotLogger:
+    return logging.get_logger("ProcessBotStateDumper")

octobot/task_manager.py

@@ -24,6 +24,7 @@
 import octobot_commons.constants as commons_constants
 
 import octobot.constants as constants
+import octobot.storage.process_bot_state_dumper as process_bot_state_dumper
 
 
 ASYNC_IGNORED_ERROR_MESSAGES = ["'message': 'Unclosed client session'"]
@@ -48,6 +49,8 @@ def __init__(self, octobot):
         self.executors = None
         self.bot_main_task = None
         self.loop_forever_thread = None
+        
+        self._process_bot_state_dump_task = None
 
     def init_async_loop(self):
         self.async_loop = asyncio.new_event_loop()
@@ -63,6 +66,12 @@ async def start_tools_tasks(self):
         self.ready = True
         self.tools_task_group = asyncio.gather(*task_list)
         self.create_pool_executor()
+        if self.octobot.dump_state_path:
+            self._process_bot_state_dump_task = asyncio.create_task(
+                process_bot_state_dumper.run_periodic_dump_loop(
+                    self.octobot.dump_state_path, self.octobot
+                )
+            )
 
     def run_bot_in_thread(self, coroutine):
         self.init_async_loop()
@@ -99,6 +108,10 @@ async def stop_timeout(timeout):
         if self.tools_task_group:
             self.tools_task_group.cancel()
 
+        if self._process_bot_state_dump_task is not None and not self._process_bot_state_dump_task.done():
+            self._process_bot_state_dump_task.cancel()
+        self._process_bot_state_dump_task = None
+
         # close community session
         if self.octobot.community_handler:
             stop_coroutines.append(self.octobot.community_handler.stop_task())

packages/commons/octobot_commons/dsl_interpreter/__init__.py

@@ -39,6 +39,9 @@
     ExpressionOperator,
     PreComputingCallOperator,
     ReCallableOperatorMixin,
+    SignalableOperatorMixin,
+    OperatorSignal,
+    OperatorSignals,
     ReCallingOperatorResult,
     ReCallingOperatorResultKeys,
     ProcessBoundOperatorMixin,
@@ -75,6 +78,9 @@
     "ExpressionOperator",
     "PreComputingCallOperator",
     "ReCallableOperatorMixin",
+    "SignalableOperatorMixin",
+    "OperatorSignal",
+    "OperatorSignals",
     "ProcessBoundOperatorMixin",
     "is_process_bound",
     "InterpreterDependency",

packages/commons/octobot_commons/dsl_interpreter/operators/__init__.py

@@ -50,6 +50,11 @@
     ReCallingOperatorResult,
     ReCallingOperatorResultKeys,
 )
+from octobot_commons.dsl_interpreter.operators.signalable_operator_mixin import (
+    OperatorSignal,
+    OperatorSignals,
+    SignalableOperatorMixin,
+)
 from octobot_commons.dsl_interpreter.operators.process_bound_operator_mixin import (
     ProcessBoundOperatorMixin,
     is_process_bound,
@@ -69,6 +74,9 @@
     "ReCallableOperatorMixin",
     "ReCallingOperatorResult",
     "ReCallingOperatorResultKeys",
+    "SignalableOperatorMixin",
+    "OperatorSignal",
+    "OperatorSignals",
     "ProcessBoundOperatorMixin",
     "is_process_bound",
 ]

packages/commons/octobot_commons/dsl_interpreter/operators/process_bound_operator_mixin.py

@@ -14,13 +14,15 @@
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
 
+import asyncio
 import pathlib
-import socket
 import subprocess
+import time
 import typing
 
 import octobot_commons.dsl_interpreter.operator as dsl_interpreter_operator
 import octobot_commons.errors as commons_errors
+import octobot_commons.logging as commons_logging
 import octobot_commons.process_util as process_util
 
 
@@ -52,6 +54,37 @@ def request_graceful_stop(
             )
         return process_util.request_graceful_stop_via_sigterm(self.pid, logger=logger)
 
+    async def wait_until_pid_stopped(
+        self,
+        pid: int,
+        *,
+        logger: typing.Optional[typing.Any] = None,
+        timeout_seconds: float,
+        poll_interval: float = 0.2,
+    ) -> None:
+        """Poll until ``pid`` is gone or ``timeout_seconds`` elapses (after e.g. SIGTERM)."""
+        resolved_logger = logger or commons_logging.get_logger(self.__class__.__name__)
+        if pid <= 0:
+            resolved_logger.info(
+                "wait_until_pid_stopped: pid=%s treated as already stopped (non-positive)",
+                pid,
+            )
+            return
+        resolved_logger.info(
+            "wait_until_pid_stopped: waiting for pid=%s to exit (timeout=%ss)",
+            pid,
+            timeout_seconds,
+        )
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            if not process_util.pid_is_running(pid):
+                resolved_logger.info("wait_until_pid_stopped: pid=%s exited", pid)
+                return
+            await asyncio.sleep(poll_interval)
+        raise commons_errors.DSLInterpreterError(
+            f"Timed out after {timeout_seconds}s waiting for pid={pid} to exit."
+        )
+
     def spawn_subprocess(
         self,
         argv: list[str],
@@ -78,42 +111,6 @@ def reject_user_path_segment(path_value: str) -> None:
                 "Invalid path: parent directory segments are not allowed."
             )
 
-    @staticmethod
-    def _tcp_port_is_free(bind_host: str, port: int) -> bool:
-        """True if nothing is currently bound to (host, port) for TCP."""
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            try:
-                sock.bind((bind_host, port))
-            except OSError:
-                return False
-        return True
-
-    @staticmethod
-    def find_first_free_listen_port_after_base(
-        bind_host_for_probe: str,
-        listen_port_base: int,
-        max_offset: int = 256,
-        blocklist: list[int] = None,
-    ) -> int:
-        """
-        First offset where ``listen_port_base + offset`` is TCP-free on ``bind_host_for_probe``
-        (optional: require ``paired_listen_port_base + offset`` free as well, same scan step).
-        Returns ``listen_port``.
-        """
-        for offset_from_base in range(max_offset):
-            listen_port = listen_port_base + offset_from_base
-            if blocklist and listen_port in blocklist:
-                continue
-            if not ProcessBoundOperatorMixin._tcp_port_is_free(
-                bind_host_for_probe, listen_port
-            ):
-                continue
-            return listen_port
-        raise commons_errors.DSLInterpreterError(
-            "No free listen port found in the scanned range."
-        )
-
     @staticmethod
     def bind_address_for_env_and_probe_hosts(
         params: dict,

packages/commons/octobot_commons/dsl_interpreter/operators/re_callable_operator_mixin.py

@@ -13,8 +13,6 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-import contextlib
-import contextvars
 import dataclasses
 import enum
 import time
@@ -73,66 +71,19 @@ def get_script_override(result: typing.Any) -> typing.Optional[str]:
         )
 
     @staticmethod
-    def get_keyword(result: typing.Any) -> typing.Optional[str]:
+    def get_keyword(result: dict[str, typing.Any]) -> typing.Optional[str]:
         """
         Returns the keyword from the re-calling operator result.
         """
         return result[ReCallingOperatorResult.__name__]["keyword"]
 
 
-# Per logical context (e.g. asyncio Task), not process-global: concurrent or nested `interprete`
-# calls do not share this value. Values are operator `get_name()` strings for which
-# `set_execution_stop()` is active; `get_execution_stop()` checks membership for one class only.
-_execution_stop_operator_names: contextvars.ContextVar[frozenset[str]] = contextvars.ContextVar(
-    "re_callable_execution_stop_operator_names",
-    default=frozenset(),
-)
-
-
 class ReCallableOperatorMixin:
     """
     Mixin for re-callable operators.
     """
     LAST_EXECUTION_RESULT_KEY = "last_execution_result"
 
-    @classmethod
-    def get_execution_stop(cls) -> bool:
-        """
-        True when this operator class is the current execution_stop target
-        (see set_execution_stop). Scopes per operator get_name() so one class
-        does not see another's stop mode.
-        """
-        return cls.get_name() in _execution_stop_operator_names.get()
-
-    @classmethod
-    @contextlib.contextmanager
-    def set_execution_stop(cls) -> typing.Iterator[None]:
-        """
-        Context manager: for the duration of the block, get_execution_stop() is
-        True for this class only (other re-callable classes remain False unless
-        also entered in an outer or parallel context).
-        """
-        previous = _execution_stop_operator_names.get()
-        token = _execution_stop_operator_names.set(
-            frozenset(previous | {cls.get_name()})
-        )
-        try:
-            yield
-        finally:
-            _execution_stop_operator_names.reset(token)
-
-    @classmethod
-    def should_run_execution_stop_for_result( # pylint: disable=unused-argument
-        cls, re_calling_result: typing.Optional[dict]
-    ) -> bool:
-        """
-        When draining execution_stop for automation shutdown, return whether this
-        operator should run a stop branch for the given previous re-calling
-        result dict (the same shape as last_execution_result on the action).
-        Default: do nothing (subclasses e.g. run_octobot may override).
-        """
-        return False
-
     @classmethod
     def get_re_callable_parameters(cls) -> list[operator_parameter.OperatorParameter]:
         """