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):
    """
    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.
    """
    ...

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>"

attach_simulation

attach_simulation(file_manager)

Attach the simulation to the Observer via a FileManager.

Parameters:

• file_manager (FileManager) –

  The FileManager instance to attach to the observer.

Source code in quansino/io/core.py

def attach_simulation(self, file_manager: FileManager) -> None:
    """
    Attach the simulation to the `Observer` via a `FileManager`.

    Parameters
    ----------
    file_manager : FileManager
        The `FileManager` instance to attach to the observer.
    """
    file_manager.register(self.close)

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