Supply-chain API

Used by the runnable scripts in the supply-chain showcase.

The data dictionary defines all summary, detailed-report, and sweep columns, including solver signs and graph scope.

SupplyChainAnalyzer

SupplyChainAnalyzer(
    economy: Economy,
    scenario: Scenario,
    terminal_good: str,
    *,
    tolerance: float = 1e-09,
)

Compile and analyze one nominal single-terminal production scenario.

Parameters:

• economy

  (Economy) –

  Economy containing production, goods, and workforce data.

• scenario

  (Scenario) –

  Scenario with positive terminal exports and an enabled balance floor.

• terminal_good

  (str) –

  Good key used to focus the graph and reports.

• tolerance

  (float, default: 1e-09 ) –

  Positive numerical threshold for active levels and edges.

Raises:

• ValueError –

  If the target, objective, terminal good, or tolerance is invalid for a version-1 nominal analysis.

Methods:

• run –

  Solve once and return an immutable supply-chain result.

Source code in src/vic3_analysis/analysis/supply_chain/analyzer.py

def __init__(
    self,
    economy: Economy,
    scenario: Scenario,
    terminal_good: str,
    *,
    tolerance: float = 1e-9,
) -> None:
    """Configure one nominal analysis without solving it.

    Args:
        economy: Economy containing production, goods, and workforce data.
        scenario: Scenario with positive terminal exports and an enabled balance floor.
        terminal_good: Good key used to focus the graph and reports.
        tolerance: Positive numerical threshold for active levels and edges.

    Raises:
        ValueError: If the target, objective, terminal good, or tolerance is
            invalid for a version-1 nominal analysis.
    """
    if not np.isfinite(tolerance) or tolerance <= 0:
        raise ValueError("tolerance must be a positive finite number.")
    goods_index = economy.goods_index
    if terminal_good not in goods_index:
        raise ValueError(f"Good '{terminal_good}' not found in goods index.")
    if scenario.min_goods_balance is None:
        raise ValueError(
            "SupplyChainAnalyzer requires an enabled min_goods_balance."
        )
    exports = goods_order_vector(goods_index, scenario.goods_export, "goods_export")
    amount = float(exports[goods_index.index(terminal_good)])
    if amount <= 0:
        raise ValueError("Terminal exports must be a positive finite number.")
    if scenario.objective == "gdp_per_capita":
        raise ValueError(
            "SupplyChainAnalyzer v1 supports nominal objectives only; "
            "gdp_per_capita requires MarketOptimizer."
        )
    self._economy = economy
    self._scenario = scenario
    self._terminal_good = terminal_good
    self._target_quantity = amount
    self._tolerance = float(tolerance)

run

run() -> SupplyChainResult

Solve once and return an immutable supply-chain result.

Source code in src/vic3_analysis/analysis/supply_chain/analyzer.py

def run(self) -> SupplyChainResult:
    """Solve once and return an immutable supply-chain result."""
    optimizer = NominalOptimizer(self._economy)
    problem = optimizer.compile(self._scenario)
    state = optimizer.solve_problem(problem)
    solver_result = optimizer.result
    if solver_result is None:
        raise RuntimeError("Nominal optimizer returned no solver result.")

    fun = float(solver_result.fun)
    objective_value = (
        -fun if self._scenario.objective in ("gdp", "employment") else fun
    )
    diagnostics = SolverDiagnostics(
        status=int(solver_result.status),
        message=str(solver_result.message),
        objective_value=objective_value,
        inequality_marginals=_marginals(getattr(solver_result, "ineqlin", None)),
        equality_marginals=_marginals(getattr(solver_result, "eqlin", None)),
    )
    production = self._economy.df_production
    allowed = allowed_process_mask(
        production,
        self._scenario,
        tolerance=self._tolerance,
    )
    active = state.building_levels > self._tolerance
    prices = state.market_prices
    realized_graph = build_supply_graph(
        production=production,
        goods_index=self._economy.goods_index,
        config_index=self._economy.building_index,
        input_matrix=problem.input_matrix,
        output_matrix=problem.output_matrix,
        levels=state.building_levels,
        prices=prices,
        terminal_good=self._terminal_good,
        selected_processes=active,
        allowed_processes=allowed,
        tolerance=self._tolerance,
    )
    allowed_graph = build_supply_graph(
        production=production,
        goods_index=self._economy.goods_index,
        config_index=self._economy.building_index,
        input_matrix=problem.input_matrix,
        output_matrix=problem.output_matrix,
        levels=state.building_levels,
        prices=prices,
        terminal_good=self._terminal_good,
        selected_processes=allowed,
        allowed_processes=allowed,
        tolerance=self._tolerance,
    )
    return SupplyChainResult(
        scenario=self._scenario,
        terminal_good=self._terminal_good,
        target_quantity=self._target_quantity,
        problem=problem,
        state=state,
        diagnostics=diagnostics,
        production=production,
        goods_index=self._economy.goods_index,
        config_index=self._economy.building_index,
        profession_index=self._economy.pop_index,
        employment_matrix=self._economy.employment_matrix,
        employment_vector=self._economy.employment_vector,
        construction_cost_vector=self._economy.construction_cost_vector,
        arable_land_vector=self._economy.arable_land_vector,
        infrastructure_vector=self._economy.infrastructure_vector,
        realized_graph=realized_graph,
        allowed_graph=allowed_graph,
        tolerance=self._tolerance,
    )

SupplyChainResult dataclass

SupplyChainResult(
    scenario: Scenario,
    terminal_good: str,
    target_quantity: float,
    problem: LinearProblem,
    state: EconomyState,
    diagnostics: SolverDiagnostics,
    production: DataFrame,
    goods_index: tuple[str, ...],
    config_index: tuple[str, ...],
    profession_index: tuple[str, ...],
    employment_matrix: ndarray,
    employment_vector: ndarray,
    construction_cost_vector: ndarray,
    arable_land_vector: ndarray,
    infrastructure_vector: ndarray,
    realized_graph: DiGraph,
    allowed_graph: DiGraph,
    tolerance: float = 1e-09,
)

Immutable, reproducible result of one terminal-good analysis.

Returned by:

• API Reference Supply-chain API
  •  SupplyChainAnalyzer  run
  •  SupplyChainSweepResult
    •  results
    •  result_for

Used by:

• API Reference Supply-chain API  SupplyChainSweepResult

Methods:

• graph –

  Return a defensive copy of the requested bipartite graph.

• processes –

  Return process economics and resource attributes for one view.

• goods –

  Return good balances for processes present in one graph view.

• flows –

  Return one row per goods/process edge in one graph view.

• workforce –

  Return realized terminal-chain employment by process and profession.

• constraints –

  Return named LP constraint values, slacks, and solver marginals.

• bottlenecks –

  Return binding inequalities with a useful relaxation marginal.

• alternatives –

  Return scenario-allowed producer alternatives for one or all goods.

• summary –

  Return headline, structural, and aggregate production-unit metrics.

• to_mermaid –

  Return a deterministic Mermaid representation.

• plot_network –

  Return a deterministic weighted network figure.

• plot_contributions –

  Return a process-contribution bar chart.

• plot_goods_balance –

  Return a goods supply-and-demand chart.

• plot_constraints –

  Return a binding-constraint marginal chart.

graph

graph(view: str = 'realized') -> DiGraph

Return a defensive copy of the requested bipartite graph.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def graph(self, view: str = "realized") -> nx.DiGraph:
    """Return a defensive copy of the requested bipartite graph."""
    narrowed = validate_view(view)
    return self._source_graph(narrowed).copy()

processes

processes(view: str = 'realized') -> DataFrame

Return process economics and resource attributes for one view.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def processes(self, view: str = "realized") -> pd.DataFrame:
    """Return process economics and resource attributes for one view."""
    narrowed = validate_view(view)
    return self._process_table(self._rows_for_view(narrowed))

goods

goods(view: str = 'realized') -> DataFrame

Return good balances for processes present in one graph view.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def goods(self, view: str = "realized") -> pd.DataFrame:
    """Return good balances for processes present in one graph view."""
    narrowed = validate_view(view)
    graph = self._source_graph(narrowed)
    rows = self._rows_for_view(narrowed)
    levels = self.state.building_levels
    if rows:
        idx = np.asarray(rows, dtype=np.int64)
        output = levels[idx] @ self.problem.output_matrix[idx]
        inputs = levels[idx] @ self.problem.input_matrix[idx]
    else:
        output = np.zeros(len(self.goods_index), dtype=np.float64)
        inputs = np.zeros(len(self.goods_index), dtype=np.float64)
    present = set(good_keys(graph, self.goods_index))
    total_input_value = float(inputs @ self.state.market_prices)
    rows_out: list[dict[str, object]] = []
    for j, good in enumerate(self.goods_index):
        if good not in present:
            continue
        input_value = float(inputs[j] * self.state.market_prices[j])
        allowed_producers = sum(
            1
            for predecessor in self.allowed_graph.predecessors(good_node_id(good))
            if self.allowed_graph.nodes[predecessor].get("kind") == "process"
        )
        active_producers = sum(
            1
            for predecessor in self.realized_graph.predecessors(good_node_id(good))
            if self.realized_graph.nodes[predecessor].get("kind") == "process"
        )
        sell = float(output[j] + self.state.imports[j])
        buy = float(
            inputs[j] + self.state.exports[j] + self.state.pop_buy_orders[j]
        )
        rows_out.append(
            {
                "good": good,
                "terminal": good == self.terminal_good,
                "price": float(self.state.market_prices[j]),
                "production": float(output[j]),
                "consumption": float(inputs[j]),
                "imports": float(self.state.imports[j]),
                "exports": float(self.state.exports[j]),
                "pop_needs": float(self.state.pop_buy_orders[j]),
                "sell_orders": sell,
                "buy_orders": buy,
                "net_balance": sell - buy,
                "output_value": float(output[j] * self.state.market_prices[j]),
                "input_value": input_value,
                "input_cost_share": _safe_divide(input_value, total_input_value),
                "allowed_producers": allowed_producers,
                "active_producers": active_producers,
            }
        )
    return pd.DataFrame(rows_out, columns=GOOD_COLUMNS)

flows

flows(view: str = 'realized') -> DataFrame

Return one row per goods/process edge in one graph view.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def flows(self, view: str = "realized") -> pd.DataFrame:
    """Return one row per goods/process edge in one graph view."""
    narrowed = validate_view(view)
    graph = self._source_graph(narrowed)
    in_cycle = cyclic_nodes(graph)
    rows: list[dict[str, object]] = []
    for source, target, data in sorted(graph.edges(data=True)):
        rows.append(
            {
                "source": source,
                "target": target,
                "role": str(data["role"]),
                "good": str(data["good"]),
                "process": str(data["process"]),
                "unit_rate": float(data["unit_rate"]),
                "realized_amount": float(data["realized_amount"]),
                "price": float(data["price"]),
                "realized_value": float(data["realized_value"]),
                "active": bool(data["active"]),
                "in_cycle": source in in_cycle and target in in_cycle,
            }
        )
    return pd.DataFrame(rows, columns=FLOW_COLUMNS)

workforce

workforce() -> DataFrame

Return realized terminal-chain employment by process and profession.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def workforce(self) -> pd.DataFrame:
    """Return realized terminal-chain employment by process and profession."""
    rows: list[dict[str, object]] = []
    for i in self._rows_for_view("realized"):
        for j, profession in enumerate(self.profession_index):
            employment = float(self.state.pops[i, j])
            if employment <= self.tolerance:
                continue
            rows.append(
                {
                    "config": self.config_index[i],
                    "building": str(self.production.iloc[i]["building"]),
                    "profession": profession,
                    "employment": employment,
                }
            )
    return pd.DataFrame(rows, columns=WORKFORCE_COLUMNS)

constraints

constraints() -> DataFrame

Return named LP constraint values, slacks, and solver marginals.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def constraints(self) -> pd.DataFrame:
    """Return named LP constraint values, slacks, and solver marginals."""
    levels = self.state.building_levels
    rows: list[dict[str, object]] = []

    def append_blocks(
        kind: Literal["inequality", "equality"],
        matrix: np.ndarray | None,
        bounds: np.ndarray | None,
        slices: ConstraintSlices,
        marginals: np.ndarray | None,
    ) -> None:
        if matrix is None or bounds is None:
            return
        values = matrix @ levels
        for family, row_slice in slices.items():
            start = 0 if row_slice.start is None else row_slice.start
            stop = start if row_slice.stop is None else row_slice.stop
            labels = self._constraint_labels(str(family), stop - start)
            for offset, position in enumerate(range(start, stop)):
                lhs = float(values[position])
                bound = float(bounds[position])
                slack = bound - lhs if kind == "inequality" else float("nan")
                residual = lhs - bound if kind == "equality" else 0.0
                binding = (
                    slack <= self.tolerance
                    if kind == "inequality"
                    else abs(residual) <= self.tolerance
                )
                marginal = (
                    float(marginals[position])
                    if marginals is not None and position < len(marginals)
                    else float("nan")
                )
                rows.append(
                    {
                        "kind": kind,
                        "family": str(family),
                        "label": labels[offset],
                        "realized_value": lhs,
                        "bound": bound,
                        "slack": slack,
                        "residual": residual,
                        "binding": binding,
                        "solver_marginal": marginal,
                        "improvement_per_unit": (
                            max(0.0, -marginal)
                            if kind == "inequality" and np.isfinite(marginal)
                            else float("nan")
                        ),
                    }
                )

    append_blocks(
        "inequality",
        self.problem.A_ub,
        self.problem.b_ub,
        self.problem.inequality_slices,
        self.diagnostics.inequality_marginals,
    )
    append_blocks(
        "equality",
        self.problem.A_eq,
        self.problem.b_eq,
        self.problem.equality_slices,
        self.diagnostics.equality_marginals,
    )
    return pd.DataFrame(rows, columns=CONSTRAINT_COLUMNS)

bottlenecks

bottlenecks() -> DataFrame

Return binding inequalities with a useful relaxation marginal.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def bottlenecks(self) -> pd.DataFrame:
    """Return binding inequalities with a useful relaxation marginal."""
    constraints = self.constraints()
    selected = constraints[
        constraints["kind"].eq("inequality")
        & constraints["binding"].eq(True)
        & constraints["improvement_per_unit"].gt(self.tolerance)
    ].copy()
    order = np.lexsort(
        (
            np.asarray(selected["label"], dtype=str),
            np.asarray(selected["family"], dtype=str),
            -np.asarray(selected["improvement_per_unit"], dtype=np.float64),
        )
    )
    return selected.iloc[order].reset_index(drop=True)

alternatives

alternatives(good: str | None = None) -> DataFrame

Return scenario-allowed producer alternatives for one or all goods.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def alternatives(self, good: str | None = None) -> pd.DataFrame:
    """Return scenario-allowed producer alternatives for one or all goods."""
    if good is not None and good not in self.goods_index:
        raise ValueError(f"Good '{good}' not found in goods index.")
    prices = self.state.market_prices
    rows: list[dict[str, object]] = []
    selected_rows = set(process_rows(self.realized_graph))
    goods = (
        (good,)
        if good is not None
        else good_keys(self.allowed_graph, self.goods_index)
    )
    for output_good in goods:
        good_id = good_node_id(output_good)
        if good_id not in self.allowed_graph:
            continue
        good_position = self.goods_index.index(output_good)
        for process_id in self.allowed_graph.predecessors(good_id):
            data = self.allowed_graph.nodes[process_id]
            if data.get("kind") != "process":
                continue
            i = int(data["row"])
            unit_input = float(self.problem.input_matrix[i] @ prices)
            unit_output = float(self.problem.output_matrix[i] @ prices)
            rows.append(
                {
                    "good": output_good,
                    "config": self.config_index[i],
                    "building": str(self.production.iloc[i]["building"]),
                    "production_method": str(
                        self.production.iloc[i]["production_method"]
                    ),
                    "era": int(self.production.iloc[i]["era"]),
                    "unit_output": float(
                        self.problem.output_matrix[i, good_position]
                    ),
                    "unit_input_cost": unit_input,
                    "unit_output_value": unit_output,
                    "unit_value_added": unit_output - unit_input,
                    "selected": i in selected_rows,
                    "level": float(self.state.building_levels[i]),
                    "parent_group": str(data.get("parent_group", "")),
                    "land_usage": str(data.get("land_usage", "")),
                    "is_extractive": bool(data.get("is_extractive", False)),
                    "is_subsistence": bool(data.get("is_subsistence", False)),
                    "discoverable_resource": bool(
                        data.get("discoverable_resource", False)
                    ),
                    "depletable_resource": bool(
                        data.get("depletable_resource", False)
                    ),
                }
            )
    frame = pd.DataFrame(rows, columns=ALTERNATIVE_COLUMNS)
    if frame.empty:
        return frame
    frame = frame.sort_values("config", kind="mergesort")
    frame = frame.sort_values("unit_output", ascending=False, kind="mergesort")
    frame = frame.sort_values("selected", ascending=False, kind="mergesort")
    return frame.sort_values("good", kind="mergesort").reset_index(drop=True)

summary

summary() -> Series

Return headline, structural, and aggregate production-unit metrics.

Chain fields include whole realized processes and their coproducts; system fields include all solved configurations. Resource columns use full extraction-building keys from the production snapshot, including unused types, and report levels and levels per 10,000 employed workers. Wage metrics require finite coefficients for contributing processes. Known zero denominators produce zero; unavailable wage data produce NaN.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def summary(self) -> pd.Series:
    """Return headline, structural, and aggregate production-unit metrics.

    Chain fields include whole realized processes and their coproducts;
    system fields include all solved configurations. Resource columns use
    full extraction-building keys from the production snapshot, including
    unused types, and report levels and levels per 10,000 employed workers.
    Wage metrics require finite coefficients for contributing processes.
    Known zero denominators produce zero; unavailable wage data produce NaN.
    """
    terminal_position = self.goods_index.index(self.terminal_good)
    terminal_output = float(self.state.building_goods_output[terminal_position])
    terminal_net = float(
        self.state.building_levels @ self.problem.net_matrix[:, terminal_position]
    )
    chain = self.processes("realized")
    system_levels = self.state.building_levels
    system_gdp = self.state.gdp_weekly
    chain_gdp = float(np.asarray(chain["value_added"], dtype=np.float64).sum())
    system_employment = self.state.total_employment
    chain_employment = float(
        np.asarray(chain["employment"], dtype=np.float64).sum()
    )
    system_construction = float(system_levels @ self.construction_cost_vector)
    chain_construction = float(
        np.asarray(chain["construction_cost"], dtype=np.float64).sum()
    )
    system_land = float(system_levels @ self.arable_land_vector)
    chain_land = float(np.asarray(chain["arable_land"], dtype=np.float64).sum())
    system_infrastructure = float(system_levels @ self.infrastructure_vector)
    chain_infrastructure = float(
        np.asarray(chain["net_infrastructure_usage"], dtype=np.float64).sum()
    )
    realized_cycles = cyclic_nodes(self.realized_graph)
    realized_processes = self.processes("realized")
    allowed_processes = self.processes("allowed")
    values: dict[str, object] = {
        "scenario": self.scenario.display_name(),
        "terminal_good": self.terminal_good,
        "objective": self.scenario.objective,
        "solver_status": self.diagnostics.status,
        "solver_message": self.diagnostics.message,
        "objective_value": self.diagnostics.objective_value,
        "target_quantity": self.target_quantity,
        "target_base_value": self.target_quantity
        * float(self.state.market_prices[terminal_position]),
        "terminal_base_price": float(self.state.market_prices[terminal_position]),
        "terminal_gross_output": terminal_output,
        "terminal_net_output": terminal_net,
        "terminal_surplus": (
            terminal_net
            + float(self.state.imports[terminal_position])
            - float(self.state.exports[terminal_position])
            - (self.scenario.min_goods_balance or 0.0)
        ),
        "system_gdp_weekly": system_gdp,
        "system_gdp_annual": system_gdp * 52.0,
        "chain_gdp_weekly": chain_gdp,
        "chain_gdp_annual": chain_gdp * 52.0,
        "system_employment": system_employment,
        "chain_employment": chain_employment,
        "system_construction_cost": system_construction,
        "chain_construction_cost": chain_construction,
        "system_arable_land": system_land,
        "chain_arable_land": chain_land,
        "system_net_infrastructure_usage": system_infrastructure,
        "chain_net_infrastructure_usage": chain_infrastructure,
        "employment_per_terminal_unit": _safe_divide(
            chain_employment, terminal_net
        ),
        "construction_per_terminal_unit": _safe_divide(
            chain_construction, terminal_net
        ),
        "gdp_per_terminal_unit": _safe_divide(chain_gdp, terminal_net),
        "output_per_10k_workers": _safe_divide(
            terminal_net * 10_000.0, chain_employment
        ),
        "realized_process_count": len(realized_processes),
        "realized_good_count": sum(
            data.get("kind") == "good"
            for _node, data in self.realized_graph.nodes(data=True)
        ),
        "realized_edge_count": self.realized_graph.number_of_edges(),
        "allowed_process_count": len(allowed_processes),
        "allowed_good_count": sum(
            data.get("kind") == "good"
            for _node, data in self.allowed_graph.nodes(data=True)
        ),
        "allowed_edge_count": self.allowed_graph.number_of_edges(),
        "cyclic_component_count": cyclic_component_count(self.realized_graph),
        "cyclic_node_count": len(realized_cycles),
        "condensation_depth": condensation_depth(self.realized_graph),
        "extractive_process_count": int(
            np.asarray(realized_processes["is_extractive"], dtype=bool).sum()
        ),
        "resource_process_count": int(
            (
                np.asarray(realized_processes["discoverable_resource"], dtype=bool)
                | np.asarray(realized_processes["depletable_resource"], dtype=bool)
            ).sum()
        ),
        "chain_gdp_share": _safe_divide(chain_gdp, system_gdp),
        "chain_employment_share": _safe_divide(chain_employment, system_employment),
        "chain_construction_share": _safe_divide(
            chain_construction, system_construction
        ),
        "chain_arable_land_share": _safe_divide(chain_land, system_land),
        "chain_infrastructure_share": _safe_divide(
            chain_infrastructure, system_infrastructure
        ),
    }
    chain_levels = np.zeros_like(system_levels)
    chain_rows = list(self._rows_for_view("realized"))
    chain_levels[chain_rows] = system_levels[chain_rows]
    building_keys = self.production["building"].astype(str).to_numpy()
    buildings = resource_buildings(self.production)
    for scope, levels, gdp, employment, construction in (
        (
            "system",
            system_levels,
            system_gdp,
            system_employment,
            system_construction,
        ),
        ("chain", chain_levels, chain_gdp, chain_employment, chain_construction),
    ):
        wage_employment = float("nan")
        if "wage_normalized_employment" in self.production.columns:
            contributing = levels != 0.0
            coefficients = self.production["wage_normalized_employment"].to_numpy(
                dtype=np.float64, na_value=np.nan
            )[contributing]
            if np.isfinite(coefficients).all():
                wage_employment = float(levels[contributing] @ coefficients)
        values[f"{scope}_wage_normalized_employment"] = wage_employment
        for metric, denominator in (
            ("gdp_per_capita", employment / self.state.workforce_ratio),
            ("gdp_per_wage_normalized_employment", wage_employment),
            ("gdp_per_construction_cost", construction),
        ):
            weekly = _safe_divide(gdp, denominator)
            values[f"{scope}_{metric}_weekly"] = weekly
            values[f"{scope}_{metric}_annual"] = weekly * 52.0
        for building in buildings:
            building_levels = float(levels[building_keys == building].sum())
            values[f"{scope}_levels_{building}"] = building_levels
            values[f"{scope}_levels_per_10k_pop_{building}"] = _safe_divide(
                building_levels * 10_000.0, employment
            )
    return pd.Series(
        values,
        index=[*SUMMARY_FIELDS, *resource_level_fields(self.production)],
        name=self.terminal_good,
    )

to_mermaid

to_mermaid(
    view: str = "realized",
    *,
    max_producers_per_good: int = 3,
) -> str

Return a deterministic Mermaid representation.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def to_mermaid(
    self,
    view: str = "realized",
    *,
    max_producers_per_good: int = 3,
) -> str:
    """Return a deterministic Mermaid representation."""
    from vic3_analysis.analysis.supply_chain.visualization import to_mermaid

    return to_mermaid(
        self,
        view=view,
        max_producers_per_good=max_producers_per_good,
    )

plot_network

plot_network(
    view: str = "realized",
    *,
    metric: str = "value",
    max_producers_per_good: int = 3,
) -> Figure

Return a deterministic weighted network figure.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def plot_network(
    self,
    view: str = "realized",
    *,
    metric: str = "value",
    max_producers_per_good: int = 3,
) -> Figure:
    """Return a deterministic weighted network figure."""
    from vic3_analysis.analysis.supply_chain.visualization import plot_network

    return plot_network(
        self,
        view=view,
        metric=metric,
        max_producers_per_good=max_producers_per_good,
    )

plot_contributions

plot_contributions(
    metric: str = "gdp",
    *,
    scope: str = "chain",
    top_n: int = 15,
) -> Figure

Return a process-contribution bar chart.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def plot_contributions(
    self,
    metric: str = "gdp",
    *,
    scope: str = "chain",
    top_n: int = 15,
) -> Figure:
    """Return a process-contribution bar chart."""
    from vic3_analysis.analysis.supply_chain.visualization import (
        plot_contributions,
    )

    return plot_contributions(self, metric=metric, scope=scope, top_n=top_n)

plot_goods_balance

plot_goods_balance(*, top_n: int = 15) -> Figure

Return a goods supply-and-demand chart.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def plot_goods_balance(self, *, top_n: int = 15) -> Figure:
    """Return a goods supply-and-demand chart."""
    from vic3_analysis.analysis.supply_chain.visualization import (
        plot_goods_balance,
    )

    return plot_goods_balance(self, top_n=top_n)

plot_constraints

plot_constraints(*, top_n: int = 15) -> Figure

Return a binding-constraint marginal chart.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def plot_constraints(self, *, top_n: int = 15) -> Figure:
    """Return a binding-constraint marginal chart."""
    from vic3_analysis.analysis.supply_chain.visualization import (
        plot_constraints,
    )

    return plot_constraints(self, top_n=top_n)

SupplyChainSweepResult dataclass

SupplyChainSweepResult(
    _summary: DataFrame,
    _failures: DataFrame,
    _building_levels: DataFrame,
    _results: Mapping[str, SupplyChainResult],
)

Result of applying one scenario template across terminal goods.

Returned by:

• API Reference Supply-chain API  sweep_supply_chains

Methods:

• result_for –

  Return the retained analysis for good.

Attributes:

• summary (DataFrame) –

  Return one fixed-schema row per requested good.

• failures (DataFrame) –

  Return unsuccessful sweep rows.

• df_building_levels (DataFrame) –

  Return whole-system levels by production unit and terminal good.

• results (Mapping[str, SupplyChainResult]) –

  Return retained successful results, empty when not requested.

summary property

summary: DataFrame

Return one fixed-schema row per requested good.

failures property

failures: DataFrame

Return unsuccessful sweep rows.

df_building_levels property

df_building_levels: DataFrame

Return whole-system levels by production unit and terminal good.

Rows follow the economy's production-unit order, indexed by config; columns follow requested goods order. Failed solves contain NaN, while successful unused units retain zero. Available even without retained results; an empty sweep keeps every row with no goods columns.

results property

results: Mapping[str, SupplyChainResult]

Return retained successful results, empty when not requested.

result_for

result_for(good: str) -> SupplyChainResult

Return the retained analysis for good.

Source code in src/vic3_analysis/analysis/supply_chain/result.py

def result_for(self, good: str) -> SupplyChainResult:
    """Return the retained analysis for *good*."""
    if good not in self._results:
        raise KeyError(
            f"No retained result for {good!r}; use keep_results=True and "
            "ensure the analysis succeeds."
        )
    return self._results[good]

sweep_supply_chains

sweep_supply_chains(
    economy: Economy,
    scenario_template: Scenario,
    *,
    goods: Iterable[str] | None = None,
    target_value: float = 100000.0,
    keep_results: bool = False,
    tolerance: float = 1e-09,
) -> SupplyChainSweepResult

Analyze goods with at least the requested base-price export value.

Each terminal export amount is the higher of its aggregated template exports and target_value / base_price. Other exports and all imports are retained. Target metadata reports the effective terminal exports.

Summary columns include fixed economic metrics and per-building resource levels derived from the full economy, including banned and unused types. Empty sweeps retain the same columns; failed solves use NaN for metrics. df_building_levels retains every production unit's solved level for each requested good, independently of keep_results; failures use NaN.

Source code in src/vic3_analysis/analysis/supply_chain/sweep.py

def sweep_supply_chains(
    economy: Economy,
    scenario_template: Scenario,
    *,
    goods: Iterable[str] | None = None,
    target_value: float = 100_000.0,
    keep_results: bool = False,
    tolerance: float = 1e-9,
) -> SupplyChainSweepResult:
    """Analyze goods with at least the requested base-price export value.

    Each terminal export amount is the higher of its aggregated template
    exports and ``target_value / base_price``. Other exports and all imports
    are retained. Target metadata reports the effective terminal exports.

    Summary columns include fixed economic metrics and per-building resource
    levels derived from the full economy, including banned and unused types.
    Empty sweeps retain the same columns; failed solves use NaN for metrics.
    ``df_building_levels`` retains every production unit's solved level for
    each requested good, independently of ``keep_results``; failures use NaN.
    """
    if not np.isfinite(target_value) or target_value <= 0:
        raise ValueError("target_value must be a positive finite number.")
    selected_goods = list(economy.producible_goods) if goods is None else list(goods)
    if len(selected_goods) != len(set(selected_goods)):
        raise ValueError("goods must not contain duplicates.")
    goods_index = economy.goods_index
    unknown = [good for good in selected_goods if good not in goods_index]
    if unknown:
        raise ValueError(f"Goods not found in goods index: {unknown}")

    template_exports = goods_order_vector(
        goods_index, scenario_template.goods_export, "goods_export"
    )
    prices = dict(zip(goods_index, economy.base_prices))
    columns = sweep_columns(economy.df_production)
    rows: list[dict[str, object]] = []
    retained: dict[str, SupplyChainResult] = {}
    building_levels = pd.DataFrame(
        np.nan,
        index=pd.Index(economy.building_index, name="config"),
        columns=selected_goods,
        dtype=np.float64,
    )
    for good in selected_goods:
        price = float(prices[good])
        if not np.isfinite(price) or price <= 0:
            raise ValueError(f"Good '{good}' must have a positive finite base price.")
        amount = max(
            float(template_exports[goods_index.index(good)]),
            float(target_value / price),
        )
        name = (
            f"{scenario_template.name}:{good}"
            if scenario_template.name is not None
            else good
        )
        scenario = replace(
            scenario_template,
            goods_export=tuple(
                (key, value)
                for key, value in scenario_template.goods_export
                if key != good
            )
            + ((good, amount),),
            name=name,
        )
        try:
            result = SupplyChainAnalyzer(
                economy,
                scenario,
                good,
                tolerance=tolerance,
            ).run()
            row: dict[str, object] = {
                str(key): value for key, value in result.summary().items()
            }
            row.update(
                {
                    "good": good,
                    "status": "success",
                    "error_type": "",
                    "error_message": "",
                }
            )
            building_levels[good] = result.state.building_levels
            if keep_results:
                retained[good] = result
        except ValueError as exc:
            row = {field: float("nan") for field in columns}
            row.update(
                {
                    "good": good,
                    "status": "failure",
                    "error_type": type(exc).__name__,
                    "error_message": str(exc),
                    "scenario": name,
                    "terminal_good": good,
                    "objective": scenario.objective,
                    "solver_status": -1,
                    "solver_message": str(exc),
                    "target_quantity": amount,
                    "target_base_value": amount * price,
                    "terminal_base_price": price,
                }
            )
        rows.append(row)

    summary = pd.DataFrame(rows, columns=columns)
    failures = pd.DataFrame(
        summary.loc[summary["status"].eq("failure"), :]
    ).reset_index(drop=True)
    return SupplyChainSweepResult(
        _summary=summary,
        _failures=failures,
        _building_levels=building_levels,
        _results=retained,
    )