Improvements

    - Measured, improved compatibilty with older Python versions
    - Added debug, error message logging

docs/source/index.rst

@@ -243,15 +243,15 @@ On the other hand, *SpecialPlace* class constructor accepts two extra optional a
 
 .. _producer:
 
-1. ``producer``: It is a callback function which is called to acquire tokens instead of the
-   default ``_process_input_arcs`` (:py:func:`soyutnet.pt_common.PTCommon._process_input_arcs`).
-   This function can be used to produce custom tokens by generating a label and ID. Then, it will
-   be injected into the PT net's flow.
+1. ``producer``: It is a callback function which is called to acquire tokens after the
+   default ``_process_input_arcs`` (:py:func:`soyutnet.pt_common.PTCommon._process_input_arcs`)
+   is called. This function can be used to produce custom tokens by generating a label and ID.
+   Then, it will be injected into the PT net's flow.
 
 .. _consumer:
 
-2. ``consumer``: This callback function is used instead of the default ``_process_output_arcs``
-   function (:py:func:`soyutnet.pt_common.PTCommon._process_output_arcs`) if it is provided.
+2. ``consumer``: This callback function is called before the default ``_process_output_arcs``
+   function (:py:func:`soyutnet.pt_common.PTCommon._process_output_arcs`) is called.
    It can be used as an end point of PT net model. The tokens acquired by this function can be
    redirected to other utilities.
 

pyproject.toml

@@ -7,7 +7,7 @@ name = "soyutnet"
 version = "0.3.1"
 dependencies = [
 ]
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 authors = [
   {name = "Okan Demir", email = "[email protected]"},
 ]

soyutnet/__init__.py

@@ -4,6 +4,7 @@
 import signal
 import functools
 from typing import Any, Type, Coroutine, TextIO
+import logging
 
 from .constants import *
 from .registry import PTRegistry, TokenRegistry
@@ -87,38 +88,80 @@ def __init__(self) -> None:
         self._LOOP_DELAY: float = 0.5
         self.DEBUG_ENABLED: bool = False
         """if set, :py:func:`soyutnet.SoyutNet.DEBUG` will print."""
-        self.VERBOSE_ENABLED: bool = False
+        self._VERBOSE_ENABLED: bool = False
         """if set, :py:func:`soyutnet.SoyutNet.DEBUG_V` will print."""
         self.SLOW_MOTION: bool = False
         """If set, task loops are delayed for :py:attr:`soyutnet.SoyutNet.LOOP_DELAY` seconds"""
         self.FLOAT_DECIMAL_PLACE_FORMAT: int = 6
         """Number of decimal places of floats in debug prints"""
+        self._LOG_FILE: str = ""
+        """Name of the log file"""
+        self._logger: logging.Logger | None = None
+        """Log handler"""
 
-    def _print(
-        self,
-        *args: Any,
-        file: TextIO = sys.stdout,
-        depth: int = 0,
-        separator: str = " ",
-    ) -> None:
+    @property
+    def LOG_FILE(self) -> str:
+        return self._LOG_FILE
+
+    @LOG_FILE.setter
+    def LOG_FILE(self, filename: str | None) -> None:
+        if filename is not None:
+            self._LOG_FILE = filename
+            self._logger = logging.getLogger(self.__class__.__name__)
+            self._logger.addHandler(logging.FileHandler(filename))
+        else:
+            self._logger = None
+
+    @property
+    def VERBOSE_ENABLED(self) -> bool:
+        return self._VERBOSE_ENABLED
+
+    @VERBOSE_ENABLED.setter
+    def VERBOSE_ENABLED(self, enabled: bool) -> None:
+        self._VERBOSE_ENABLED = enabled
+        logging.basicConfig(level=logging.INFO)
+
+    def __sprint(self, *args: Any, depth: int = 0, separator: str = " ") -> str:
+        output = ""
         for a in args:
             match a:
                 case tuple():
-                    file.write("(")
-                    self._print(*a, file=file, depth=depth + 1, separator=", ")
-                    file.write(")")
+                    output += (
+                        "(" + self.__sprint(*a, depth=depth + 1, separator=", ") + ")"
+                    )
                 case list():
-                    file.write("[")
-                    self._print(*a, file=file, depth=depth + 1, separator=", ")
-                    file.write("]" + os.linesep)
+                    output += (
+                        "["
+                        + self.__sprint(*a, depth=depth + 1, separator=", ")
+                        + "]"
+                        + os.linesep
+                    )
                 case float():
-                    file.write(f"{a:.{self.FLOAT_DECIMAL_PLACE_FORMAT}f}")
+                    output += f"{a:.{self.FLOAT_DECIMAL_PLACE_FORMAT}f}"
                 case _:
-                    file.write(str(a))
-            file.write(separator)
+                    output += str(a)
+            output += separator
 
         if depth == 0:
-            file.write(os.linesep)
+            output += os.linesep
+
+        return output
+
+    def __print(self, *args: Any, file: TextIO = sys.stdout, **kwargs: Any) -> None:
+        file.write(self.__sprint(*args, **kwargs))
+
+    def _print(self, *args: Any, file: TextIO = sys.stdout, **kwargs: Any) -> None:
+        if self._logger is not None:
+            self._logger.info(self.__sprint(*args, depth=1, **kwargs))
+        else:
+            self.__print(*args, file=file, **kwargs)
+
+    def _error(self, *args: Any, file: TextIO = sys.stdout, **kwargs: Any) -> None:
+        if self._logger is not None:
+            self._logger.error(self.__sprint(*args, depth=1, **kwargs))
+        else:
+            self.__print("ERR:", file=file, depth=1, **kwargs)
+            self._print(*args, **kwargs)
 
     @property
     def LOOP_DELAY(self) -> float:
@@ -172,15 +215,15 @@ def DEBUG_V(self, *args: Any) -> None:
         """
         Print debug messages when :py:attr:`soyutnet.SoyutNet.VERBOSE_ENABLED`.
         """
-        if self.DEBUG_ENABLED and self.VERBOSE_ENABLED:
-            self._print(f"{self.get_loop_name()}:", *args)
+        if self.DEBUG_ENABLED and self._VERBOSE_ENABLED:
+            self._error(f"{self.get_loop_name()}:", *args)
 
     def ERROR_V(self, *args: Any) -> None:
         """
         Print error messages when :py:attr:`soyutnet.SoyutNet.VERBOSE_ENABLED`.
         """
-        if self.VERBOSE_ENABLED:
-            self._print(f"{self.get_loop_name()}:", *args, file=sys.stderr)
+        if self._VERBOSE_ENABLED:
+            self._error(f"{self.get_loop_name()}:", *args, file=sys.stderr)
 
     def DEBUG(self, *args: Any) -> None:
         """
@@ -193,7 +236,7 @@ def ERROR(self, *args: Any) -> None:
         """
         Print error messages.
         """
-        self._print(f"{self.get_loop_name()}:", *args, file=sys.stderr)
+        self._error(f"{self.get_loop_name()}:", *args, file=sys.stderr)
 
     def Token(self, *args: Any, **kwargs: Any) -> Token:
         kwargs["net"] = self

soyutnet/constants.py

@@ -3,12 +3,18 @@
 import weakref
 from weakref import ReferenceType
 from typing import (
+    Any,
     Tuple,
-    Never,
     Dict,
     TYPE_CHECKING,
 )
 
+try:
+    from typing import Never
+except ImportError:
+    """Never requires python>=3.11"""
+    Never = Any
+
 label_t = int
 """Label type"""
 id_t = int

soyutnet/pt_common.py

@@ -4,7 +4,6 @@
 from typing import (
     Any,
     AsyncGenerator,
-    Self,
     Dict,
     Tuple,
     Generator,
@@ -13,6 +12,12 @@
     TYPE_CHECKING,
 )
 
+try:
+    from typing import Self
+except ImportError:
+    """Self requires python>=3.11"""
+    Self = Any
+
 from .constants import *
 from .token import Token
 from .observer import Observer

tests/simple_example.py

@@ -1,4 +1,5 @@
 import asyncio
+from pathlib import Path
 
 import soyutnet
 from soyutnet import SoyutNet
@@ -35,6 +36,45 @@ def on_comparison_ends(observer):
             soyutnet.terminate()
 
     net = SoyutNet()
+    net.DEBUG_ENABLED = True
+    net.VERBOSE_ENABLED = True
+    net.ERROR(
+        [
+            {
+                0: "ABCD",
+                1: [
+                    1,
+                    2,
+                    3,
+                ],
+            },
+            (0, "ABCD", [1, 3, 3.0, b"ABCD"]),
+        ]
+    )
+    log_file = Path(__file__).resolve().parent / "log.tmp"
+    log_filename = str(log_file)
+    net.LOG_FILE = log_filename
+    assert net.LOG_FILE == log_filename
+    net.SLOW_MOTION = True
+    net.LOOP_DELAY = 0
+    net.ERROR(
+        [
+            {
+                0: "ABCD",
+                1: [
+                    1,
+                    2,
+                    3,
+                ],
+            },
+            (0, "ABCD", [1, 3, 3.0, b"ABCD"]),
+            tuple(),
+            list(),
+            dict(),
+        ]
+    )
+
+    assert net.VERBOSE_ENABLED
 
     o1 = net.ComparativeObserver(
         expected={1: [((GENERIC_LABEL, 1),)] * 5},
@@ -61,6 +101,8 @@ def on_comparison_ends(observer):
     soyutnet.run(reg)
     print("Simulation is terminated.")
 
+    log_file.unlink()
+
     records = reg.get_merged_records()
     for r in records:
         print(r)