"""Structured access to ABC's own network statistics.
ABC reports what it thinks of a network through ``print_stats`` (classic store)
and ``&ps`` (GIA store). Both print a single human-readable line; these helpers
run them and parse that line into an :class:`AbcStats`, so a script can compare
what ABC measured against what ``aigverse`` measures without scraping text.
"""
from __future__ import annotations
import re
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
from ._errors import AbcExecutionError
from ._runner import check_gia_supported, check_supported, resolve_binary, run_commands
if TYPE_CHECKING:
import os
from ..networks import Aig
__all__ = ["AbcStats", "collect_stats", "stats"]
_INPUT_FILE = "in.aig"
# ABC colours its statistics line, so the escape sequences come off first.
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
_IO = re.compile(r"i/o\s*=\s*(\d+)\s*/\s*(\d+)")
# `print_stats` calls them latches, `&ps` calls them flops; both mean registers.
_REGISTERS = re.compile(r"(?:lat|ff)\s*=\s*(\d+)")
_AND_GATES = re.compile(r"and\s*=\s*(\d+)")
_LEVELS = re.compile(r"lev\s*=\s*(\d+)")
_AVERAGE_LEVEL = re.compile(r"lev\s*=\s*\d+\s*\(([0-9.]+)\)")
_MEMORY = re.compile(r"mem\s*=\s*([0-9.]+)\s*MB")
[docs]
@dataclass(frozen=True)
class AbcStats:
"""What ABC reports about a network.
ABC's own network name is deliberately absent: the bridge transfers through
a temporary file, so it is always that file's stem and never says anything
about the network. It survives verbatim in :attr:`raw`.
"""
#: Number of primary inputs.
num_pis: int
#: Number of primary outputs.
num_pos: int
#: Number of AND nodes.
num_gates: int
#: Depth in AND levels.
num_levels: int
#: Number of registers, or ``None`` where ABC reported no such field.
#: ``print_stats`` calls them ``lat`` and ``&ps`` calls them ``ff``; both are
#: read into this field, and ``&ps`` omits it entirely for a purely
#: combinational network.
num_registers: int | None = None
#: Mean level over the outputs, reported by ``&ps`` only.
average_level: float | None = None
#: Memory ABC used for the network in megabytes, reported by ``&ps`` only.
memory_mb: float | None = None
#: The unparsed line, so nothing ABC said is lost.
raw: str = ""
def _parse(output: str, *, binary: str, command: str) -> AbcStats:
"""Parse ABC's statistics line.
Args:
output: Everything ABC wrote.
binary: The executable that produced it, for the error message.
command: The command that produced it, for the error message.
Returns:
The parsed statistics.
Raises:
AbcExecutionError: If no statistics line could be found.
"""
for raw_line in output.splitlines():
line = _ANSI.sub("", raw_line).strip()
io = _IO.search(line)
gates = _AND_GATES.search(line)
levels = _LEVELS.search(line)
if not (io and gates and levels):
continue
registers = _REGISTERS.search(line)
average = _AVERAGE_LEVEL.search(line)
memory = _MEMORY.search(line)
return AbcStats(
num_pis=int(io.group(1)),
num_pos=int(io.group(2)),
num_gates=int(gates.group(1)),
num_levels=int(levels.group(1)),
num_registers=int(registers.group(1)) if registers else None,
average_level=float(average.group(1)) if average else None,
memory_mb=float(memory.group(1)) if memory else None,
raw=line,
)
msg = "could not find a statistics line in ABC's output"
raise AbcExecutionError(msg, binary=binary, command=command, output=output)
def collect_stats(
ntk: Aig,
stats_command: str,
*,
gia: bool,
timeout: float | None,
binary: str | os.PathLike[str] | None,
) -> AbcStats:
"""Run a statistics command on a network and parse the result.
Args:
ntk: The network to measure.
stats_command: ABC command printing the statistics.
gia: If ``True``, load the network into the GIA store rather than the
classic one. The read command and the guards follow from it, so the
two cannot disagree about which store is in play.
timeout: Seconds to wait for ABC to terminate, or ``None`` for no limit.
binary: Overrides the resolved ABC executable for this call only.
Returns:
The parsed statistics.
Raises:
ValueError: If the network is loaded into the GIA store and has a register
whose reset value is undefined.
"""
check_supported(ntk)
if gia:
check_gia_supported(ntk)
executable = resolve_binary(binary)
from ..io import write_aiger
read_command = "&read" if gia else "read_aiger"
with tempfile.TemporaryDirectory(prefix="aigverse-abc-") as tmpdir:
directory = Path(tmpdir)
write_aiger(ntk, directory / _INPUT_FILE)
command = f"{read_command} {_INPUT_FILE}; {stats_command}"
output = run_commands(command, timeout=timeout, cwd=directory, binary=executable)
return _parse(output, binary=str(executable), command=command)
[docs]
def stats(
ntk: Aig,
*,
timeout: float | None = None,
binary: str | os.PathLike[str] | None = None,
) -> AbcStats:
"""Reports ABC's ``print_stats`` for a network.
.. warning::
These are ABC's counts, not ``aigverse``'s, and the two can differ for the
very same network. ABC structurally hashes as it reads, so any structural
redundancy the network carried is gone before ``print_stats`` sees it: a
16-bit carry-lookahead adder that ``aigverse`` reports as 186 gates comes
back from here as 182. Use :attr:`~aigverse.networks.Aig.num_gates` to
describe the network you hold, and this to describe what ABC worked on.
Args:
ntk: The network to measure.
timeout: Seconds to wait for ABC to terminate, or ``None`` for no limit.
binary: Overrides the resolved ABC executable for this call only.
Returns:
What ABC reports about the network.
Raises:
TypeError: If ``ntk`` is not an ``Aig``.
AbcNotFoundError: If no ABC executable could be located.
AbcTimeoutError: If ABC did not terminate within ``timeout`` seconds.
AbcExecutionError: If ABC reported an error or printed nothing usable.
"""
return collect_stats(ntk, "print_stats", gia=False, timeout=timeout, binary=binary)