Merge branch 'develop' into enhancement/1470-yn-0067-publisher-crashed-plugins

client/ayon_core/tools/console_interpreter/abstract.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import List, Dict, Optional
+from typing import Optional
 
 
 @dataclass
@@ -13,8 +15,8 @@ class TabItem:
 class InterpreterConfig:
     width: Optional[int]
     height: Optional[int]
-    splitter_sizes: List[int] = field(default_factory=list)
-    tabs: List[TabItem] = field(default_factory=list)
+    splitter_sizes: list[int] = field(default_factory=list)
+    tabs: list[TabItem] = field(default_factory=list)
 
 
 class AbstractInterpreterController(ABC):
@@ -27,7 +29,7 @@ class AbstractInterpreterController(ABC):
         self,
         width: int,
         height: int,
-        splitter_sizes: List[int],
-        tabs: List[Dict[str, str]],
-    ):
+        splitter_sizes: list[int],
+        tabs: list[dict[str, str]],
+    ) -> None:
         pass

client/ayon_core/tools/console_interpreter/control.py

@@ -1,4 +1,5 @@
-from typing import List, Dict
+from __future__ import annotations
+from typing import Optional
 
 from ayon_core.lib import JSONSettingRegistry
 from ayon_core.lib.local_settings import get_launcher_local_dir
@@ -11,13 +12,15 @@ from .abstract import (
 
 
 class InterpreterController(AbstractInterpreterController):
-    def __init__(self):
+    def __init__(self, name: Optional[str] = None) -> None:
+        if name is None:
+            name = "python_interpreter_tool"
         self._registry = JSONSettingRegistry(
-            "python_interpreter_tool",
+            name,
             get_launcher_local_dir(),
         )
 
-    def get_config(self):
+    def get_config(self) -> InterpreterConfig:
         width = None
         height = None
         splitter_sizes = []
@@ -54,9 +57,9 @@ class InterpreterController(AbstractInterpreterController):
         self,
         width: int,
         height: int,
-        splitter_sizes: List[int],
-        tabs: List[Dict[str, str]],
-    ):
+        splitter_sizes: list[int],
+        tabs: list[dict[str, str]],
+    ) -> None:
         self._registry.set_item("width", width)
         self._registry.set_item("height", height)
         self._registry.set_item("splitter_sizes", splitter_sizes)

client/ayon_core/tools/console_interpreter/ui/utils.py

@@ -1,42 +1,42 @@
-import os
 import sys
 import collections
 
 
+class _CustomSTD:
+    def __init__(self, orig_std, write_callback):
+        self.orig_std = orig_std
+        self._valid_orig = bool(orig_std)
+        self._write_callback = write_callback
+
+    def __getattr__(self, attr):
+        return getattr(self.orig_std, attr)
+
+    def __setattr__(self, key, value):
+        if key in ("orig_std", "_valid_orig", "_write_callback"):
+            super().__setattr__(key, value)
+        else:
+            setattr(self.orig_std, key, value)
+
+    def write(self, text):
+        if self._valid_orig:
+            self.orig_std.write(text)
+        self._write_callback(text)
+
+
 class StdOEWrap:
     def __init__(self):
-        self._origin_stdout_write = None
-        self._origin_stderr_write = None
-        self._listening = False
         self.lines = collections.deque()
-
-        if not sys.stdout:
-            sys.stdout = open(os.devnull, "w")
-
-        if not sys.stderr:
-            sys.stderr = open(os.devnull, "w")
-
-        if self._origin_stdout_write is None:
-            self._origin_stdout_write = sys.stdout.write
-
-        if self._origin_stderr_write is None:
-            self._origin_stderr_write = sys.stderr.write
-
         self._listening = True
-        sys.stdout.write = self._stdout_listener
-        sys.stderr.write = self._stderr_listener
+
+        self._stdout_wrap = _CustomSTD(sys.stdout, self._listener)
+        self._stderr_wrap = _CustomSTD(sys.stderr, self._listener)
+
+        sys.stdout = self._stdout_wrap
+        sys.stderr = self._stderr_wrap
 
     def stop_listen(self):
         self._listening = False
 
-    def _stdout_listener(self, text):
+    def _listener(self, text):
         if self._listening:
             self.lines.append(text)
-        if self._origin_stdout_write is not None:
-            self._origin_stdout_write(text)
-
-    def _stderr_listener(self, text):
-        if self._listening:
-            self.lines.append(text)
-        if self._origin_stderr_write is not None:
-            self._origin_stderr_write(text)