core

Observer

Observer(interval=1)

Base class for observers which are used to monitor and record the state of the simulation at specified intervals.

Parameters:

• interval (int, default: 1 ) –

  The interval at which the observer will be called, by default 1.

Attributes:

• interval (int) –

  The interval at which the observer will be called.

Source code in quansino/io/core.py

def __init__(self, interval: int = 1) -> None:
    """Initialize the `Observer` object."""
    self.interval = interval

__call__ abstractmethod

__call__(*args, **kwargs)

Call the observer with the given arguments. This method should be overridden by subclasses to implement specific behavior.

Parameters:

• *args (Any, default: () ) –

  Positional arguments passed to the observer.

• **kwargs (Any, default: {} ) –

  Keyword arguments passed to the observer.

Source code in quansino/io/core.py

@abstractmethod
def __call__(self, *args: Any, **kwargs: Any) -> None:
    """
    Call the observer with the given arguments. This method should be overridden by subclasses to implement specific behavior.

    Parameters
    ----------
    *args : Any
        Positional arguments passed to the observer.
    **kwargs : Any
        Keyword arguments passed to the observer.
    """
    ...

attach_simulation abstractmethod

attach_simulation(*args, **kwargs)

Attach a simulation to the observer. This method should be overridden by subclasses to implement specific attachment behavior.

Parameters:

• *args (Any, default: () ) –

  Positional arguments passed to the observer.

• **kwargs (Any, default: {} ) –

  Keyword arguments passed to the observer.

Source code in quansino/io/core.py

@abstractmethod
def attach_simulation(self, *args: Any, **kwargs: Any) -> None:
    """
    Attach a simulation to the observer. This method should be overridden by subclasses to implement specific attachment behavior.

    Parameters
    ----------
    *args : Any
        Positional arguments passed to the observer.
    **kwargs : Any
        Keyword arguments passed to the observer.
    """
    ...

close abstractmethod

close()

Close the observer and release any resources. This method should be overridden by subclasses to implement specific cleanup behavior.

Source code in quansino/io/core.py

@abstractmethod
def close(self) -> None:
    """
    Close the observer and release any resources. This method should be overridden by subclasses to implement specific cleanup behavior.
    """
    ...

to_dict

to_dict()

Convert the Observer object to a dictionary.

Returns:

• dict[str, Any] –

  A dictionary representation of the Observer object.

Source code in quansino/io/core.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert the `Observer` object to a dictionary.

    Returns
    -------
    dict[str, Any]
        A dictionary representation of the `Observer` object.
    """
    return {"name": self.__class__.__name__, "kwargs": {"interval": self.interval}}

TextObserver

TextObserver(file, interval=1, mode='a', encoding=None)

Bases: Observer

Base class for text-based observers in a simulation. TextObservers are used to write output to a file or stream at specified intervals.

Parameters:

• file (IO | Path | str) –

  The file or stream to write output to. This can be a file object, a string representing a file path, or a Path object.

• interval (int, default: 1 ) –

  The interval at which the observer will be called, by default 1.

• mode (str, default: 'a' ) –

  The mode in which to open the file, by default "a".

• encoding (str | None, default: None ) –

  The encoding to use when opening the file, by default None. If None, default to 'utf-8' for text files stays None for binary files.

Attributes:

• accept_stream (bool) –

  Whether the observer accepts a stream of data.

• file (IO) –

  The file or stream to write output to.

• mode (str) –

  The mode in which to open the file.

• encoding (str | None) –

  The encoding to use when opening the file.

Source code in quansino/io/core.py

def __init__(
    self,
    file: IO | Path | str,
    interval: int = 1,
    mode: str = "a",
    encoding: str | None = None,
) -> None:
    """Initialize the `TextObserver` object."""
    super().__init__(interval)

    self.mode: str = mode
    self.encoding: str | None = encoding or ("utf-8" if "b" not in mode else None)

    self.file = file

file property writable

file

Get the file object associated with the TextObserver.

Returns:

• IO –

  The file object associated with the TextObserver.

__repr__

__repr__()

Return a representation of the TextObserver.

Returns:

• str –

  The representation of the TextObserver.

Source code in quansino/io/core.py

def __repr__(self) -> str:
    """
    Return a representation of the `TextObserver`.

    Returns
    -------
    str
        The representation of the `TextObserver`.
    """
    return f"{self.__class__.__name__}({self.__str__()}, mode={self.mode}, encoding={self.encoding}, interval={self.interval})"

__str__

__str__()

Return a string representation of the TextObserver.

Returns:

• str –

  The string representation of the TextObserver, including the file type and name.

Source code in quansino/io/core.py

def __str__(self) -> str:
    """
    Return a string representation of the `TextObserver`.

    Returns
    -------
    str
        The string representation of the `TextObserver`, including the file type and name.
    """
    if self._file in (sys.stdout, sys.stderr, sys.stdin):
        return f"Stream:{self._file.name}"

    if hasattr(self._file, "name"):
        name = self._file.name

        if (
            not isinstance(name, int)
            and isinstance(name, str)
            and name not in ("", ".")
        ):
            return f"Path:{name}"

    if hasattr(self._file, "__class__"):
        return f"Class:{self._file.__class__.__name__}"

    return "Class:<Unknown>"

close

close()

Close the file if it is not a standard stream.

Source code in quansino/io/core.py

def close(self) -> None:
    """Close the file if it is not a standard stream."""
    if not hasattr(self, "_file"):
        return

    if self._file not in (sys.stdout, sys.stderr, sys.stdin):
        with suppress(OSError, AttributeError, ValueError):
            self._file.close()

            atexit.unregister(self.close)

to_dict

to_dict()

Convert the TextObserver object to a dictionary.

Returns:

• dict[str, Any] –

  A dictionary representation of the TextObserver.

Source code in quansino/io/core.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert the `TextObserver` object to a dictionary.

    Returns
    -------
    dict[str, Any]
        A dictionary representation of the `TextObserver`.
    """
    dictionary = super().to_dict()
    dictionary.setdefault("kwargs", {})

    dictionary["kwargs"]["mode"] = self.mode
    dictionary["kwargs"]["encoding"] = self.encoding

    return dictionary