aspartik.b3.loggers

Classes which record the state of the simulation.

All classes here adhere to the Callback protocol and should be passed as such.

class TreeLogger:

@dataclass(slots=True)
class TreeLogger(Callback):
    """Records the topology of the tree into a `.trees` file."""

    tree: Tree
    path: str
    """
    Path to the file where the trees will be appended in Newick format, one per
    line.  It's opened verbatim (the `.trees` extension won't be added).
    """
    every: int
    """How often the logger will be called"""

    def __post_init__(self):
        self._file = open(self.path, "w")

    def log(self, mcmc: MCMC):
        line = self.tree.newick()
        self._file.write(line)
        self._file.write("\n")

    def __getstate__(self):
        # None ignores __dict__ which contains the file handle
        return (None, self.__slots__)
#

Records the topology of the tree into a .trees file.

tree: aspartik.b3.Tree

#

path: str

#

Path to the file where the trees will be appended in Newick format, one per line. It's opened verbatim (the .trees extension won't be added).

every: int

#

How often the logger will be called

def log(self, mcmc: aspartik.b3.MCMC)

    def log(self, mcmc: MCMC):
        line = self.tree.newick()
        self._file.write(line)
        self._file.write("\n")
#

def call(self, mcmc: aspartik.b3.MCMC) -> None

    def call(self, mcmc: MCMC) -> None:
        """
        A custom operation

        Used by loggers and other periodic actions.
        """
#

A custom operation

Used by loggers and other periodic actions.

class PrintLogger:

@dataclass(slots=True)
class PrintLogger(Callback):
    every: int

    _last_time: Optional[float] = field(init=False, default=None)

    def log(self, mcmc: MCMC):
        if mcmc.current_step == 0:
            print(
                f"{'Step':>10}{'Posterior':>15}{'Likelihood':>15}{'Prior':>15}{'Speed t/m':>15}"
            )

        current_time = time.perf_counter()
        if self._last_time:
            # in seconds
            speed = (current_time - self._last_time) / self.every * 1_000_000
            if speed >= 60:
                speed = f"{speed / 60:.1f}min"
            else:
                speed = f"{speed:.1f}sec"
        else:
            speed = "-"

        likelihood = mcmc.likelihood.likelihood()
        print(
            f"{mcmc.current_step:>10}{mcmc.posterior:>15.2f}{likelihood:>15.2f}{mcmc.prior:>15.2f}{speed:>15}"
        )

        self._last_time = current_time
#

every: int

#

How often this callback should be called

The MCMC will call each callback object when index % every is 0. This value is read once when MCMC is created, so if it's changed during execution, the old every value will continue to be used.

def log(self, mcmc: aspartik.b3.MCMC)

    def log(self, mcmc: MCMC):
        if mcmc.current_step == 0:
            print(
                f"{'Step':>10}{'Posterior':>15}{'Likelihood':>15}{'Prior':>15}{'Speed t/m':>15}"
            )

        current_time = time.perf_counter()
        if self._last_time:
            # in seconds
            speed = (current_time - self._last_time) / self.every * 1_000_000
            if speed >= 60:
                speed = f"{speed / 60:.1f}min"
            else:
                speed = f"{speed:.1f}sec"
        else:
            speed = "-"

        likelihood = mcmc.likelihood.likelihood()
        print(
            f"{mcmc.current_step:>10}{mcmc.posterior:>15.2f}{likelihood:>15.2f}{mcmc.prior:>15.2f}{speed:>15}"
        )

        self._last_time = current_time
#

def call(self, mcmc: aspartik.b3.MCMC) -> None

    def call(self, mcmc: MCMC) -> None:
        """
        A custom operation

        Used by loggers and other periodic actions.
        """
#

A custom operation

Used by loggers and other periodic actions.

class ValueLogger:

@dataclass(slots=True)
class ValueLogger(Callback):
    items: Mapping[str, Any]
    path: str
    every: int

    _file: TextIOBase = field(init=False)

    def __post_init__(self):
        self._file = open(self.path, "w")

    def log(self, mcmc: MCMC):
        entry_json = json.dumps(self.items, default=_serialize)
        self._file.write(entry_json)
        self._file.write("\n")
        self._file.flush()

    def __getstate__(self):
        return (None, self.__slots__)
#

items: collections.abc.Mapping

#

path: str

#

every: int

#

How often this callback should be called

The MCMC will call each callback object when index % every is 0. This value is read once when MCMC is created, so if it's changed during execution, the old every value will continue to be used.

def log(self, mcmc: aspartik.b3.MCMC)

    def log(self, mcmc: MCMC):
        entry_json = json.dumps(self.items, default=_serialize)
        self._file.write(entry_json)
        self._file.write("\n")
        self._file.flush()
#

def call(self, mcmc: aspartik.b3.MCMC) -> None

    def call(self, mcmc: MCMC) -> None:
        """
        A custom operation

        Used by loggers and other periodic actions.
        """
#

A custom operation

Used by loggers and other periodic actions.