feat: allow empty output workables

Author: artdotlis

src/mpyflow/library/worker.py

@@ -56,12 +56,14 @@ def __init__(
         self,
         name: str,
         workable_in: tuple[Workable[IN, OT], ...],
-        workable_out: tuple[Workable[OT, Any], ...],
+        workable_out: tuple[Workable[OT, Any], ...] = tuple(),
         hub: bool = False,
         print_size: int = 10000,
         /,
     ) -> None:
         super().__init__()
+        if len(workable_in) == 0:
+            raise WorkerEx("At least one workable must be defined as input!")
         self.__name = name
         self.__syo: SyncStdoutInterface | None = None
         self.__sha: RedirectSysHandler | None = None
@@ -229,6 +231,13 @@ async def __write_to_all(self, erg: OT, writer_len: int, /) -> None:
             pointers.remove(local_pointer)
         self.evq.worker_queue.task_done()
 
+    async def __write(self, erg: OT, wpl: list[int], writer_len: int, /) -> None:
+        if len(self.__wm.workable_out) > 0:
+            if self.__hub:
+                await self.__write_to_all(erg, writer_len)
+            else:
+                await self.__write_to_one(erg, wpl, writer_len)
+
     async def __writer(self, p_name: str, /) -> int:
         counter_write = 0
         wpl: list[int] = []
@@ -247,10 +256,8 @@ async def __writer(self, p_name: str, /) -> int:
                 else:
                     counter_write += 1
                     self.__print_con(p_name, f"write_hub_{self.__hub}", counter_write)
-                    if self.__hub:
-                        await self.__write_to_all(erg, writer_len)
-                    else:
-                        await self.__write_to_one(erg, wpl, writer_len)
+                    await self.__write(erg, wpl, writer_len)
+
         except Exception:
             self.evq.stop_event_error.set()
             raise

src/mpyflow/shared/logger/manager.py

@@ -1,7 +1,3 @@
-import atexit
-
-from rich.status import Status
-
 import json
 from io import TextIOWrapper
 
@@ -186,15 +182,11 @@ class _EndConD:
     flush_log: ValueP[int]
 
 
-def _print_at_exit(output_stream: TextIOWrapper | None, sta: Status, /) -> None:
+def _print_at_exit(output_stream: TextIOWrapper | None, /) -> None:
     if output_stream is not None:
         output_stream.write("\n")
         output_stream.flush()
         output_stream.close()
-    try:
-        sta.__exit__(None, None, None)
-    except ValueError:
-        pass
 
 
 def _prog_bar_calc(count: int, prog: float, /) -> str:
@@ -242,99 +234,70 @@ def __rich__(self) -> str:
         return self.__formatter()
 
 
-@final
-class _SysOutPrinter:
-    __slots__ = ("__console", "__log_manager")
-
-    def __init__(self, manager: LogManager, /) -> None:
-        super().__init__()
-        self.__console = Console(log_path=False)
-        sta = self.__console.status(_RichCast(lambda: self.__print_status()))
-        sta.__enter__()
-        atexit.register(lambda: _print_at_exit(sys.__stdout__, sta))
-        self.__log_manager = manager
-
-    def __print_status(self) -> str:
-        uptime_loc = self.__log_manager.wa_process_attributes.started
-        with self.__log_manager.manager_lock:
-            act_obj = len(self.__log_manager.sync_object_manager.managed_dict)
-            if act_obj > 0:
-                mana = self.__log_manager.sync_object_manager.managed_dict
-                gen = (elem[0] for elem in mana.values())
-                prog: float = reduce((lambda x, y: x + y), gen) / (10 * act_obj)
-            else:
-                prog = 0.0
-            msg_st = _format_global_status(
-                act_obj, self.__log_manager.manager_done_counter.value, uptime_loc, prog
-            )
-            return f"[bold green]{msg_st}"
-
-    def print_to_console(self, message: SyncOutMsg, /) -> None:
-        with self.__log_manager.manager_lock:
-            _check_message(message)
-            new_act_object = self.__log_manager.sync_object_manager.managed_dict.get(
-                message.s_id, None
-            )
-            if new_act_object is not None:
-                _check_object_status(message, new_act_object)
-            new_act_object = (message.status, message.done)
-            if message.done:
-                self.__log_manager.manager_done_counter.value += 1
-            self.__log_manager.sync_object_manager.managed_dict[message.s_id] = (
-                new_act_object
-            )
-
-            for msg in message.msg.split("\n"):
-                if msg == "":
-                    continue
-                self.__console.log(msg)
+def _print_status(manager: LogManager, /) -> str:
+    uptime_loc = manager.wa_process_attributes.started
+    with manager.manager_lock:
+        act_obj = len(manager.sync_object_manager.managed_dict)
+        if act_obj > 0:
+            mana = manager.sync_object_manager.managed_dict
+            gen = (elem[0] for elem in mana.values())
+            prog: float = reduce((lambda x, y: x + y), gen) / (10 * act_obj)
+        else:
+            prog = 0.0
+        msg_st = _format_global_status(
+            act_obj, manager.manager_done_counter.value, uptime_loc, prog
+        )
+        return f"[bold green]{msg_st}"
 
 
-def _print_empty(
-    printer: _SysOutPrinter, pid_cur: int | None, manager: LogManager, /
-) -> None:
-    printer.print_to_console(
-        SyncOutMsg(
-            s_id=f"_OutputWaitingProcess_{pid_cur!s}_{manager.id_cnt!s}",
-            msg="",
-            status=0.0,
-            done=False,
-        )
-    )
+def _p2c(manager: LogManager, message: SyncOutMsg, console: Console, /) -> None:
+    with manager.manager_lock:
+        _check_message(message)
+        new_act_object = manager.sync_object_manager.managed_dict.get(message.s_id, None)
+        if new_act_object is not None:
+            _check_object_status(message, new_act_object)
+        new_act_object = (message.status, message.done)
+        if message.done:
+            manager.manager_done_counter.value += 1
+        manager.sync_object_manager.managed_dict[message.s_id] = new_act_object
+        if len(message.msg) > 5:
+            console.log(message.msg)
+            time.sleep(1)
 
 
 def _waiting_output_str(manager: LogManager, /) -> None:
     pid_cur = current_process().pid
     running = True
-    printer = _SysOutPrinter(manager)
-    _print_empty(printer, pid_cur, manager)
-    try:
-        while running:
-            try:
-                erg = manager.get_from_queue()
-            except queue.Empty:
-                _print_empty(printer, pid_cur, manager)
-            else:
-                printer.print_to_console(erg)
-            running = manager.run or not manager.queue_empty
-        out_msg = r"_OutputWaitingProcess \[done]"
-        act_obj = len(manager.sync_object_manager.managed_dict) - 1
-        act_obj -= manager.manager_done_counter.value
-        if act_obj > 0:
-            out_msg = f"{out_msg} unfinished objects: {act_obj}"
-        printer.print_to_console(
-            SyncOutMsg(
-                s_id=f"_OutputWaitingProcess_{pid_cur!s}_{manager.id_cnt!s}",
-                msg=out_msg,
-                status=100.0,
-                done=True,
-            )
-        )
-    except Exception:
-        manager.set_manager_error()
-        raise
-    finally:
-        manager.zero_status()
+    cons = Console(log_path=False)
+    with cons.status(_RichCast(lambda: _print_status(manager))):
+        s_id = f"_OutputWaitingProcess_{pid_cur!s}_{manager.id_cnt!s}"
+        sync = SyncOutMsg(s_id=s_id, msg="", status=0.0, done=False)
+        _p2c(manager, sync, cons)
+        try:
+            while running:
+                try:
+                    erg = manager.get_from_queue()
+                except queue.Empty:
+                    s_id = f"_OutputWaitingProcess_{pid_cur!s}_{manager.id_cnt!s}"
+                    sync = SyncOutMsg(s_id=s_id, msg="", status=0.0, done=False)
+                    _p2c(manager, sync, cons)
+                else:
+                    _p2c(manager, erg, cons)
+                running = manager.run or not manager.queue_empty
+            out_msg = r"_OutputWaitingProcess \[done]"
+            act_obj = len(manager.sync_object_manager.managed_dict) - 1
+            act_obj -= manager.manager_done_counter.value
+            if act_obj > 0:
+                out_msg = f"{out_msg} unfinished objects: {act_obj}"
+            s_id = f"_OutputWaitingProcess_{pid_cur!s}_{manager.id_cnt!s}"
+            sync = SyncOutMsg(s_id=s_id, msg=out_msg, status=100.0, done=True)
+            _p2c(manager, sync, cons)
+        except Exception:
+            manager.set_manager_error()
+            raise
+        finally:
+            manager.zero_status()
+    _print_at_exit(sys.__stdout__)
 
 
 def _logger_process(