Skip to content

tramdag.spec

tramdag.spec

The DAG specification: nodes, terms, validation and (de)serialization.

A model is one dict {node_name: NodeSpec} of ContinuousNode and OrdinalNode. Each node declares its transformation as an additive formula of Term subclasses, a + sum, whose first entry is the intercept (Intercept, written or prepended as I()) followed by shifts (LinearShift, ComplexShift, VaryingCoefficient). The paper's symbols I, LS, CS, VC are the same classes; SI() and CI(*parents) are the two intercept spellings with their arity checked.

validate_and_sort checks that every parent exists and enters through exactly one edge-owning term (VC modifiers may repeat) and gives the topological order. spec_to_dict and spec_from_dict are the checkpoint form. What the terms mean is docs/model.md.

The NodeSpec union follows the classes, because Python evaluates the union at definition time. The alias block at the end holds the paper's symbols, the notation of the docs and the spelling nearly every caller uses.

Term

One additive term of a node's transformation; each kind is a subclass.

Terms add: I("a") + CS("b") is the plain list of the two terms, a node's formula. A term is plain data (comparable, hashable, serializable by spec_to_dict) and knows its own spec-level rules (check, edge_parents, cells, classical). module is the class in modules that trains it, constructed as module(term, schema); the intercept slot adds n_params.

Subclass to add a term: set name (what the term key serializes) and module (a ShiftModule subclass) as class attributes, assign the options (the keyword arguments of __init__, with their defaults) to self, and raise ValueError in __init__ on a bad option or parent count.

A term is exactly its __dict__: the parents the base assigns and the options each subclass assigns from its own signature. That is what makes options, equality and serialization one line each.

Attributes:

Name Type Description
parents tuple[str, ...]

Ordered parent names the term depends on. Empty only for the bare simple intercept I(). For a VaryingCoefficient term, parents[0] is the treatment t and the rest are the effect modifiers; every other built-in term's parents all own their edges.
Source code in src/tramdag/spec.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
class Term:
    """One additive term of a node's transformation; each kind is a subclass.

    Terms add: ``I("a") + CS("b")`` is the plain list of the two terms, a
    node's formula. A term is plain data (comparable, hashable,
    serializable by [`spec_to_dict`][]) and knows its own spec-level rules
    (``check``, ``edge_parents``, ``cells``, ``classical``). ``module`` is the
    class in [`modules`][tramdag.modules] that trains it, constructed as
    ``module(term, schema)``; the intercept slot adds ``n_params``.

    Subclass to add a term: set ``name`` (what the ``term`` key serializes)
    and ``module`` (a [`ShiftModule`][tramdag.modules.ShiftModule] subclass)
    as class attributes, assign the options (the keyword arguments of
    ``__init__``, with their defaults) to ``self``, and raise ``ValueError``
    in ``__init__`` on a bad option or parent count.

    A term is exactly its ``__dict__``: the parents the base assigns and the
    options each subclass assigns from its own signature. That is what makes
    [`options`][], equality and serialization one line each.

    Attributes
    ----------
    parents : tuple[str, ...]
        Ordered parent names the term depends on. Empty only for the bare simple
        intercept ``I()``. For a [`VaryingCoefficient`][] term,
        ``parents[0]`` is the treatment ``t`` and the rest are the effect modifiers;
        every other built-in term's parents all own their edges.
    """

    def __init_subclass__(cls, **kwargs):
        """Refuse a term class without ``name`` (the wire key) or ``module``."""
        super().__init_subclass__(**kwargs)
        for attr in ("name", "module"):
            if attr not in cls.__dict__:
                raise TypeError(f"{cls.__name__}: a Term subclass sets `{attr} = ...`")

    def __init__(self, *parents: str):
        self.parents = tuple(parents)

    def options(self) -> dict:
        """Give everything the term carries, by name, the parents included."""
        return dict(vars(self))

    def acting_options(
        self, ordinal: bool, schema: dict[str, str | int]
    ) -> tuple[str, ...]:
        """Give the options that act on this term, the parents excluded.

        An option acts when setting it changes the model. ``ordinal`` says
        whether the node is ordinal. ``schema`` maps each node name to
        "continuous" or its level count. A subclass overrides this to drop
        the options that cannot act on it.
        """
        return tuple(k for k in self.options() if k != "parents")

    @classmethod
    def from_serialized(cls, parents: tuple[str, ...], options: dict) -> Term:
        """Rebuild a term from its serialized parents and options."""
        return cls(*parents, **options)

    @property
    def classical(self) -> bool:
        """Say whether the exact classical fit (``fit_classical``) handles this term."""
        return False

    def check(self, name: str, spec: dict[str, NodeSpec]) -> None:
        """Check the term against the spec it sits in; nothing to check here.

        A built-in term's own checks need only its arguments and run in
        ``__init__``. ``VC`` overrides this: its treatment and centering rules
        need the other nodes.
        """

    @property
    def edge_parents(self) -> tuple[str, ...]:
        """The parents that own an edge; every parent, for a plain term."""
        return self.parents

    def cells(self) -> list[tuple[str, str, bool]]:
        """Give the term's adjacency cells as ``(parent, tag, joint)`` triples.

        The tag is the term's name. ``joint`` says whether the term is one
        network over several parents; it comes from the term, not from its
        parent count, because a ``VC`` has a treatment plus modifiers and is
        not one network over them.
        """
        joint = len(self.parents) > 1
        return [(p, self.name, joint) for p in self.parents]

    def __eq__(self, other):
        """Compare the class and everything the term carries."""
        return type(self) is type(other) and vars(self) == vars(other)

    def __hash__(self):
        """Hash the class and the parents (coarser than ``__eq__``, by design)."""
        return hash((type(self), self.parents))

    def __repr__(self):
        """Name everything the term carries, the parents included.

        The one ``__repr__`` of the term classes: each entry of
        [`options`][] reads as ``name=value``, so a term with a
        keyword-only parent (a ``VC`` treatment) needs no spelling of its own.
        """
        args = [f"{k}={v!r}" for k, v in self.options().items()]
        return f"{self.name}({', '.join(args)})"

    def __add__(self, other: Term) -> list[Term]:
        """Start a plain term list."""
        if isinstance(other, Term):
            return [self, other]
        return NotImplemented

    def __radd__(self, other: list[Term]) -> list[Term]:
        """Extend a term list from the right.

        ``I("a") + CS("b") + LS("c")`` evaluates left to right: the first
        ``+`` gives a list, so the second is ``list + Term`` and lands here.
        """
        if isinstance(other, list):
            return [*other, self]
        return NotImplemented

classical property 

classical

Say whether the exact classical fit (fit_classical) handles this term.

edge_parents property 

edge_parents

The parents that own an edge; every parent, for a plain term.

__init_subclass__ 

__init_subclass__(**kwargs)

Refuse a term class without name (the wire key) or module.

Source code in src/tramdag/spec.py 
447
448
449
450
451
452
def __init_subclass__(cls, **kwargs):
    """Refuse a term class without ``name`` (the wire key) or ``module``."""
    super().__init_subclass__(**kwargs)
    for attr in ("name", "module"):
        if attr not in cls.__dict__:
            raise TypeError(f"{cls.__name__}: a Term subclass sets `{attr} = ...`")

options 

options()

Give everything the term carries, by name, the parents included.

Source code in src/tramdag/spec.py 
457
458
459
def options(self) -> dict:
    """Give everything the term carries, by name, the parents included."""
    return dict(vars(self))

acting_options 

acting_options(ordinal, schema)

Give the options that act on this term, the parents excluded.

An option acts when setting it changes the model. ordinal says whether the node is ordinal. schema maps each node name to "continuous" or its level count. A subclass overrides this to drop the options that cannot act on it.

Source code in src/tramdag/spec.py 
461
462
463
464
465
466
467
468
469
470
471
def acting_options(
    self, ordinal: bool, schema: dict[str, str | int]
) -> tuple[str, ...]:
    """Give the options that act on this term, the parents excluded.

    An option acts when setting it changes the model. ``ordinal`` says
    whether the node is ordinal. ``schema`` maps each node name to
    "continuous" or its level count. A subclass overrides this to drop
    the options that cannot act on it.
    """
    return tuple(k for k in self.options() if k != "parents")

from_serialized classmethod 

from_serialized(parents, options)

Rebuild a term from its serialized parents and options.

Source code in src/tramdag/spec.py 
473
474
475
476
@classmethod
def from_serialized(cls, parents: tuple[str, ...], options: dict) -> Term:
    """Rebuild a term from its serialized parents and options."""
    return cls(*parents, **options)

check 

check(name, spec)

Check the term against the spec it sits in; nothing to check here.

A built-in term's own checks need only its arguments and run in __init__. VC overrides this: its treatment and centering rules need the other nodes.

Source code in src/tramdag/spec.py 
483
484
485
486
487
488
489
def check(self, name: str, spec: dict[str, NodeSpec]) -> None:
    """Check the term against the spec it sits in; nothing to check here.

    A built-in term's own checks need only its arguments and run in
    ``__init__``. ``VC`` overrides this: its treatment and centering rules
    need the other nodes.
    """

cells 

cells()

Give the term's adjacency cells as (parent, tag, joint) triples.

The tag is the term's name. joint says whether the term is one network over several parents; it comes from the term, not from its parent count, because a VC has a treatment plus modifiers and is not one network over them.

Source code in src/tramdag/spec.py 
496
497
498
499
500
501
502
503
504
505
def cells(self) -> list[tuple[str, str, bool]]:
    """Give the term's adjacency cells as ``(parent, tag, joint)`` triples.

    The tag is the term's name. ``joint`` says whether the term is one
    network over several parents; it comes from the term, not from its
    parent count, because a ``VC`` has a treatment plus modifiers and is
    not one network over them.
    """
    joint = len(self.parents) > 1
    return [(p, self.name, joint) for p in self.parents]

__eq__ 

__eq__(other)

Compare the class and everything the term carries.

Source code in src/tramdag/spec.py 
507
508
509
def __eq__(self, other):
    """Compare the class and everything the term carries."""
    return type(self) is type(other) and vars(self) == vars(other)

__hash__ 

__hash__()

Hash the class and the parents (coarser than __eq__, by design).

Source code in src/tramdag/spec.py 
511
512
513
def __hash__(self):
    """Hash the class and the parents (coarser than ``__eq__``, by design)."""
    return hash((type(self), self.parents))

__repr__ 

__repr__()

Name everything the term carries, the parents included.

The one __repr__ of the term classes: each entry of options reads as name=value, so a term with a keyword-only parent (a VC treatment) needs no spelling of its own.

Source code in src/tramdag/spec.py 
515
516
517
518
519
520
521
522
523
def __repr__(self):
    """Name everything the term carries, the parents included.

    The one ``__repr__`` of the term classes: each entry of
    [`options`][] reads as ``name=value``, so a term with a
    keyword-only parent (a ``VC`` treatment) needs no spelling of its own.
    """
    args = [f"{k}={v!r}" for k, v in self.options().items()]
    return f"{self.name}({', '.join(args)})"

__add__ 

__add__(other)

Start a plain term list.

Source code in src/tramdag/spec.py 
525
526
527
528
529
def __add__(self, other: Term) -> list[Term]:
    """Start a plain term list."""
    if isinstance(other, Term):
        return [self, other]
    return NotImplemented

__radd__ 

__radd__(other)

Extend a term list from the right.

I("a") + CS("b") + LS("c") evaluates left to right: the first + gives a list, so the second is list + Term and lands here.

Source code in src/tramdag/spec.py 
531
532
533
534
535
536
537
538
539
def __radd__(self, other: list[Term]) -> list[Term]:
    """Extend a term list from the right.

    ``I("a") + CS("b") + LS("c")`` evaluates left to right: the first
    ``+`` gives a list, so the second is ``list + Term`` and lands here.
    """
    if isinstance(other, list):
        return [*other, self]
    return NotImplemented

Intercept

Bases: Term

The intercept term I: the parents reshape the monotone transform.

Without parents it is the paper's simple intercept SI: one free parameter vector, the same for every row. With parents it is the complex intercept CI: the transform parameters become a function of them. SI and CI are the two spellings with their arity checked.

Parameters:

Name Type Description Default
*parents str

Parent names. Several parents form one joint network (an interaction) unless allow_interaction=False.

 ()
transform str | None

A continuous node's monotone transform: "bernstein", "spline" or "affine". None, the default, means the node picks "bernstein"; an ordinal node's intercept is the cutpoint vector and refuses a transform.

 None
transform_kwargs dict | None

The transform's keyword arguments as one mapping, for example I(transform="spline", transform_kwargs={"bins": 16}). This is also how a spec YAML and a checkpoint carry them.

 None
allow_interaction bool

False makes a multi-parent term additive: one network per parent, their parameter vectors summed in coefficient space. A node takes at most one intercept term with parents. Write an additive intercept with this flag, not with several intercept terms. Default True.

 True

Other Parameters:

Name Type Description
units list[int] | tuple[int, ...]

Hidden layers of the term's network, for example units=[16], by default (8, 8).
activation str

Activation of the network's hidden layers, by default "relu"; one of the keys of modules.ACTIVATIONS.
batch_norm bool

Batch-normalize the network's hidden layers, by default False.
input_transform str | callable | None

"minmax", "standardize" or a callable fn(x, train) applied per continuous parent column (train is that column's raw training data, frozen at calibration). Parents only. None, the default, applies no transform.
Notes

The transform class takes its keywords only through transform_kwargs, so a misspelt option is a TypeError at once. module is a function, so staticmethod stops it from binding as a method.

Raises:

Type Description
ValueError

If a parentless term carries input_transform, or allow_interaction=False comes with fewer than two parents (an interaction to disallow needs two).
Source code in src/tramdag/spec.py 
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
class Intercept(Term):
    """The intercept term ``I``: the parents reshape the monotone transform.

    Without parents it is the paper's simple intercept **SI**: one free
    parameter vector, the same for every row. With parents it is the
    complex intercept **CI**: the transform parameters become a function
    of them. [`SI`][] and [`CI`][] are the two spellings with their
    arity checked.

    Parameters
    ----------
    *parents : str
        Parent names. Several parents form one **joint** network (an
        interaction) unless ``allow_interaction=False``.
    transform : str | None, optional
        A continuous node's monotone transform: ``"bernstein"``,
        ``"spline"`` or ``"affine"``. ``None``, the default, means the node
        picks ``"bernstein"``; an ordinal node's intercept is the cutpoint
        vector and refuses a transform.
    transform_kwargs : dict | None, optional
        The transform's keyword arguments as one mapping, for example
        ``I(transform="spline", transform_kwargs={"bins": 16})``. This is
        also how a spec YAML and a checkpoint carry them.
    allow_interaction : bool, optional
        ``False`` makes a multi-parent term **additive**: one network per
        parent, their parameter vectors summed in coefficient space. A node
        takes at most one intercept term with parents. Write an additive
        intercept with this flag, not with several intercept terms. Default
        ``True``.

    Other Parameters
    ----------------
    units : list[int] | tuple[int, ...], optional
        Hidden layers of the term's network, for example ``units=[16]``, by
        default ``(8, 8)``.
    activation : str, optional
        Activation of the network's hidden layers, by default ``"relu"``;
        one of the keys of ``modules.ACTIVATIONS``.
    batch_norm : bool, optional
        Batch-normalize the network's hidden layers, by default False.
    input_transform : str | callable | None, optional
        ``"minmax"``, ``"standardize"`` or a callable ``fn(x, train)``
        applied per continuous parent column (``train`` is that column's
        raw training data, frozen at calibration). Parents only.
        ``None``, the default, applies no transform.

    Notes
    -----
    The transform class takes its keywords only through ``transform_kwargs``,
    so a misspelt option is a ``TypeError`` at once. ``module`` is a function,
    so ``staticmethod`` stops it from binding as a method.

    Raises
    ------
    ValueError
        If a parentless term carries ``input_transform``, or
        ``allow_interaction=False`` comes with fewer than two parents (an
        interaction to disallow needs two).
    """

    name = "I"
    module = staticmethod(intercept_module)

    def __init__(
        self,
        *parents: str,
        transform: str | None = None,
        transform_kwargs: dict | None = None,
        allow_interaction: bool = True,
        units: tuple[int, ...] | list[int] = (8, 8),
        activation: str = "relu",
        batch_norm: bool = False,
        input_transform: object = None,
    ):
        super().__init__(*parents)
        self.transform = transform
        self.transform_kwargs = dict(transform_kwargs or {})
        self.allow_interaction = allow_interaction
        self.units = tuple(units)
        self.activation = activation
        self.batch_norm = batch_norm
        self.input_transform = _checked_input_transform(input_transform)
        if not self.parents and self.input_transform is not None:
            raise ValueError(
                "a simple intercept has no network inputs; input_transform= "
                "belongs on CI/CS/VC terms"
            )
        if not self.allow_interaction and len(self.parents) < 2:
            raise ValueError(
                "allow_interaction=False makes a MULTI-parent intercept additive; "
                "with one parent there is no interaction to disallow; drop the "
                "argument"
            )

    def acting_options(
        self, ordinal: bool, schema: dict[str, str | int]
    ) -> tuple[str, ...]:
        """Give the options that act on this intercept.

        An ordinal intercept is the cutpoint vector and takes no transform. A
        parentless intercept has no network, and neither has one with
        ``units=()``. ``allow_interaction`` acts with two parents or more.
        """
        names = [] if ordinal else ["transform", "transform_kwargs"]
        if self.parents:
            names.append("units")
            if self.units:
                names += ["activation", "batch_norm", "input_transform"]
        if len(self.parents) > 1:
            names.append("allow_interaction")
        return tuple(names)

    @property
    def classical(self) -> bool:
        """Say yes only for a parentless ``I()``, the simple baseline."""
        return not self.parents

    def cells(self) -> list[tuple[str, str, bool]]:
        """Tag an intercept edge ``CI``: a cell exists only when it has parents.

        It is joint only when one network reads several parents; an additive
        intercept (``allow_interaction=False``) has one network per parent.
        """
        joint = len(self.parents) > 1 and self.allow_interaction
        return [(p, "CI", joint) for p in self.parents]

classical property 

classical

Say yes only for a parentless I(), the simple baseline.

acting_options 

acting_options(ordinal, schema)

Give the options that act on this intercept.

An ordinal intercept is the cutpoint vector and takes no transform. A parentless intercept has no network, and neither has one with units=(). allow_interaction acts with two parents or more.

Source code in src/tramdag/spec.py 
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def acting_options(
    self, ordinal: bool, schema: dict[str, str | int]
) -> tuple[str, ...]:
    """Give the options that act on this intercept.

    An ordinal intercept is the cutpoint vector and takes no transform. A
    parentless intercept has no network, and neither has one with
    ``units=()``. ``allow_interaction`` acts with two parents or more.
    """
    names = [] if ordinal else ["transform", "transform_kwargs"]
    if self.parents:
        names.append("units")
        if self.units:
            names += ["activation", "batch_norm", "input_transform"]
    if len(self.parents) > 1:
        names.append("allow_interaction")
    return tuple(names)

cells 

cells()

Tag an intercept edge CI: a cell exists only when it has parents.

It is joint only when one network reads several parents; an additive intercept (allow_interaction=False) has one network per parent.

Source code in src/tramdag/spec.py 
659
660
661
662
663
664
665
666
def cells(self) -> list[tuple[str, str, bool]]:
    """Tag an intercept edge ``CI``: a cell exists only when it has parents.

    It is joint only when one network reads several parents; an additive
    intercept (``allow_interaction=False``) has one network per parent.
    """
    joint = len(self.parents) > 1 and self.allow_interaction
    return [(p, "CI", joint) for p in self.parents]

LinearShift

Bases: Term

The linear shift LS: \(\beta x\), one interpretable raw-unit coefficient.

Parameters:

Name Type Description Default
*parents str

Exactly one parent name. The weight stays the interpretable raw-unit coefficient, so an LS takes no input_transform.

 ()

Raises:

Type Description
ValueError

If the parent count is not one.
Source code in src/tramdag/spec.py 
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
class LinearShift(Term):
    r"""The linear shift ``LS``: $\beta x$, one interpretable raw-unit coefficient.

    Parameters
    ----------
    *parents : str
        Exactly one parent name. The weight stays the interpretable
        raw-unit coefficient, so an LS takes no ``input_transform``.

    Raises
    ------
    ValueError
        If the parent count is not one.
    """

    name = "LS"
    module = LinearShiftModule

    def __init__(self, *parents: str):
        super().__init__(*parents)
        if len(self.parents) != 1:
            raise ValueError("LS() takes exactly one parent")

    @property
    def classical(self) -> bool:
        """Say yes: an LS is a classical coefficient."""
        return True

classical property 

classical

Say yes: an LS is a classical coefficient.

ComplexShift

Bases: Term

The complex shift CS: an additive network g(x) on the latent scale.

Parameters:

Name Type Description Default
*parents str

At least one parent name. Several parents feed one joint network; CS("a") + CS("b") are two additive terms instead.

 ()

Other Parameters:

Name Type Description
units list[int] | tuple[int, ...]

Hidden layers, for example units=[16], by default (64, 128, 64).
activation str

Activation of the hidden layers, by default "relu".
batch_norm bool

Batch-normalize the hidden layers, by default False.
input_transform str | callable | None

As for Intercept. None, the default, applies no transform.

Raises:

Type Description
ValueError

If no parent is given.
Source code in src/tramdag/spec.py 
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
class ComplexShift(Term):
    """The complex shift ``CS``: an additive network ``g(x)`` on the latent scale.

    Parameters
    ----------
    *parents : str
        At least one parent name. Several parents feed one joint network;
        ``CS("a") + CS("b")`` are two additive terms instead.

    Other Parameters
    ----------------
    units : list[int] | tuple[int, ...], optional
        Hidden layers, for example ``units=[16]``, by default ``(64, 128, 64)``.
    activation : str, optional
        Activation of the hidden layers, by default ``"relu"``.
    batch_norm : bool, optional
        Batch-normalize the hidden layers, by default False.
    input_transform : str | callable | None, optional
        As for [`Intercept`][]. ``None``, the default, applies no
        transform.

    Raises
    ------
    ValueError
        If no parent is given.
    """

    name = "CS"
    module = ComplexShiftModule

    def __init__(
        self,
        *parents: str,
        units: tuple[int, ...] | list[int] = (64, 128, 64),
        activation: str = "relu",
        batch_norm: bool = False,
        input_transform: object = None,
    ):
        super().__init__(*parents)
        self.units = tuple(units)
        self.activation = activation
        self.batch_norm = batch_norm
        self.input_transform = _checked_input_transform(input_transform)
        if not self.parents:
            raise ValueError("CS() needs at least one parent")

    def acting_options(
        self, ordinal: bool, schema: dict[str, str | int]
    ) -> tuple[str, ...]:
        """Give the options that act; ``input_transform`` needs a continuous parent."""
        names = super().acting_options(ordinal, schema)
        if any(schema.get(p) == "continuous" for p in self.parents):
            return names
        return tuple(k for k in names if k != "input_transform")

acting_options 

acting_options(ordinal, schema)

Give the options that act; input_transform needs a continuous parent.

Source code in src/tramdag/spec.py 
744
745
746
747
748
749
750
751
def acting_options(
    self, ordinal: bool, schema: dict[str, str | int]
) -> tuple[str, ...]:
    """Give the options that act; ``input_transform`` needs a continuous parent."""
    names = super().acting_options(ordinal, schema)
    if any(schema.get(p) == "continuous" for p in self.parents):
        return names
    return tuple(k for k in names if k != "input_transform")

VaryingCoefficient

Bases: Term

The varying-coefficient shift VC: \((\beta_0 + b_\Theta(\text{mod}))\, x_t\).

\(b_\Theta\) is a small network whose weights carry the L2 penalty; beta0 is not penalized. The network's output is zero-initialized and re-centered to mean zero over the training rows after the fit, so beta0 is the main effect. Read the fitted effect out with varying_coef. Only t owns its edge, so the treatment leads the parents; the modifiers may also appear in the node's other terms. docs/varying-coefficients.md is the guide.

Parameters:

Name Type Description Default
*modifiers str

The effect modifiers, the covariates that enter \(b_\Theta\). Empty means a constant effect.

 ()
t str

The treatment (required keyword). Must be a continuous node or a binary (2-level) ordinal node. The term is linear in x_t.

 required

Other Parameters:

Name Type Description
penalty float

L2 weight on the \(b_\Theta\) weights, on the total-NLL scale, by default 1.0. Must be >= 0.
propensity str | None

The training-frame column holding the out-of-fold propensities \(P(t = 1 \mid \mathrm{pa}_t)\) per row, by default None (no centering). With a column the regressor becomes \(\beta(x)\,(x_t - \hat e(\mathrm{pa}_t))\). Training reads the column as frozen data; every query after the fit recomputes \(\hat e\) from the flow's own treatment node. Requires a binary ordinal t.
units list[int] | tuple[int, ...]

Hidden layers of \(b_\Theta\), by default (16,).
activation str

Activation of \(b_\Theta\)'s hidden layers, by default "relu".
batch_norm bool

Batch-normalize \(b_\Theta\)'s hidden layers, by default False.
input_transform str | callable | None

As for Intercept, over the modifiers. None, the default, applies no transform.

Raises:

Type Description
ValueError

If t is also a modifier or if penalty is negative.
Source code in src/tramdag/spec.py 
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
class VaryingCoefficient(Term):
    r"""The varying-coefficient shift ``VC``: $(\beta_0 + b_\Theta(\text{mod}))\, x_t$.

    $b_\Theta$ is a small network whose weights carry the L2 ``penalty``;
    ``beta0`` is not penalized. The network's output is zero-initialized and
    re-centered to mean zero over the training rows after the fit, so
    ``beta0`` is the main effect. Read the fitted effect out with
    [`varying_coef`][tramdag.flow.CausalFlowDAG.varying_coef]. Only ``t``
    owns its edge, so the treatment leads the parents; the modifiers may also
    appear in the node's other terms.
    ``docs/varying-coefficients.md`` is the guide.

    Parameters
    ----------
    *modifiers : str
        The effect modifiers, the covariates that enter $b_\Theta$.
        Empty means a constant effect.
    t : str
        The treatment (required keyword). Must be a continuous node or a
        binary (2-level) ordinal node. The term is linear in ``x_t``.

    Other Parameters
    ----------------
    penalty : float, optional
        L2 weight on the $b_\Theta$ weights, on the total-NLL scale, by
        default 1.0. Must be >= 0.
    propensity : str | None, optional
        The training-frame column holding the out-of-fold propensities
        $P(t = 1 \mid \mathrm{pa}_t)$ per row, by default ``None`` (no
        centering). With a column the regressor becomes
        $\beta(x)\,(x_t - \hat e(\mathrm{pa}_t))$. Training reads the column as
        frozen data; every query after the fit recomputes $\hat e$ from the
        flow's own treatment node. Requires a binary ordinal ``t``.
    units : list[int] | tuple[int, ...], optional
        Hidden layers of $b_\Theta$, by default ``(16,)``.
    activation : str, optional
        Activation of $b_\Theta$'s hidden layers, by default ``"relu"``.
    batch_norm : bool, optional
        Batch-normalize $b_\Theta$'s hidden layers, by default False.
    input_transform : str | callable | None, optional
        As for [`Intercept`][], over the modifiers. ``None``, the default,
        applies no transform.

    Raises
    ------
    ValueError
        If ``t`` is also a modifier or if ``penalty`` is negative.
    """

    name = "VC"
    module = VaryingCoefficientModule

    def __init__(
        self,
        *modifiers: str,
        t: str,
        penalty: float = 1.0,
        propensity: str | None = None,
        units: tuple[int, ...] | list[int] = (16,),
        activation: str = "relu",
        batch_norm: bool = False,
        input_transform: object = None,
    ):
        if penalty < 0:
            raise ValueError(f"VC(): penalty must be >= 0, got {penalty!r}")
        super().__init__(t, *modifiers)
        self.penalty = float(penalty)
        self.propensity = propensity
        self.units = tuple(units)
        self.activation = activation
        self.batch_norm = batch_norm
        self.input_transform = _checked_input_transform(input_transform)
        if t in modifiers:
            raise ValueError(
                f"VC(): {t!r} cannot be both the treatment (t) and a modifier"
            )

    def acting_options(
        self, ordinal: bool, schema: dict[str, str | int]
    ) -> tuple[str, ...]:
        """Give the options that act, ``input_transform`` over a continuous modifier."""
        names = super().acting_options(ordinal, schema)
        if any(schema.get(p) == "continuous" for p in self.parents[1:]):
            return names
        return tuple(k for k in names if k != "input_transform")

    @classmethod
    def from_serialized(cls, parents: tuple[str, ...], options: dict) -> Term:
        """Rebuild from the serialized parents: the treatment comes first."""
        return cls(*parents[1:], t=parents[0], **options)

    def check(self, name: str, spec: dict[str, NodeSpec]) -> None:
        """Check the treatment and the centering against the spec.

        Raises
        ------
        ValueError
            If the centering column is malformed or collides with a node,
            the treatment is a multi-level ordinal, or centering meets a
            continuous or an itself-centered treatment.
        """
        t = self.parents[0]
        t_node = spec[t]
        if self.propensity is not None and not isinstance(self.propensity, str):
            raise ValueError(
                f"node {name!r}: VC(propensity=) names the propensity COLUMN of "
                "the training frame (out-of-fold P(t=1|pa_t) per row), or is "
                f"None; got {self.propensity!r}. Cross-fit the propensities "
                "outside, merge them as a column and pass "
                "propensity='<column>'."
            )
        if self.propensity and self.propensity in spec:
            raise ValueError(
                f"node {name!r}: the propensity column {self.propensity!r} "
                "collides with a node name"
            )
        if t_node.kind == "ordinal" and t_node.levels != 2:
            raise ValueError(
                f"node {name!r}: VC treatment {t!r} is ordinal with "
                f"{t_node.levels} levels. Only a 2-level (binary) ordinal "
                "treatment is supported."
            )
        if self.propensity and t_node.kind != "ordinal":
            raise ValueError(
                f"node {name!r}: VC(propensity=...) needs a binary ordinal "
                f"treatment, and {t!r} is continuous"
            )
        if self.propensity and any(
            isinstance(term, VaryingCoefficient) and term.propensity
            for term in t_node.terms
        ):
            raise ValueError(
                f"node {name!r}: treatment {t!r} carries a centered VC term "
                "itself; chained centering is not supported"
            )

    @property
    def edge_parents(self) -> tuple[str, ...]:
        """Only the treatment owns an edge; the modifiers may repeat elsewhere."""
        return self.parents[:1]

    def cells(self) -> list[tuple[str, str, bool]]:
        """Tag the treatment cell ``VC`` and the modifiers ``VCm``; never joint."""
        t, mods = self.parents[0], self.parents[1:]
        return [(t, "VC", False)] + [(p, "VCm", False) for p in mods]

edge_parents property 

edge_parents

Only the treatment owns an edge; the modifiers may repeat elsewhere.

acting_options 

acting_options(ordinal, schema)

Give the options that act, input_transform over a continuous modifier.

Source code in src/tramdag/spec.py 
831
832
833
834
835
836
837
838
def acting_options(
    self, ordinal: bool, schema: dict[str, str | int]
) -> tuple[str, ...]:
    """Give the options that act, ``input_transform`` over a continuous modifier."""
    names = super().acting_options(ordinal, schema)
    if any(schema.get(p) == "continuous" for p in self.parents[1:]):
        return names
    return tuple(k for k in names if k != "input_transform")

from_serialized classmethod 

from_serialized(parents, options)

Rebuild from the serialized parents: the treatment comes first.

Source code in src/tramdag/spec.py 
840
841
842
843
@classmethod
def from_serialized(cls, parents: tuple[str, ...], options: dict) -> Term:
    """Rebuild from the serialized parents: the treatment comes first."""
    return cls(*parents[1:], t=parents[0], **options)

check 

check(name, spec)

Check the treatment and the centering against the spec.

Raises:

Type Description
ValueError

If the centering column is malformed or collides with a node, the treatment is a multi-level ordinal, or centering meets a continuous or an itself-centered treatment.
Source code in src/tramdag/spec.py 
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
def check(self, name: str, spec: dict[str, NodeSpec]) -> None:
    """Check the treatment and the centering against the spec.

    Raises
    ------
    ValueError
        If the centering column is malformed or collides with a node,
        the treatment is a multi-level ordinal, or centering meets a
        continuous or an itself-centered treatment.
    """
    t = self.parents[0]
    t_node = spec[t]
    if self.propensity is not None and not isinstance(self.propensity, str):
        raise ValueError(
            f"node {name!r}: VC(propensity=) names the propensity COLUMN of "
            "the training frame (out-of-fold P(t=1|pa_t) per row), or is "
            f"None; got {self.propensity!r}. Cross-fit the propensities "
            "outside, merge them as a column and pass "
            "propensity='<column>'."
        )
    if self.propensity and self.propensity in spec:
        raise ValueError(
            f"node {name!r}: the propensity column {self.propensity!r} "
            "collides with a node name"
        )
    if t_node.kind == "ordinal" and t_node.levels != 2:
        raise ValueError(
            f"node {name!r}: VC treatment {t!r} is ordinal with "
            f"{t_node.levels} levels. Only a 2-level (binary) ordinal "
            "treatment is supported."
        )
    if self.propensity and t_node.kind != "ordinal":
        raise ValueError(
            f"node {name!r}: VC(propensity=...) needs a binary ordinal "
            f"treatment, and {t!r} is continuous"
        )
    if self.propensity and any(
        isinstance(term, VaryingCoefficient) and term.propensity
        for term in t_node.terms
    ):
        raise ValueError(
            f"node {name!r}: treatment {t!r} carries a centered VC term "
            "itself; chained centering is not supported"
        )

cells 

cells()

Tag the treatment cell VC and the modifiers VCm; never joint.

Source code in src/tramdag/spec.py 
895
896
897
898
def cells(self) -> list[tuple[str, str, bool]]:
    """Tag the treatment cell ``VC`` and the modifiers ``VCm``; never joint."""
    t, mods = self.parents[0], self.parents[1:]
    return [(t, "VC", False)] + [(p, "VCm", False) for p in mods]

ContinuousNode

Continuous variable, modelled by a monotone 1-D transform + shifts.

Parameters:

Name Type Description Default
terms Term | list[Term] | None

The additive formula for h: a + sum of terms or a single term. None (default) is a source node. The class of the monotone transform is chosen on the intercept term, I(..., transform="spline"); the default is "bernstein". The arguments of the transform class go in transform_kwargs={...}. If they are wrong, that class says so; this layer does not second-guess it.

 None

Raises:

Type Description
TypeError

If a formula entry is not a term.
ValueError

If the formula has more than one intercept term, or the intercept is not first.
Source code in src/tramdag/spec.py 
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
class ContinuousNode:
    """Continuous variable, modelled by a monotone 1-D transform + shifts.

    Parameters
    ----------
    terms : Term | list[Term] | None, optional
        The additive formula for ``h``: a ``+`` sum of terms or a single
        term. ``None`` (default) is a source node. The
        class of the monotone transform is chosen on the intercept term,
        ``I(..., transform="spline")``; the default is ``"bernstein"``. The
        arguments of the transform class go in
        ``transform_kwargs={...}``. If they are wrong, that class says so; this
        layer does not second-guess it.

    Raises
    ------
    TypeError
        If a formula entry is not a term.
    ValueError
        If the formula has more than one intercept term, or the intercept
        is not first.
    """

    kind = "continuous"

    def __init__(self, terms=None):
        self.terms = _normalize_terms(terms)
        intercept = self.terms[0]
        self.transform = intercept.transform or DEFAULT_TRANSFORM
        self.transform_kwargs = dict(intercept.transform_kwargs)

    def __repr__(self):
        """Show the terms and the transform."""
        return f"ContinuousNode({self.terms!r}, transform={self.transform!r})"

    def __eq__(self, other):
        """Compare the terms; the transform is derived from them."""
        return isinstance(other, ContinuousNode) and self.terms == other.terms

    def __hash__(self):
        """Hash what ``__eq__`` compares, so nodes work in sets and as keys."""
        return hash((self.kind, tuple(self.terms)))

__repr__ 

__repr__()

Show the terms and the transform.

Source code in src/tramdag/spec.py 
932
933
934
def __repr__(self):
    """Show the terms and the transform."""
    return f"ContinuousNode({self.terms!r}, transform={self.transform!r})"

__eq__ 

__eq__(other)

Compare the terms; the transform is derived from them.

Source code in src/tramdag/spec.py 
936
937
938
def __eq__(self, other):
    """Compare the terms; the transform is derived from them."""
    return isinstance(other, ContinuousNode) and self.terms == other.terms

__hash__ 

__hash__()

Hash what __eq__ compares, so nodes work in sets and as keys.

Source code in src/tramdag/spec.py 
940
941
942
def __hash__(self):
    """Hash what ``__eq__`` compares, so nodes work in sets and as keys."""
    return hash((self.kind, tuple(self.terms)))

OrdinalNode

Ordinal variable with levels ordered classes, stored 0 to levels-1.

An ordered logit models it: increasing cutpoints plus the shift terms.

Parameters:

Name Type Description Default
levels int

Number of ordered classes.

 required
terms Term | list[Term] | None

The additive formula, as for ContinuousNode, by default None.

 None

Raises:

Type Description
ValueError

If levels is below 2, the intercept names a transform, or the formula breaks a rule of ContinuousNode.
TypeError

If a formula entry is not a term.
Source code in src/tramdag/spec.py 
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
class OrdinalNode:
    """Ordinal variable with ``levels`` ordered classes, stored 0 to levels-1.

    An ordered logit models it: increasing cutpoints plus the shift terms.

    Parameters
    ----------
    levels : int
        Number of ordered classes.
    terms : Term | list[Term] | None, optional
        The additive formula, as for [`ContinuousNode`][], by default
        ``None``.

    Raises
    ------
    ValueError
        If ``levels`` is below 2, the intercept names a transform, or the
        formula breaks a rule of [`ContinuousNode`][].
    TypeError
        If a formula entry is not a term.
    """

    kind = "ordinal"

    def __init__(self, levels: int, terms=None):
        self.levels = int(levels)
        if self.levels < 2:
            raise ValueError(f"OrdinalNode(levels=) must be >= 2, got {self.levels!r}")
        self.terms = _normalize_terms(terms)
        intercept = self.terms[0]
        if intercept.transform or intercept.transform_kwargs:
            raise ValueError(
                "I(transform=...) is for continuous nodes. An ordinal node's "
                "intercept is the cutpoint vector, it has no transform to choose."
            )

    def __repr__(self):
        """Show the levels and the terms."""
        return f"OrdinalNode({self.levels}, {self.terms!r})"

    def __eq__(self, other):
        """Compare levels and terms."""
        return (
            isinstance(other, OrdinalNode)
            and self.levels == other.levels
            and self.terms == other.terms
        )

    def __hash__(self):
        """Hash what ``__eq__`` compares, so nodes work in sets and as keys."""
        return hash((self.kind, self.levels, tuple(self.terms)))

__repr__ 

__repr__()

Show the levels and the terms.

Source code in src/tramdag/spec.py 
981
982
983
def __repr__(self):
    """Show the levels and the terms."""
    return f"OrdinalNode({self.levels}, {self.terms!r})"

__eq__ 

__eq__(other)

Compare levels and terms.

Source code in src/tramdag/spec.py 
985
986
987
988
989
990
991
def __eq__(self, other):
    """Compare levels and terms."""
    return (
        isinstance(other, OrdinalNode)
        and self.levels == other.levels
        and self.terms == other.terms
    )

__hash__ 

__hash__()

Hash what __eq__ compares, so nodes work in sets and as keys.

Source code in src/tramdag/spec.py 
993
994
995
def __hash__(self):
    """Hash what ``__eq__`` compares, so nodes work in sets and as keys."""
    return hash((self.kind, self.levels, tuple(self.terms)))

SI 

SI(**options)

Build the simple-intercept baseline, the paper's SI: I() without parents.

Parameters:

Name Type Description Default
**options

As for Intercept; transform keywords go in transform_kwargs.

 {}

Returns:

Type Description
Intercept

The parentless intercept term.
Source code in src/tramdag/spec.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def SI(**options) -> Intercept:
    """Build the simple-intercept baseline, the paper's SI: ``I()`` without parents.

    Parameters
    ----------
    **options
        As for [`Intercept`][]; transform keywords go in ``transform_kwargs``.

    Returns
    -------
    Intercept
        The parentless intercept term.
    """
    return Intercept(**options)

CI 

CI(*parents, **options)

Build the complex intercept, the paper's CI: I(*parents) with parents.

Parameters:

Name Type Description Default
*parents str

Parent names, at least one.

 ()
**options

As for Intercept.

 {}

Returns:

Type Description
Intercept

The parent-conditioned intercept term.

Raises:

Type Description
ValueError

If no parent is given.
Source code in src/tramdag/spec.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def CI(*parents: str, **options) -> Intercept:
    """Build the complex intercept, the paper's CI: ``I(*parents)`` with parents.

    Parameters
    ----------
    *parents : str
        Parent names, at least one.
    **options
        As for [`Intercept`][].

    Returns
    -------
    Intercept
        The parent-conditioned intercept term.

    Raises
    ------
    ValueError
        If no parent is given.
    """
    if not parents:
        raise ValueError(
            "CI() needs at least one parent; the parentless baseline is SI()"
        )
    return Intercept(*parents, **options)

node_parents 

node_parents(node)

Give the parent names that the terms of a node reference.

Parameters:

Name Type Description Default
node NodeSpec

The node specification.

 required

Returns:

Type Description
list[str]

The parent names, ordered by first appearance, without duplicates.
Source code in src/tramdag/spec.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def node_parents(node: NodeSpec) -> list[str]:
    """Give the parent names that the terms of a node reference.

    Parameters
    ----------
    node : NodeSpec
        The node specification.

    Returns
    -------
    list[str]
        The parent names, ordered by first appearance, without
        duplicates.
    """
    seen: dict[str, None] = {}
    for term in node.terms:
        for p in term.parents:
            seen.setdefault(p, None)
    return list(seen)

validate_and_sort 

validate_and_sort(spec)

Validate the spec and return a topological ordering of the nodes.

Edge ownership: every parent must enter through exactly one edge-owning term. Edge-owning are all parents of I/LS/CS terms and the t of a VC term. VC modifiers are exempt and can repeat across terms.

Parameters:

Name Type Description Default
spec dict[str, NodeSpec]

The DAG specification.

 required

Returns:

Type Description
list[str]

The node names in topological order.

Raises:

Type Description
ValueError

If a parent is unknown, a parent enters through more than one edge-owning term, a VC treatment is unsupported, or the graph has a cycle.
Source code in src/tramdag/spec.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def validate_and_sort(spec: dict[str, NodeSpec]) -> list[str]:
    """Validate the spec and return a topological ordering of the nodes.

    Edge ownership: every parent must enter through exactly one
    edge-owning term. Edge-owning are all parents of I/LS/CS terms and
    the ``t`` of a VC term. VC *modifiers* are exempt and can repeat
    across terms.

    Parameters
    ----------
    spec : dict[str, NodeSpec]
        The DAG specification.

    Returns
    -------
    list[str]
        The node names in topological order.

    Raises
    ------
    ValueError
        If a parent is unknown, a parent enters through more than one
        edge-owning term, a VC treatment is unsupported, or the graph has a
        cycle.
    """
    for name, node in spec.items():
        _check_node(name, node, spec)
    return _kahn_sort(spec)

spec_to_dict 

spec_to_dict(spec)

Give the serialized representation of a spec, for checkpoints.

A term serializes as its term name, its parents and every option it carries, so the result describes the model in full and does not depend on what the defaults happen to be today. A hand-written spec may still name only the options it cares about: spec_from_dict passes them to the term's constructor, which fills in the rest.

The result is JSON- and YAML-safe (tuple options such as units become lists), so a spec round-trips through json/YAML as well as through torch.save, except when a term carries a callable input_transform, which serializes only through pickle (torch.save) and only as a module-level function.

A custom term is written as its import path, module.ClassName, and spec_from_dict imports it from there. The class must be defined at module level; one defined in __main__ (a script, a notebook cell) loads in the same process only.

Parameters:

Name Type Description Default
spec dict[str, NodeSpec]

The DAG specification.

 required

Returns:

Type Description
dict

The serialized spec. spec_from_dict inverts it.
Source code in src/tramdag/spec.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def spec_to_dict(spec: dict[str, NodeSpec]) -> dict:
    """Give the serialized representation of a spec, for checkpoints.

    A term serializes as its term name, its parents and **every** option it
    carries, so the result describes the model in full and does not depend on
    what the defaults happen to be today. A hand-written spec may still name
    only the options it cares about: [`spec_from_dict`][] passes them to the
    term's constructor, which fills in the rest.

    The result is JSON- and YAML-safe (tuple options such as ``units`` become
    lists), so a spec round-trips through ``json``/YAML as well as through
    ``torch.save``, except when a term carries a *callable*
    ``input_transform``, which serializes only through pickle
    (``torch.save``) and only as a module-level function.

    A custom term is written as its import path, ``module.ClassName``, and
    [`spec_from_dict`][] imports it from there. The class must be defined at
    module level; one defined in ``__main__`` (a script, a notebook cell)
    loads in the same process only.

    Parameters
    ----------
    spec : dict[str, NodeSpec]
        The DAG specification.

    Returns
    -------
    dict
        The serialized spec. ``spec_from_dict`` inverts it.
    """
    out = {}
    for name, node in spec.items():
        d = {
            "kind": node.kind,
            "terms": [_serialized(t) for t in node.terms],
        }
        if node.kind == "ordinal":
            d["levels"] = node.levels
        out[name] = d
    return out

spec_from_dict 

spec_from_dict(d, *, strict=False)

Rebuild a spec from its serialized form.

Each term is rebuilt through its class, so a wrong arity fails as the term's own ValueError and a misspelled option key as Python's own TypeError, naming the keyword. An option the entry does not mention takes its constructor default, unless strict is set.

Parameters:

Name Type Description Default
d dict

The serialized spec, as produced by spec_to_dict.

 required
strict bool

Refuse an entry that omits an option that shapes its term, see acting_options. Default False.

 False

Returns:

Type Description
dict[str, NodeSpec]

The node specification, keyed by node name.

Raises:

Type Description
ValueError

If the entry names an unknown term, a term's own checks refuse it, or strict finds an omitted option.
TypeError

If a term does not take an option the entry names.
Source code in src/tramdag/spec.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
def spec_from_dict(d: dict, *, strict: bool = False) -> dict[str, NodeSpec]:
    """Rebuild a spec from its serialized form.

    Each term is rebuilt through its class, so a wrong arity fails as the
    term's own ``ValueError`` and a misspelled option key as Python's own
    ``TypeError``, naming the keyword. An option the entry does not mention
    takes its constructor default, unless ``strict`` is set.

    Parameters
    ----------
    d : dict
        The serialized spec, as produced by [`spec_to_dict`][].
    strict : bool, optional
        Refuse an entry that omits an option that shapes its term, see
        [`acting_options`][tramdag.spec.Term.acting_options]. Default False.

    Returns
    -------
    dict[str, NodeSpec]
        The node specification, keyed by node name.

    Raises
    ------
    ValueError
        If the entry names an unknown term, a term's own checks refuse it, or
        ``strict`` finds an omitted option.
    TypeError
        If a term does not take an option the entry names.
    """
    schema = {
        n: "continuous" if e["kind"] == "continuous" else e.get("levels")
        for n, e in d.items()
    }
    spec: dict[str, NodeSpec] = {}
    for name, entry in d.items():
        terms = []
        for t in entry["terms"]:
            cls = _term_class(t["term"])
            term = cls.from_serialized(tuple(t["parents"]), t["options"])
            if strict:
                _check_written(name, entry["kind"] == "ordinal", schema, t, term)
            terms.append(term)
        if entry["kind"] == "continuous":
            spec[name] = ContinuousNode(terms)
        elif entry["kind"] == "ordinal":
            spec[name] = OrdinalNode(int(entry["levels"]), terms)
        else:
            raise ValueError(f"node {name!r}: unknown kind {entry['kind']!r}")
    return spec