References and Traces

References Summary

delphyne.core.refs

References to nodes, values, spaces and space elements.

References are serializable, immutable values that can be used to identify nodes, spaces and values in a tree (possibly deeply nested). References are useful for tooling and for representing serializable traces (Trace). Also, references are attached to success nodes and query answers (Tracked) so as to allow caching and enforce the locality invariant (see Tree).

Local references identify a node, space or space element relative to a given tree node. Global references are expressed relative to a single, global origin.

References in this module are full references, which are produced by reify. Query answers are stored as strings and elements of spaces induced by strategies are denoted by sequences of value references.

See modules irefs and hrefs for two alternative kinds of references: id-based references and hint-based references.

delphyne.core.irefs

Id-Based References.

These are shorter references, where query answers and success values are associated unique identifiers. This concise format is used for exporting traces (see Trace).

delphyne.core.hrefs

Hint-Based References.

Query answers and success values are identified by sequences of hints. This format is used in the demonstration language (e.g. argument of test instruction go compare(['', 'foo bar'])) and when visualizing traces resulting from demonstrations.

Query Answers

Answer dataclass

An answer to a query.

It can serve as a space element reference if the space in question is a query and the proposed answer correctly parses.

Attributes:

Name	Type	Description
mode	AnswerMode	The answer mode (see AnswerMode).
content	str | Structured	The answer content, which can be a raw string or a structured answer (see Structured).
tool_calls	tuple[ToolCall, ...]	An optional sequence of tool calls.
justification	str | None	Additional explanations for the answers, which are not passed to the parser but can be appended at the end of the answer in examples. In particular, this is useful when defining queries for which the oracle is not asked to produce a justification for its answer, but justifications can still be provided in examples for the sake of few-shot prompting.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Answer:
    """
    An answer to a query.

    It can serve as a _space element reference_ if the space in question
    is a query and the proposed answer correctly parses.

    Attributes:
        mode: The answer mode (see `AnswerMode`).
        content: The answer content, which can be a raw string or a
            structured answer (see `Structured`).
        tool_calls: An optional sequence of tool calls.
        justification: Additional explanations for the answers, which
            are not passed to the parser but can be appended at the end
            of the answer in examples. In particular, this is useful
            when defining queries for which the oracle is not asked to
            produce a justification for its answer, but justifications
            can still be provided in examples for the sake of few-shot
            prompting.
    """

    mode: AnswerMode
    content: str | Structured
    tool_calls: tuple[ToolCall, ...] = ()
    justification: str | None = None

    def digest(self) -> str:
        return _answer_digest(self)

AnswerMode

AnswerMode = str | None

A name for an answer mode, which can be a string or None (the latter is typically used for naming default modes).

Queries are allowed to define multiple answer modes, each mode being possibly associated with different settings and with a different parser. An Answer value features the mode that must be used to parse it.

Structured dataclass

Wrapper for structured LLM answers.

Many LLM APIs allow producing JSON answers (sometimes following a given schema) instead of plain text.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Structured:
    """
    Wrapper for structured LLM answers.

    Many LLM APIs allow producing JSON answers (sometimes following a
    given schema) instead of plain text.
    """

    structured: Any  # JSON object

    def _hashable_repr(self) -> str:
        # See comment in ToolCall._hashable_repr
        import json

        return json.dumps(self.__dict__, sort_keys=True)

    def __hash__(self) -> int:
        return hash(self._hashable_repr())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Structured):
            return NotImplemented
        return self._hashable_repr() == other._hashable_repr()

ToolCall dataclass

A tool call, usually produced by an LLM oracle.

Tool calls can be attached to LLM answers (see Answer).

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class ToolCall:
    """
    A tool call, usually produced by an LLM oracle.

    Tool calls can be attached to LLM answers (see `Answer`).
    """

    name: str
    args: Mapping[str, Any]

    def _hashable_repr(self) -> str:
        # Tool calls need to be hashable since they are part of answers
        # and references. However, they can feature arbitrary JSON
        # objects.
        import json

        return json.dumps(self.__dict__, sort_keys=True)

    def __hash__(self) -> int:
        return hash(self._hashable_repr())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, ToolCall):
            return NotImplemented
        return self._hashable_repr() == other._hashable_repr()

Full References

NodePath dataclass

Encodes a sequence of actions leading to a node with respect to a given root.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class NodePath:
    """
    Encodes a sequence of actions leading to a node with respect to a
    given root.
    """

    actions: tuple[ValueRef, ...]

    def append(self, action: ValueRef) -> "NodePath":
        return NodePath((*self.actions, action))

    def digest(self) -> str:
        body = ", ".join(value_digest(action) for action in self.actions)
        return f"<{body}>"

ValueRef

ValueRef = Assembly[AtomicValueRef]

A reference to a local value, which is obtained by combining elements of (possibly multiple) local spaces.

Assembly

Assembly = T | None | tuple[Assembly[T], ...]

An S-expression whose atoms have type T.

AtomicValueRef

AtomicValueRef = IndexedRef | SpaceElementRef

An atomic value reference is a space element reference that is indexed zero or a finite number of times: space_elt_ref[i1][i2]...[in].

IndexedRef dataclass

Indexing an atomic value reference.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class IndexedRef:
    """
    Indexing an atomic value reference.
    """

    ref: AtomicValueRef
    index: int

    def digest(self) -> str:
        return f"{self.digest()}[{self.index}]"

SpaceElementRef dataclass

A reference to an element of a local space.

Attributes:

Name	Type	Description
space	SpaceRef | None	The space containing the element, or None if this is the top-level main space.
element	Answer | NodePath	The element pointer.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class SpaceElementRef:
    """
    A reference to an element of a local space.

    Attributes:
        space: The space containing the element, or `None` if this is
            the top-level main space.
        element: The element pointer.
    """

    space: SpaceRef | None
    element: Answer | NodePath

    def digest(self) -> str:
        space = (
            _MAIN_SPACE_DEBUG_NAME
            if self.space is None
            else self.space.digest()
        )
        if isinstance(self.element, Answer):
            element = _answer_digest(self.element)
        else:
            element = self.element.digest()
        return f"{space}{{{element}}}"

SpaceRef dataclass

A reference to a specific local space.

The arg argument should be () for nonparametric spaces and a n-uple for spaces parametric in n arguments. This differs from Orakell where all parametric spaces have one argument.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class SpaceRef:
    """
    A reference to a specific local space.

    The `arg` argument should be `()` for nonparametric spaces and a
    n-uple for spaces parametric in n arguments. This differs from
    Orakell where all parametric spaces have one argument.
    """

    name: SpaceName
    args: tuple[ValueRef, ...]

    def digest(self) -> str:
        if not self.args:
            return str(self.name)
        args_str = ", ".join(value_digest(a) for a in self.args)
        return f"{self.name}({args_str})"

SpaceName dataclass

A name identifying a parametric space.

This name can feature integer indices. For example, subs[0] denotes the first subgoal of a Join node.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class SpaceName:
    """
    A name identifying a parametric space.

    This name can feature integer indices. For example, `subs[0]`
    denotes the first subgoal of a `Join` node.
    """

    name: str
    indices: tuple[int, ...]

    def __getitem__(self, index: int) -> "SpaceName":
        return SpaceName(self.name, (*self.indices, index))

    def __str__(self) -> str:
        ret = self.name
        for i in self.indices:
            ret += f"[{i}]"
        return ret

GlobalNodeRef dataclass

Global reference to a node.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class GlobalNodeRef:
    """
    Global reference to a node.
    """

    space: GlobalSpacePath
    path: NodePath

    def child(self, action: ValueRef) -> "GlobalNodeRef":
        return GlobalNodeRef(self.space, self.path.append(action))

    def nested_space(self, space: SpaceRef) -> "GlobalSpacePath":
        return GlobalSpacePath((*self.space.steps, (self.path, space)))

    def nested_tree(self, space: SpaceRef) -> "GlobalNodeRef":
        return GlobalNodeRef(self.nested_space(space), NodePath(()))

    def digest(self) -> str:
        return f"{self.space.digest()} / {self.path.digest()}"

GlobalSpacePath dataclass

A path to a global space, which alternates between following a path from a local root and entering a space within the reached node.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class GlobalSpacePath:
    """
    A path to a global space, which alternates between following a path from
    a local root and entering a space within the reached node.
    """

    steps: tuple[tuple[NodePath, SpaceRef], ...]

    def append(self, path: NodePath, space: SpaceRef) -> "GlobalSpacePath":
        return GlobalSpacePath((*self.steps, (path, space)))

    def split(self) -> "tuple[GlobalNodeRef, SpaceRef] | tuple[None, None]":
        if not self.steps:
            return (None, None)
        last_path, last_space = self.steps[-1]
        parent_steps = self.steps[:-1]
        gsref = GlobalNodeRef(GlobalSpacePath(parent_steps), last_path)
        return (gsref, last_space)

    def parent_node(self) -> "GlobalNodeRef | None":
        return self.split()[0]

    def local_ref(self) -> "SpaceRef | None":
        return self.split()[1]

    def digest(self) -> str:
        elts = [
            f"{path.digest()} / {space.digest()}" for path, space in self.steps
        ]
        return " / ".join([_MAIN_SPACE_DEBUG_NAME, *elts])

Tracked Values

Tracked dataclass

Bases: Generic[T]

A tracked value, which pairs a value with a reference.

Attributes:

Name	Type	Description
value	T	The value being tracked.
ref	AtomicValueRef	A global reference to the space that the value belongs to.
node	GlobalNodeRef | None	A reference to the node that the value is local to, or None if the value is a top-level result.
type_annot	TypeAnnot[T] | NoTypeInfo	An optional type annotation for the value field. This is mostly used for improving the rendering of values when exporting trace information for external tools.

Tracked sequences (or pairs) can be indexed using __getitem__, resulting in tracked values with IndexedRef references. Since __getitem__ is defined, tracked values are also iterable.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Tracked(Generic[T]):
    """
    A tracked value, which pairs a value with a reference.

    Attributes:
        value: The value being tracked.
        ref: A global reference to the space that the value belongs to.
        node: A reference to the node that the value is local to, or
            `None` if the value is a top-level result.
        type_annot: An optional type annotation for the `value` field.
            This is mostly used for improving the rendering of values
            when exporting trace information for external tools.

    Tracked sequences (or pairs) can be indexed using `__getitem__`,
    resulting in tracked values with `IndexedRef` references. Since
    `__getitem__` is defined, tracked values are also iterable.
    """

    value: T
    ref: AtomicValueRef
    node: GlobalNodeRef | None
    type_annot: TypeAnnot[T] | NoTypeInfo

    @overload
    def __getitem__[A, B](
        self: "Tracked[tuple[A, B]]", index: Literal[0]
    ) -> "Tracked[A]": ...

    @overload
    def __getitem__[A, B](
        self: "Tracked[tuple[A, B]]", index: Literal[1]
    ) -> "Tracked[B]": ...

    @overload
    def __getitem__[U](
        self: "Tracked[Sequence[U]]", index: int
    ) -> "Tracked[U]": ...

    def __getitem__[U](
        self: "Tracked[Sequence[U]] | Tracked[tuple[Any, ...]]", index: int
    ) -> "Tracked[U | Any]":
        return Tracked(
            self.value[index],
            IndexedRef(self.ref, index),
            self.node,
            # TODO: will not work well for union of tuples for example
            insp.element_type_of_sequence_type(self.type_annot, index),
        )

Value

Value = ExtAssembly[Tracked[Any]]

An assembly of local, tracked values.

Values can serve as actions or space parameters.

ExtAssembly

ExtAssembly = T | None | Sequence[ExtAssembly[T]]

Generalizing Assembly to allow arbitrary sequences (and not just tuples). The distinction is important because ValueRef needs to be hashable and so cannot contain lists, while Value can contain lists.

check_local_value

check_local_value(val: Value, node: GlobalNodeRef | None)

Raise a LocalityError exception if a given value is not a local value relative to a given node.

Source code in src/delphyne/core/refs.py

def check_local_value(val: Value, node: GlobalNodeRef | None):
    """
    Raise a `LocalityError` exception if a given value is not a local
    value relative to a given node.
    """
    match val:
        case None:
            pass
        case Sequence():
            for v in val:
                check_local_value(v, node)
        case Tracked():
            if val.node != node:
                raise LocalityError(
                    expected_node_ref=node,
                    node_ref=val.node,
                    local_ref=val.ref,
                )
        case _:
            assert False

LocalityError dataclass

Bases: Exception

Exception raised when the locality invariant is violated.

See Tree and check_local_value.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class LocalityError(Exception):
    """
    Exception raised when the locality invariant is violated.

    See `Tree` and `check_local_value`.
    """

    expected_node_ref: GlobalNodeRef | None
    node_ref: GlobalNodeRef | None
    local_ref: AtomicValueRef

Id-Based References

AnswerId dataclass

The identifier to an Answer object stored within a trace.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class AnswerId:
    """
    The identifier to an `Answer` object stored within a trace.
    """

    id: int

    def __str__(self) -> str:
        return f"@{self.id}"

NodeId dataclass

Global identifier of a node within a trace.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class NodeId:
    """
    Global identifier of a node within a trace.
    """

    id: int

    def __str__(self) -> str:
        return f"%{self.id}"

ValueRef

ValueRef = Assembly[AtomicValueRef]

AtomicValueRef

AtomicValueRef = IndexedRef | SpaceElementRef

An atomic value reference is a space element reference that is indexed zero or a finite number of times: space_elt_ref[i1][i2]...[in].

IndexedRef dataclass

Indexing an atomic value reference.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class IndexedRef:
    """
    Indexing an atomic value reference.
    """

    ref: AtomicValueRef
    index: int

    def __str__(self) -> str:
        return f"{self.ref}[{self.index}]"

SpaceRef dataclass

A reference to a specific local space.

The arg argument should be () for nonparametric spaces and a n-uple for spaces parametric in n arguments. This differs from Orakell where all parametric spaces have one argument.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class SpaceRef:
    """
    A reference to a specific local space.

    The `arg` argument should be `()` for nonparametric spaces and a
    n-uple for spaces parametric in n arguments. This differs from
    Orakell where all parametric spaces have one argument.
    """

    name: SpaceName
    args: tuple[ValueRef, ...]

    def __str__(self) -> str:
        name = str(self.name)
        if not self.args:
            return name
        args_str = ", ".join(show_value_ref(a) for a in self.args)
        return f"{name}({args_str})"

SpaceElementRef dataclass

A reference to an element of a local space.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class SpaceElementRef:
    """
    A reference to an element of a local space.
    """

    space: SpaceId
    element: AnswerId | NodeId

    def __str__(self) -> str:
        return f"{self.space}{{{self.element}}}"

NodeOrigin

NodeOrigin = ChildOf | NestedIn

Origin of a tree.

A tree is either the child of another tree or the root of a nested tree. Traces can be exported as mappings from node identifiers to node origin information featuring id-based references (see Trace).

ChildOf dataclass

The tree of interest is the child of another one.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class ChildOf:
    """
    The tree of interest is the child of another one.
    """

    node: NodeId
    action: ValueRef

    def __str__(self) -> str:
        return f"child({self.node}, {show_value_ref(self.action)})"

NestedIn dataclass

The tree of interest is the root of a tree that induces a given space.

Source code in src/delphyne/core/irefs.py

@dataclass(frozen=True)
class NestedIn:
    """
    The tree of interest is the root of a tree that induces a given
    space.
    """

    space: SpaceId

    def __str__(self) -> str:
        return f"nested({self.space})"

Hint-Based References

Hint dataclass

A hint for selecting a query answer.

A hint can be associated to a qualifier, which is the name of an imported demonstration defining the hint.

Source code in src/delphyne/core/hrefs.py

@dataclass(frozen=True)
class Hint:
    """A hint for selecting a query answer.

    A hint can be associated to a qualifier, which is the name of an
    imported demonstration defining the hint.
    """

    qualifier: str | None
    hint: HintValue

    def __str__(self) -> str:
        if not self.qualifier:
            return self.hint
        return f"{self.qualifier}:{self.hint}"

HintValue

HintValue = str

A string that hints at a query answer.

ValueRef

ValueRef = Assembly[AtomicValueRef]

AtomicValueRef

AtomicValueRef = IndexedRef | SpaceElementRef

An atomic value reference is a space element reference that is indexed zero or a finite number of times: space_elt_ref[i1][i2]...[in].

IndexedRef dataclass

Indexing an atomic value reference.

Source code in src/delphyne/core/hrefs.py

@dataclass(frozen=True)
class IndexedRef:
    """
    Indexing an atomic value reference.
    """

    ref: AtomicValueRef
    index: int

    def __str__(self) -> str:
        return f"{self.ref}[{self.index}]"

SpaceRef dataclass

A reference to a specific local space.

The arg argument should be () for nonparametric spaces and a n-uple for spaces parametric in n arguments. This differs from Orakell where all parametric spaces have one argument.

Source code in src/delphyne/core/hrefs.py

@dataclass(frozen=True)
class SpaceRef:
    """
    A reference to a specific local space.

    The `arg` argument should be `()` for nonparametric spaces and a
    n-uple for spaces parametric in n arguments. This differs from
    Orakell where all parametric spaces have one argument.
    """

    name: SpaceName
    args: tuple[ValueRef, ...]

    def __str__(self) -> str:
        name = str(self.name)
        if not self.args:
            return name
        args_str = ", ".join(show_value_ref(a) for a in self.args)
        return f"{name}({args_str})"

SpaceElementRef dataclass

A reference to an element of a local space.

When the space field is None, the primary field is considered instead (if it exists).

Source code in src/delphyne/core/hrefs.py

@dataclass(frozen=True)
class SpaceElementRef:
    """
    A reference to an element of a local space.

    When the `space` field is `None`, the primary field is considered
    instead (if it exists).
    """

    space: SpaceRef | None
    element: tuple[Hint, ...]

    def __str__(self) -> str:
        hints = "'" + " ".join(str(h) for h in self.element) + "'"
        if self.space is None:
            return hints
        else:
            return f"{self.space}{{{hints}}}"

Traces

Trace

A collection of reachable nodes and spaces, which is encoded in a concise way by introducing unique identifiers for answers and nodes.

Traces are mutable. Methods are provided to convert full references into id-based references, creating fresh identifiers for new nodes and queries on the fly. Backward conversion methods are also provided for converting id-based references back into full references (assuming id-based references are valid, without which these methods fail with assertion errors).

Attributes:

Name	Type	Description
nodes	dict[NodeId, NodeOrigin]	a mapping from node identifiers to their origin.
node_ids	dict[NodeOrigin, NodeId]	reverse map of nodes.
answers	dict[AnswerId, LocatedAnswer]	a mapping from answer identifiers to actual answers, along with origin information on the associated query.
answer_ids	dict[LocatedAnswer, AnswerId]	reverse map of answers.
spaces	dict[SpaceId, SpaceOrigin]	a mapping from space identifiers to actual space definitions
space_ids	dict[SpaceOrigin, SpaceId]	reverse map of spaces.

Note

answer_ids can be nonempty while answers is empty, since one must be able to include unanswered queries in the trace.

Source code in src/delphyne/core/traces.py

class Trace:
    """
    A collection of reachable nodes and spaces, which is encoded in a
    concise way by introducing unique identifiers for answers and nodes.

    Traces are mutable. Methods are provided to convert full references
    into id-based references, creating fresh identifiers for new nodes
    and queries on the fly. Backward conversion methods are also
    provided for converting id-based references back into full
    references (assuming id-based references are valid, without which
    these methods fail with assertion errors).

    Attributes:
        nodes: a mapping from node identifiers to their origin.
        node_ids: reverse map of `nodes`.
        answers: a mapping from answer identifiers to actual answers,
            along with origin information on the associated query.
        answer_ids: reverse map of `answers`.
        spaces: a mapping from space identifiers to actual space
            definitions
        space_ids: reverse map of `spaces`.

    !!! note
        `answer_ids` can be nonempty while `answers` is empty, since
        one must be able to include unanswered queries in the trace.
    """

    MAIN_SPACE_ID = irefs.SpaceId(0)

    def __init__(self):
        """
        Create an empty trace.
        """
        self.nodes: dict[irefs.NodeId, irefs.NodeOrigin] = {}
        self.node_ids: dict[irefs.NodeOrigin, irefs.NodeId] = {}
        self.answers: dict[irefs.AnswerId, irefs.LocatedAnswer] = {}
        self.answer_ids: dict[irefs.LocatedAnswer, irefs.AnswerId] = {}
        self.spaces: dict[irefs.SpaceId, irefs.SpaceOrigin] = {}
        self.space_ids: dict[irefs.SpaceOrigin, irefs.SpaceId] = {}
        self._last_node_id: int = 0
        self._last_answer_id: int = 0
        self._last_space_id: int = 0

        self.spaces[Trace.MAIN_SPACE_ID] = irefs.MainSpace()
        self.space_ids[irefs.MainSpace()] = Trace.MAIN_SPACE_ID

    @staticmethod
    def load(trace: ExportableTrace) -> "Trace":
        """
        Load a trace from an exportable representation.
        """
        ret = Trace()
        for id, origin_str in trace.nodes.items():
            origin = parse.node_origin(origin_str)
            node_id = irefs.NodeId(id)
            ret.nodes[node_id] = origin
            ret.node_ids[origin] = node_id
        for id, origin_str in trace.spaces.items():
            origin = parse.space_origin(origin_str)
            space_id = irefs.SpaceId(id)
            ret.spaces[space_id] = origin
            ret.space_ids[origin] = space_id
        for id, located_s in trace.answers.items():
            answer_id = irefs.AnswerId(id)
            space_id = irefs.SpaceId(located_s.space)
            located = irefs.LocatedAnswer(space_id, located_s.answer)
            ret.answers[answer_id] = located
            ret.answer_ids[located] = answer_id
        ret._last_node_id = max((id.id for id in ret.nodes), default=0)
        ret._last_space_id = max((id.id for id in ret.spaces), default=0)
        ret._last_answer_id = max((id.id for id in ret.answers), default=0)
        return ret

    def fresh_or_cached_node_id(
        self, origin: irefs.NodeOrigin
    ) -> irefs.NodeId:
        """
        Obtain the identifier of a node described by its origin.
        Create a new identifier on the fly if it does not exist yet.
        """
        if origin in self.node_ids:
            return self.node_ids[origin]
        else:
            self._last_node_id += 1
            id = irefs.NodeId(self._last_node_id)
            self.nodes[id] = origin
            self.node_ids[origin] = id
            return id

    def fresh_or_cached_space_id(
        self, origin: irefs.SpaceOrigin
    ) -> irefs.SpaceId:
        """
        Obtain the identifier of a space, given its origin. Create a new,
        fresh identifier on the fly if it does not exist yet.
        """
        if origin in self.space_ids:
            return self.space_ids[origin]
        else:
            self._last_space_id += 1
            id = irefs.SpaceId(self._last_space_id)
            self.spaces[id] = origin
            self.space_ids[origin] = id
            return id

    def fresh_or_cached_answer_id(
        self, answer: irefs.LocatedAnswer
    ) -> irefs.AnswerId:
        """
        Obtain the identifier of an answer, given its content and the
        origin of the query that it corresponds to. Create a new, fresh
        identifier on the fly if it does not exist yet.
        """
        if answer in self.answer_ids:
            return self.answer_ids[answer]
        else:
            self._last_answer_id += 1
            id = irefs.AnswerId(self._last_answer_id)
            self.answers[id] = answer
            self.answer_ids[answer] = id
            return id

    def export(self) -> ExportableTrace:
        """
        Export a trace into a lightweight, serializable format.
        """
        nodes = {id.id: str(origin) for id, origin in self.nodes.items()}
        spaces = {id.id: str(origin) for id, origin in self.spaces.items()}
        answers = {
            id.id: ExportableLocatedAnswer(located.space.id, located.answer)
            for id, located in self.answers.items()
        }
        return ExportableTrace(nodes, spaces, answers)

    def check_consistency(self) -> None:
        """
        Perform a sanity check on the trace.

        Each node identifier is expanded into a full reference and then
        converted back to an identifier, which must be equal to the
        original one.
        """
        for id in self.nodes:
            expanded = self.expand_node_id(id)
            id_bis = self.convert_global_node_ref(expanded)
            assert id == id_bis

    def check_roundabout_consistency(self) -> None:
        """
        Perform a sanity check, before and after serializing and
        desarializing it.
        """
        self.check_consistency()
        exportable = self.export()
        copy = Trace.load(exportable)
        copy.check_consistency()
        exportable_copy = copy.export()
        if exportable != exportable_copy:
            print("Original exportable trace:")
            print(exportable)
            print("Exportable trace after round-trip:")
            print(exportable_copy)
            assert False

    ### Convert full references into id-based references

    def convert_global_space_path(
        self, ref: refs.GlobalSpacePath
    ) -> irefs.SpaceId:
        """
        Convert a full, global space reference denoting a quey origin
        into an id-based reference.
        """
        id = Trace.MAIN_SPACE_ID
        for path, space in ref.steps:
            nid = self.fresh_or_cached_node_id(irefs.NestedIn(id))
            nid = self._convert_node_path(nid, path)
            id = self._convert_space_ref(nid, space)
        return id

    def convert_global_node_ref(
        self, path: refs.GlobalNodeRef
    ) -> irefs.NodeId:
        """
        Convert a full, global node reference into an id-based one.
        """
        space_id = self.convert_global_space_path(path.space)
        root = self.fresh_or_cached_node_id(irefs.NestedIn(space_id))
        return self._convert_node_path(root, path.path)

    def convert_answer_ref(self, ref: refs.GlobalAnswerRef) -> irefs.AnswerId:
        """
        Convert a full answer reference into an answer id.
        """
        space = self.convert_global_space_path(ref[0])
        located = irefs.LocatedAnswer(space, ref[1])
        return self.fresh_or_cached_answer_id(located)

    def convert_location(self, location: Location) -> ShortLocation:
        """
        Convert a full location into an id-based one.
        """
        match location:
            case None:
                return None
            case refs.GlobalNodeRef():
                return self.convert_global_node_ref(location)
            case refs.GlobalSpacePath():
                return self.convert_global_space_path(location)

    def _convert_space_ref(
        self, node: irefs.NodeId, ref: refs.SpaceRef
    ) -> irefs.SpaceId:
        """
        Convert a full local space reference into an id-based one, relative
        to a given node.
        """
        args = tuple(self._convert_value_ref(node, a) for a in ref.args)
        space_ref = irefs.SpaceRef(ref.name, args)
        return self.fresh_or_cached_space_id(irefs.LocalSpace(node, space_ref))

    def _convert_node_path(
        self, node: irefs.NodeId, path: refs.NodePath
    ) -> irefs.NodeId:
        """
        Convert a full local node path into an identifier, relative to a
        given node.
        """
        for a in path.actions:
            action_ref = self._convert_value_ref(node, a)
            node = self.fresh_or_cached_node_id(
                irefs.ChildOf(node, action_ref)
            )
        return node

    def _convert_atomic_value_ref(
        self, node: irefs.NodeId, ref: refs.AtomicValueRef
    ) -> irefs.AtomicValueRef:
        """
        Convert a full local atomic value reference into an id-based one,
        relative to a given node.
        """
        if isinstance(ref, refs.IndexedRef):
            return irefs.IndexedRef(
                self._convert_atomic_value_ref(node, ref.ref), ref.index
            )
        else:
            return self.convert_space_element_ref(node, ref)

    def _convert_value_ref(
        self, node: irefs.NodeId, ref: refs.ValueRef
    ) -> irefs.ValueRef:
        """
        Convert a full local value reference into an id-based one,
        relative to a given node.
        """
        if ref is None:
            return None
        elif isinstance(ref, tuple):
            return tuple(self._convert_value_ref(node, a) for a in ref)
        else:
            return self._convert_atomic_value_ref(node, ref)

    def convert_space_element_ref(
        self, node: irefs.NodeId, ref: refs.SpaceElementRef
    ) -> irefs.SpaceElementRef:
        """
        Convert a full local space element reference into an id-based one,
        relative to a given node.
        """
        # We leverage locality to speed up the computation.
        # The following would work but be much slower:
        #     space_id = self.convert_global_space_path(ref.space)

        # The space is attached to a node with an identifier.
        assert ref.space is not None
        space_id = self._convert_space_ref(node, ref.space)
        match ref.element:
            case refs.Answer():
                located = irefs.LocatedAnswer(space_id, ref.element)
                element = self.fresh_or_cached_answer_id(located)
            case refs.NodePath():
                nested_root_orig = irefs.NestedIn(space_id)
                nested_root = self.fresh_or_cached_node_id(nested_root_orig)
                element = self._convert_node_path(nested_root, ref.element)
        return irefs.SpaceElementRef(space_id, element)

    ### Reverse direction: expanding id-based references into full ones.

    def expand_global_space_id(
        self, id: irefs.SpaceId
    ) -> refs.GlobalSpacePath:
        rev_steps: list[tuple[refs.NodePath, refs.SpaceRef]] = []
        origin = self.spaces[id]
        while not isinstance(origin, irefs.MainSpace):
            space_ref = self.expand_space_ref(origin.space)
            id, path = self._recover_path(origin.node)
            rev_steps.append((path, space_ref))
            origin = self.spaces[id]
        return refs.GlobalSpacePath(tuple(reversed(rev_steps)))

    def _recover_path(
        self, dst: irefs.NodeId
    ) -> tuple[irefs.SpaceId, refs.NodePath]:
        """
        Find the space from which the tree containing `dst` originates,
        along with the path from the root of that tree to `dst`.
        """
        rev_path: list[refs.ValueRef] = []
        while True:
            dst_origin = self.nodes[dst]
            match dst_origin:
                case irefs.ChildOf(before, action):
                    rev_path.append(self.expand_value_ref(action))
                    dst = before
                case irefs.NestedIn(space_id):
                    path = refs.NodePath(tuple(reversed(rev_path)))
                    return (space_id, path)

    def _expand_located_answer_ref(
        self, ans: irefs.LocatedAnswer
    ) -> refs.GlobalAnswerRef:
        space = self.expand_global_space_id(ans.space)
        return (space, ans.answer)

    def expand_answer_id(self, ans: irefs.AnswerId) -> refs.GlobalAnswerRef:
        located = self.answers[ans]
        return self._expand_located_answer_ref(located)

    def expand_node_id(self, id: irefs.NodeId) -> refs.GlobalNodeRef:
        """
        Convert a node identifier into a full, global node reference.
        """
        orig, path = self._recover_path(id)
        space = self.expand_global_space_id(orig)
        return refs.GlobalNodeRef(space, path)

    def expand_space_ref(self, ref: irefs.SpaceRef) -> refs.SpaceRef:
        """
        Convert a local id-based space reference into a full one,
        relative to a given node.
        """
        args = tuple(self.expand_value_ref(a) for a in ref.args)
        return refs.SpaceRef(ref.name, args)

    def expand_value_ref(self, ref: irefs.ValueRef) -> refs.ValueRef:
        """
        Convert a local id-based value reference into a full one,
        relative to a given node.
        """
        if ref is None:
            return None
        elif isinstance(ref, tuple):
            return tuple(self.expand_value_ref(a) for a in ref)
        else:
            return self._expand_atomic_value_ref(ref)

    def _expand_atomic_value_ref(
        self, ref: irefs.AtomicValueRef
    ) -> refs.AtomicValueRef:
        """
        Convert a local id-based atomic value reference into a full one,
        relative to a given node.
        """
        if isinstance(ref, irefs.IndexedRef):
            return refs.IndexedRef(
                self._expand_atomic_value_ref(ref.ref), ref.index
            )
        else:
            return self._expand_space_element_ref(ref)

    def _expand_space_element_ref(
        self, ref: irefs.SpaceElementRef
    ) -> refs.SpaceElementRef:
        """
        Convert a local id-based space element reference into a full
        one, relative to a given node.
        """
        # The following would work but be terribly inefficient:
        #    space = self.expand_global_space_id(ref.space)
        space_def = self.spaces[ref.space]
        assert not isinstance(space_def, irefs.MainSpace)
        local_space = self.expand_space_ref(space_def.space)
        match ref.element:
            case irefs.AnswerId():
                located = self.answers[ref.element]
                element = located.answer
            case irefs.NodeId():
                space_id, element = self._recover_path(ref.element)
                assert space_id == ref.space
        return refs.SpaceElementRef(local_space, element)

    ### Extracting local space elements

    def space_elements_in_value_ref(
        self, ref: irefs.ValueRef
    ) -> Iterable[irefs.SpaceElementRef]:
        """
        Enumerate all local space elements that are used to define a
        value.

        Duplicate values can be returned.
        """

        if ref is None:
            pass
        elif isinstance(ref, tuple):
            for r in ref:
                yield from self.space_elements_in_value_ref(r)
        else:
            yield from self._space_elements_in_atomic_value_ref(ref)

    def _space_elements_in_atomic_value_ref(
        self,
        ref: irefs.AtomicValueRef,
    ) -> Iterable[irefs.SpaceElementRef]:
        if isinstance(ref, irefs.IndexedRef):
            yield from self._space_elements_in_atomic_value_ref(ref.ref)
        else:
            yield ref
            space_def = self.spaces[ref.space]
            if isinstance(space_def, irefs.LocalSpace):
                yield from self._space_elements_in_space_ref(space_def.space)

    def _space_elements_in_space_ref(
        self, ref: irefs.SpaceRef
    ) -> Iterable[irefs.SpaceElementRef]:
        for a in ref.args:
            yield from self.space_elements_in_value_ref(a)

__init__

__init__()

Create an empty trace.

Source code in src/delphyne/core/traces.py

def __init__(self):
    """
    Create an empty trace.
    """
    self.nodes: dict[irefs.NodeId, irefs.NodeOrigin] = {}
    self.node_ids: dict[irefs.NodeOrigin, irefs.NodeId] = {}
    self.answers: dict[irefs.AnswerId, irefs.LocatedAnswer] = {}
    self.answer_ids: dict[irefs.LocatedAnswer, irefs.AnswerId] = {}
    self.spaces: dict[irefs.SpaceId, irefs.SpaceOrigin] = {}
    self.space_ids: dict[irefs.SpaceOrigin, irefs.SpaceId] = {}
    self._last_node_id: int = 0
    self._last_answer_id: int = 0
    self._last_space_id: int = 0

    self.spaces[Trace.MAIN_SPACE_ID] = irefs.MainSpace()
    self.space_ids[irefs.MainSpace()] = Trace.MAIN_SPACE_ID

load staticmethod

load(trace: ExportableTrace) -> Trace

Load a trace from an exportable representation.

Source code in src/delphyne/core/traces.py

@staticmethod
def load(trace: ExportableTrace) -> "Trace":
    """
    Load a trace from an exportable representation.
    """
    ret = Trace()
    for id, origin_str in trace.nodes.items():
        origin = parse.node_origin(origin_str)
        node_id = irefs.NodeId(id)
        ret.nodes[node_id] = origin
        ret.node_ids[origin] = node_id
    for id, origin_str in trace.spaces.items():
        origin = parse.space_origin(origin_str)
        space_id = irefs.SpaceId(id)
        ret.spaces[space_id] = origin
        ret.space_ids[origin] = space_id
    for id, located_s in trace.answers.items():
        answer_id = irefs.AnswerId(id)
        space_id = irefs.SpaceId(located_s.space)
        located = irefs.LocatedAnswer(space_id, located_s.answer)
        ret.answers[answer_id] = located
        ret.answer_ids[located] = answer_id
    ret._last_node_id = max((id.id for id in ret.nodes), default=0)
    ret._last_space_id = max((id.id for id in ret.spaces), default=0)
    ret._last_answer_id = max((id.id for id in ret.answers), default=0)
    return ret

fresh_or_cached_node_id

fresh_or_cached_node_id(origin: NodeOrigin) -> NodeId

Obtain the identifier of a node described by its origin. Create a new identifier on the fly if it does not exist yet.

Source code in src/delphyne/core/traces.py

def fresh_or_cached_node_id(
    self, origin: irefs.NodeOrigin
) -> irefs.NodeId:
    """
    Obtain the identifier of a node described by its origin.
    Create a new identifier on the fly if it does not exist yet.
    """
    if origin in self.node_ids:
        return self.node_ids[origin]
    else:
        self._last_node_id += 1
        id = irefs.NodeId(self._last_node_id)
        self.nodes[id] = origin
        self.node_ids[origin] = id
        return id

fresh_or_cached_space_id

fresh_or_cached_space_id(origin: SpaceOrigin) -> SpaceId

Obtain the identifier of a space, given its origin. Create a new, fresh identifier on the fly if it does not exist yet.

Source code in src/delphyne/core/traces.py

def fresh_or_cached_space_id(
    self, origin: irefs.SpaceOrigin
) -> irefs.SpaceId:
    """
    Obtain the identifier of a space, given its origin. Create a new,
    fresh identifier on the fly if it does not exist yet.
    """
    if origin in self.space_ids:
        return self.space_ids[origin]
    else:
        self._last_space_id += 1
        id = irefs.SpaceId(self._last_space_id)
        self.spaces[id] = origin
        self.space_ids[origin] = id
        return id

fresh_or_cached_answer_id

fresh_or_cached_answer_id(answer: LocatedAnswer) -> AnswerId

Obtain the identifier of an answer, given its content and the origin of the query that it corresponds to. Create a new, fresh identifier on the fly if it does not exist yet.

Source code in src/delphyne/core/traces.py

def fresh_or_cached_answer_id(
    self, answer: irefs.LocatedAnswer
) -> irefs.AnswerId:
    """
    Obtain the identifier of an answer, given its content and the
    origin of the query that it corresponds to. Create a new, fresh
    identifier on the fly if it does not exist yet.
    """
    if answer in self.answer_ids:
        return self.answer_ids[answer]
    else:
        self._last_answer_id += 1
        id = irefs.AnswerId(self._last_answer_id)
        self.answers[id] = answer
        self.answer_ids[answer] = id
        return id

export

export() -> ExportableTrace

Export a trace into a lightweight, serializable format.

Source code in src/delphyne/core/traces.py

def export(self) -> ExportableTrace:
    """
    Export a trace into a lightweight, serializable format.
    """
    nodes = {id.id: str(origin) for id, origin in self.nodes.items()}
    spaces = {id.id: str(origin) for id, origin in self.spaces.items()}
    answers = {
        id.id: ExportableLocatedAnswer(located.space.id, located.answer)
        for id, located in self.answers.items()
    }
    return ExportableTrace(nodes, spaces, answers)

check_consistency

check_consistency() -> None

Perform a sanity check on the trace.

Each node identifier is expanded into a full reference and then converted back to an identifier, which must be equal to the original one.

Source code in src/delphyne/core/traces.py

def check_consistency(self) -> None:
    """
    Perform a sanity check on the trace.

    Each node identifier is expanded into a full reference and then
    converted back to an identifier, which must be equal to the
    original one.
    """
    for id in self.nodes:
        expanded = self.expand_node_id(id)
        id_bis = self.convert_global_node_ref(expanded)
        assert id == id_bis

check_roundabout_consistency

check_roundabout_consistency() -> None

Perform a sanity check, before and after serializing and desarializing it.

Source code in src/delphyne/core/traces.py

def check_roundabout_consistency(self) -> None:
    """
    Perform a sanity check, before and after serializing and
    desarializing it.
    """
    self.check_consistency()
    exportable = self.export()
    copy = Trace.load(exportable)
    copy.check_consistency()
    exportable_copy = copy.export()
    if exportable != exportable_copy:
        print("Original exportable trace:")
        print(exportable)
        print("Exportable trace after round-trip:")
        print(exportable_copy)
        assert False

convert_global_space_path

convert_global_space_path(ref: GlobalSpacePath) -> SpaceId

Convert a full, global space reference denoting a quey origin into an id-based reference.

Source code in src/delphyne/core/traces.py

def convert_global_space_path(
    self, ref: refs.GlobalSpacePath
) -> irefs.SpaceId:
    """
    Convert a full, global space reference denoting a quey origin
    into an id-based reference.
    """
    id = Trace.MAIN_SPACE_ID
    for path, space in ref.steps:
        nid = self.fresh_or_cached_node_id(irefs.NestedIn(id))
        nid = self._convert_node_path(nid, path)
        id = self._convert_space_ref(nid, space)
    return id

convert_global_node_ref

convert_global_node_ref(path: GlobalNodeRef) -> NodeId

Convert a full, global node reference into an id-based one.

Source code in src/delphyne/core/traces.py

def convert_global_node_ref(
    self, path: refs.GlobalNodeRef
) -> irefs.NodeId:
    """
    Convert a full, global node reference into an id-based one.
    """
    space_id = self.convert_global_space_path(path.space)
    root = self.fresh_or_cached_node_id(irefs.NestedIn(space_id))
    return self._convert_node_path(root, path.path)

convert_answer_ref

convert_answer_ref(ref: GlobalAnswerRef) -> AnswerId

Convert a full answer reference into an answer id.

Source code in src/delphyne/core/traces.py

def convert_answer_ref(self, ref: refs.GlobalAnswerRef) -> irefs.AnswerId:
    """
    Convert a full answer reference into an answer id.
    """
    space = self.convert_global_space_path(ref[0])
    located = irefs.LocatedAnswer(space, ref[1])
    return self.fresh_or_cached_answer_id(located)

convert_location

convert_location(location: Location) -> ShortLocation

Convert a full location into an id-based one.

Source code in src/delphyne/core/traces.py

def convert_location(self, location: Location) -> ShortLocation:
    """
    Convert a full location into an id-based one.
    """
    match location:
        case None:
            return None
        case refs.GlobalNodeRef():
            return self.convert_global_node_ref(location)
        case refs.GlobalSpacePath():
            return self.convert_global_space_path(location)

convert_space_element_ref

convert_space_element_ref(node: NodeId, ref: SpaceElementRef) -> SpaceElementRef

Convert a full local space element reference into an id-based one, relative to a given node.

Source code in src/delphyne/core/traces.py

def convert_space_element_ref(
    self, node: irefs.NodeId, ref: refs.SpaceElementRef
) -> irefs.SpaceElementRef:
    """
    Convert a full local space element reference into an id-based one,
    relative to a given node.
    """
    # We leverage locality to speed up the computation.
    # The following would work but be much slower:
    #     space_id = self.convert_global_space_path(ref.space)

    # The space is attached to a node with an identifier.
    assert ref.space is not None
    space_id = self._convert_space_ref(node, ref.space)
    match ref.element:
        case refs.Answer():
            located = irefs.LocatedAnswer(space_id, ref.element)
            element = self.fresh_or_cached_answer_id(located)
        case refs.NodePath():
            nested_root_orig = irefs.NestedIn(space_id)
            nested_root = self.fresh_or_cached_node_id(nested_root_orig)
            element = self._convert_node_path(nested_root, ref.element)
    return irefs.SpaceElementRef(space_id, element)

expand_node_id

expand_node_id(id: NodeId) -> GlobalNodeRef

Convert a node identifier into a full, global node reference.

Source code in src/delphyne/core/traces.py

def expand_node_id(self, id: irefs.NodeId) -> refs.GlobalNodeRef:
    """
    Convert a node identifier into a full, global node reference.
    """
    orig, path = self._recover_path(id)
    space = self.expand_global_space_id(orig)
    return refs.GlobalNodeRef(space, path)

expand_space_ref

expand_space_ref(ref: SpaceRef) -> SpaceRef

Convert a local id-based space reference into a full one, relative to a given node.

Source code in src/delphyne/core/traces.py

def expand_space_ref(self, ref: irefs.SpaceRef) -> refs.SpaceRef:
    """
    Convert a local id-based space reference into a full one,
    relative to a given node.
    """
    args = tuple(self.expand_value_ref(a) for a in ref.args)
    return refs.SpaceRef(ref.name, args)

expand_value_ref

expand_value_ref(ref: ValueRef) -> ValueRef

Convert a local id-based value reference into a full one, relative to a given node.

Source code in src/delphyne/core/traces.py

def expand_value_ref(self, ref: irefs.ValueRef) -> refs.ValueRef:
    """
    Convert a local id-based value reference into a full one,
    relative to a given node.
    """
    if ref is None:
        return None
    elif isinstance(ref, tuple):
        return tuple(self.expand_value_ref(a) for a in ref)
    else:
        return self._expand_atomic_value_ref(ref)

space_elements_in_value_ref

space_elements_in_value_ref(ref: ValueRef) -> Iterable[SpaceElementRef]

Enumerate all local space elements that are used to define a value.

Duplicate values can be returned.

Source code in src/delphyne/core/traces.py

def space_elements_in_value_ref(
    self, ref: irefs.ValueRef
) -> Iterable[irefs.SpaceElementRef]:
    """
    Enumerate all local space elements that are used to define a
    value.

    Duplicate values can be returned.
    """

    if ref is None:
        pass
    elif isinstance(ref, tuple):
        for r in ref:
            yield from self.space_elements_in_value_ref(r)
    else:
        yield from self._space_elements_in_atomic_value_ref(ref)

ExportableTrace dataclass

A lightweight trace format that can be easily exported to JSON/YAML.

Attributes:

Name	Type	Description
nodes	dict[int, NodeOriginStr]	a mapping that defines node identifiers
spaces	dict[int, SpaceOriginStr]	a mapping that defines space identifiers
answers	dict[int, ExportableLocatedAnswer]	a mapping that defines answer identifiers

Source code in src/delphyne/core/traces.py

@dataclass
class ExportableTrace:
    """
    A lightweight trace format that can be easily exported to JSON/YAML.

    Attributes:
        nodes: a mapping that defines node identifiers
        spaces: a mapping that defines space identifiers
        answers: a mapping that defines answer identifiers
    """

    nodes: dict[int, NodeOriginStr]
    spaces: dict[int, SpaceOriginStr]
    answers: dict[int, ExportableLocatedAnswer]

NodeOriginStr

NodeOriginStr = str

A concise, serialized representation for NodeOrigin.

Can be parsed back using parse.node_origin.

SpaceOriginStr

SpaceOriginStr = str

A concise, serialized representation for SpaceOrigin.

Can be parsed back using parse.space_origin.

ExportableLocatedAnswer dataclass

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True)
class ExportableLocatedAnswer:
    space: int
    answer: refs.Answer

Location

Location = GlobalNodeRef | GlobalSpacePath | None

Optional location information for log messages.

Log messages can be attached to a given node or space.

ShortLocation

ShortLocation = NodeId | SpaceId | None

Optional location information for exportable log messages.

Tracers

Tracer

A mutable trace along with a mutable list of log messages.

Both components are protected by a lock to ensure thread-safety (some policies spawn multiple concurrent threads).

Attributes:

Name	Type	Description
trace		A mutable trace.
messages	list[LogMessage]	A mutable list of log messages.
lock		A reentrant lock protecting access to the trace and log. The lock is publicly exposed so that threads can log several successive messages without other threads interleaving new messages in between.

Source code in src/delphyne/core/traces.py

class Tracer:
    """
    A mutable trace along with a mutable list of log messages.

    Both components are protected by a lock to ensure thread-safety
    (some policies spawn multiple concurrent threads).

    Attributes:
        trace: A mutable trace.
        messages: A mutable list of log messages.
        lock: A reentrant lock protecting access to the trace and log.
            The lock is publicly exposed so that threads can log several
            successive messages without other threads interleaving new
            messages in between.
    """

    # TODO: there are cleaner ways to achieve good message order beyong
    # exposing the lock.

    def __init__(self, log_level: LogLevel = "info"):
        """
        Parameters:
            log_level: The minimum severity level of messages to log.
        """
        self.trace = Trace()
        self.messages: list[LogMessage] = []
        self.log_level: LogLevel = log_level

        # Different threads may be logging information or appending to
        # the trace in parallel.
        self.lock = threading.RLock()

    def global_node_id(self, node: refs.GlobalNodeRef) -> irefs.NodeId:
        """
        Ensure that a node at a given reference is present in the trace
        and return the corresponding node identififier.
        """
        with self.lock:
            return self.trace.convert_global_node_ref(node)

    def trace_node(self, node: refs.GlobalNodeRef) -> None:
        """
        Ensure that a node at a given reference is present in the trace.

        Returns the associated node identifier.

        See `tracer_hook` for registering a hook that automatically
        calls this method on all encountered nodes.
        """
        self.global_node_id(node)

    def trace_query(self, query: AttachedQuery[Any]) -> None:
        """
        Ensure that a query at a given reference is present in the
        trace, even if no answer is provided for it.
        """
        with self.lock:
            self.trace.convert_global_space_path(query.ref)

    def trace_answer(
        self, space: refs.GlobalSpacePath, answer: refs.Answer
    ) -> None:
        """
        Ensure that a given query answer is present in the trace, even
        it is is not used to reach a node.
        """
        with self.lock:
            self.trace.convert_answer_ref((space, answer))

    def log(
        self,
        level: LogLevel,
        message: str,
        metadata: object | None = None,
        *,
        location: Location | None = None,
        related: Sequence[LogMessageId | None] = (),
    ) -> LogMessageId | None:
        """
        Log a message, with optional metadata and location information.
        The metadata must be exportable to JSON using Pydantic.
        """
        if not log_level_greater_or_equal(level, self.log_level):
            return None
        time = datetime.now()
        with self.lock:
            id = len(self.messages)
            short_location = None
            if location is not None:
                short_location = self.trace.convert_location(location)
            self.messages.append(
                LogMessage(
                    message=message,
                    level=level,
                    time=time,
                    metadata=metadata,
                    location=short_location,
                    message_id=id,
                    related=[r for r in related if r is not None],
                )
            )
            return id

    def export_log(
        self, *, remove_timing_info: bool = False
    ) -> Iterable[ExportableLogMessage]:
        """
        Export the log into an easily serializable format.
        """
        with self.lock:
            for m in self.messages:
                node = None
                space = None
                if isinstance(m.location, irefs.NodeId):
                    node = m.location.id
                if isinstance(m.location, irefs.SpaceId):
                    space = m.location.id
                yield ExportableLogMessage(
                    message=m.message,
                    level=m.level,
                    time=m.time if not remove_timing_info else None,
                    node=node,
                    space=space,
                    metadata=pydantic_dump(object, m.metadata),
                    message_id=m.message_id,
                    related=tuple(m.related),
                )

    def export_trace(self) -> ExportableTrace:
        """
        Export the trace into an easily serializable format.
        """
        with self.lock:
            return self.trace.export()

__init__

__init__(log_level: LogLevel = 'info')

Parameters:

Name	Type	Description	Default
log_level	LogLevel	The minimum severity level of messages to log.	'info'

Source code in src/delphyne/core/traces.py

def __init__(self, log_level: LogLevel = "info"):
    """
    Parameters:
        log_level: The minimum severity level of messages to log.
    """
    self.trace = Trace()
    self.messages: list[LogMessage] = []
    self.log_level: LogLevel = log_level

    # Different threads may be logging information or appending to
    # the trace in parallel.
    self.lock = threading.RLock()

global_node_id

global_node_id(node: GlobalNodeRef) -> NodeId

Ensure that a node at a given reference is present in the trace and return the corresponding node identififier.

Source code in src/delphyne/core/traces.py

def global_node_id(self, node: refs.GlobalNodeRef) -> irefs.NodeId:
    """
    Ensure that a node at a given reference is present in the trace
    and return the corresponding node identififier.
    """
    with self.lock:
        return self.trace.convert_global_node_ref(node)

trace_node

trace_node(node: GlobalNodeRef) -> None

Ensure that a node at a given reference is present in the trace.

Returns the associated node identifier.

See tracer_hook for registering a hook that automatically calls this method on all encountered nodes.

Source code in src/delphyne/core/traces.py

def trace_node(self, node: refs.GlobalNodeRef) -> None:
    """
    Ensure that a node at a given reference is present in the trace.

    Returns the associated node identifier.

    See `tracer_hook` for registering a hook that automatically
    calls this method on all encountered nodes.
    """
    self.global_node_id(node)

trace_query

trace_query(query: AttachedQuery[Any]) -> None

Ensure that a query at a given reference is present in the trace, even if no answer is provided for it.

Source code in src/delphyne/core/traces.py

def trace_query(self, query: AttachedQuery[Any]) -> None:
    """
    Ensure that a query at a given reference is present in the
    trace, even if no answer is provided for it.
    """
    with self.lock:
        self.trace.convert_global_space_path(query.ref)

trace_answer

trace_answer(space: GlobalSpacePath, answer: Answer) -> None

Ensure that a given query answer is present in the trace, even it is is not used to reach a node.

Source code in src/delphyne/core/traces.py

def trace_answer(
    self, space: refs.GlobalSpacePath, answer: refs.Answer
) -> None:
    """
    Ensure that a given query answer is present in the trace, even
    it is is not used to reach a node.
    """
    with self.lock:
        self.trace.convert_answer_ref((space, answer))

log

log(
    level: LogLevel,
    message: str,
    metadata: object | None = None,
    *,
    location: Location | None = None,
    related: Sequence[LogMessageId | None] = (),
) -> LogMessageId | None

Log a message, with optional metadata and location information. The metadata must be exportable to JSON using Pydantic.

Source code in src/delphyne/core/traces.py

def log(
    self,
    level: LogLevel,
    message: str,
    metadata: object | None = None,
    *,
    location: Location | None = None,
    related: Sequence[LogMessageId | None] = (),
) -> LogMessageId | None:
    """
    Log a message, with optional metadata and location information.
    The metadata must be exportable to JSON using Pydantic.
    """
    if not log_level_greater_or_equal(level, self.log_level):
        return None
    time = datetime.now()
    with self.lock:
        id = len(self.messages)
        short_location = None
        if location is not None:
            short_location = self.trace.convert_location(location)
        self.messages.append(
            LogMessage(
                message=message,
                level=level,
                time=time,
                metadata=metadata,
                location=short_location,
                message_id=id,
                related=[r for r in related if r is not None],
            )
        )
        return id

export_log

export_log(*, remove_timing_info: bool = False) -> Iterable[ExportableLogMessage]

Export the log into an easily serializable format.

Source code in src/delphyne/core/traces.py

def export_log(
    self, *, remove_timing_info: bool = False
) -> Iterable[ExportableLogMessage]:
    """
    Export the log into an easily serializable format.
    """
    with self.lock:
        for m in self.messages:
            node = None
            space = None
            if isinstance(m.location, irefs.NodeId):
                node = m.location.id
            if isinstance(m.location, irefs.SpaceId):
                space = m.location.id
            yield ExportableLogMessage(
                message=m.message,
                level=m.level,
                time=m.time if not remove_timing_info else None,
                node=node,
                space=space,
                metadata=pydantic_dump(object, m.metadata),
                message_id=m.message_id,
                related=tuple(m.related),
            )

export_trace

export_trace() -> ExportableTrace

Export the trace into an easily serializable format.

Source code in src/delphyne/core/traces.py

def export_trace(self) -> ExportableTrace:
    """
    Export the trace into an easily serializable format.
    """
    with self.lock:
        return self.trace.export()

LogMessage dataclass

A log message.

Attributes:

Name	Type	Description
message	str	The message to log.
time	datetime	Time at which the message was produced
metadata	object | None	Optional metadata associated with the message, as an object that can be serialized to JSON using Pydantic.
location	ShortLocation | None	An optional location in the strategy tree where the message was logged, if applicable.
message_id	LogMessageId | None	Optionally, a unique identifier for the message, which can be used to tie related messages together.
related	Sequence[LogMessageId]	Optionally, a list of identifiers of related messages.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True, kw_only=True)
class LogMessage:
    """
    A log message.

    Attributes:
        message: The message to log.
        time: Time at which the message was produced
        metadata: Optional metadata associated with the message, as an
            object that can be serialized to JSON using Pydantic.
        location: An optional location in the strategy tree where the
            message was logged, if applicable.
        message_id: Optionally, a unique identifier for the message, which can
            be used to tie related messages together.
        related: Optionally, a list of identifiers of related messages.
    """

    message: str
    level: LogLevel
    time: datetime
    metadata: object | None = None
    location: ShortLocation | None = None
    message_id: LogMessageId | None = None
    related: Sequence[LogMessageId] = ()

ExportableLogMessage dataclass

An exportable log message, as a dataclass whose fields are JSON values (as opposed to LogMessage) and is thus easier to export.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True, kw_only=True)
class ExportableLogMessage:
    """
    An exportable log message, as a dataclass whose fields are JSON
    values (as opposed to `LogMessage`) and is thus easier to export.
    """

    message: str
    level: LogLevel
    message_id: int | None = None
    related: tuple[int, ...] = ()
    time: datetime | None = None
    node: int | None = None
    space: int | None = None
    metadata: object | None = None  # JSON value

tracer_hook

tracer_hook(tracer: Tracer) -> Callable[[Tree[Any, Any, Any]], None]

Standard hook to be passed to TreeMonitor to automatically log visited nodes into a trace.

Source code in src/delphyne/core/traces.py

def tracer_hook(tracer: Tracer) -> Callable[[Tree[Any, Any, Any]], None]:
    """
    Standard hook to be passed to `TreeMonitor` to automatically log
    visited nodes into a trace.
    """
    return lambda tree: tracer.trace_node(tree.ref)