Skip to content

math_spec.separability

The walk behind :attr:~math_spec.program.Program.separability — every axis's verdict, in one pass over a program.

separabilities(program) #

Every axis's verdict, in one walk.

One traversal rather than one per axis, because every construct that ties an axis together names the axis it ties: asking each node which dimension it is about answers for all of them at what answering for one cost.

reductions_couple is the position a block stands in rather than anything about the block — a sum over the axis couples a constraint row to the whole horizon and leaves an objective additively separable. A translation reads ahead for a negative offset; what one reads behind is the window's edge, which is not asked. Each coupling carries the one modelling change that would lift it, after the dash.

Source code in src/math_spec/separability.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def separabilities(program: Program) -> dict[str, Separability]:
    """Every axis's verdict, in one walk.

    One traversal rather than one per axis, because every construct that ties an
    axis together names the axis it ties: asking each node *which* dimension it
    is about answers for all of them at what answering for one cost.

    ``reductions_couple`` is the position a block stands in rather than anything
    about the block — a sum over the axis couples a constraint row to the whole
    horizon and leaves an objective additively separable. A translation reads
    ahead for a negative offset; what one reads behind is the window's edge,
    which is not asked. Each coupling carries the one modelling change that
    would lift it, after the dash.
    """
    ahead = dict.fromkeys(program.dimensions, 0)
    reasons: dict[str, dict[str, dict[str, list[str]]]] = {
        kind: {dimension: {} for dimension in program.dimensions} for kind in ('coupled', 'restarts')
    }
    undecided: dict[str, dict[Reach, None]] = {dimension: {} for dimension in program.dimensions}

    def report(kind: str, dimension: str, label: str, reason: str) -> None:
        reasons[kind][dimension].setdefault(label, []).append(reason)

    def waits_on(dimension: str, label: str, name: str, kind: Literal['offset', 'partition', 'coordinate']) -> None:
        undecided[dimension][Reach(label, name, kind)] = None

    for label, nodes, mask, reductions_couple in _built_blocks(program):
        masks: list[Mask | None] = [mask]
        for node in walk(*nodes):
            if isinstance(node, Cases):
                masks.extend(region.when for region in node.regions)
            elif isinstance(node, Sum):
                if reductions_couple:
                    for dimension in node.over:
                        report(
                            'coupled',
                            dimension,
                            label,
                            f'sums over {dimension} — a rolling sum_back(window=n) windows, a total over the horizon does not',
                        )
            elif isinstance(node, GroupSum):
                for dimension in node.over:
                    report(
                        'coupled',
                        dimension,
                        label,
                        f'groups {dimension} into {", ".join(node.into)} — window that dimension instead, or cut only at the group edges',
                    )
            elif isinstance(node, At):
                for dimension in node.into:
                    for relation in node.coordinate:
                        waits_on(dimension, label, relation, 'coordinate')
            elif isinstance(node, (Translate, Window)):
                dimension = node.dimension
                if node.wrap:
                    report(
                        'coupled',
                        dimension,
                        label,
                        f'wraps around {dimension}, so its first row reads its last — an opening-state seed at '
                        f'position({dimension}) == 0 is what a rolling horizon replaces the wrap with',
                    )
                    continue
                if node.partition is not None:
                    waits_on(dimension, label, node.partition.name, 'partition')
                if isinstance(node, Window):
                    continue
                if isinstance(node.offset, str):
                    waits_on(dimension, label, node.offset, 'offset')
                else:
                    ahead[dimension] = max(ahead[dimension], -node.offset)
        for candidate in masks:
            for atom in candidate.atoms if candidate is not None else ():
                if isinstance(atom, DimensionPositionNode):
                    report('restarts', atom.name, label, f'counts a position along {atom.name}')

    for name, block in program.sos.items():
        report(
            'coupled',
            block.over,
            f"set '{name}'",
            f'is a set over {block.over}, which a window would cut — only a window holding every whole set keeps it',
        )

    def joined(kind: str, dimension: str) -> dict[str, str]:
        return {label: ', '.join(dict.fromkeys(found)) for label, found in reasons[kind][dimension].items()}

    return {
        dimension: Separability(
            dimension=dimension,
            ahead=ahead[dimension],
            coupled=joined('coupled', dimension),
            undecided=tuple(undecided[dimension]),
            restarts=joined('restarts', dimension),
        )
        for dimension in program.dimensions
    }