Source code for optiwindnet.terse

# SPDX-License-Identifier: MIT
# https://gitlab.windenergy.dtu.dk/TOPFARM/OptiWindNet/

"""Compact, self-describing encodings of solution and routeset links."""

import math
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from enum import StrEnum, auto
from itertools import pairwise
from typing import Any

import networkx as nx
import numpy as np

from .fingerprint import fingerprint_coordinates
from .types import Topology

__all__ = ('LinkScope', 'TerseLinks')




[docs]
class LinkScope(StrEnum):
    """Graph layer represented by :class:`TerseLinks`."""



[docs]
    TOPOLOGY = auto()




[docs]
    ROUTESET = auto()






@dataclass(frozen=True, slots=True)
class _EdgeSpec:
    u: int
    v: int
    is_open: bool = False


def _compress_ring_routes(
    routes: Sequence[tuple[int, Sequence[int], int]],
) -> tuple[int, ...]:
    """Flatten ``(open_root, walk, close_root)`` ring routes."""
    seq: list[int] = []
    previous_root = None
    last = len(routes) - 1
    for idx, (open_, walk, close) in enumerate(routes):
        if open_ != previous_root:
            seq.append(int(open_))
        seq.extend(int(x) for x in walk)
        if idx != last or close != open_:
            seq.append(int(close))
        previous_root = close
    return tuple(seq)


def _parse_ring_routes(seq: Sequence[int]) -> list[tuple[int, list[int], int]]:
    """Recover ``(open_root, walk, close_root)`` ring routes."""
    routes: list[tuple[int, list[int], int]] = []
    values = [int(x) for x in seq]
    i, length, previous_root = 0, len(values), None
    while i < length:
        if values[i] < 0:
            open_ = values[i]
            i += 1
        elif previous_root is not None:
            open_ = previous_root
        else:
            raise ValueError('ring route does not begin at a root')

        walk: list[int] = []
        while i < length and values[i] >= 0:
            walk.append(values[i])
            i += 1
        if not walk:
            raise ValueError('ring route contains no non-root nodes')
        if i >= length:
            close = open_
        else:
            close = values[i]
            i += 1
        previous_root = close
        routes.append((open_, walk, close))
    return routes


def _walk_ring(G: nx.Graph, root: int, subroot: int) -> tuple[list[int], int]:
    """Walk a ring from one root feeder to the other."""
    walk = [subroot]
    previous, current = root, subroot
    while True:
        neighbours = [node for node in G[current] if node != previous]
        if not neighbours:
            return walk, root
        if len(neighbours) != 1:
            raise ValueError(
                f'ring node {current} is not degree-2 (neighbours {neighbours})'
            )
        (next_,) = neighbours
        if next_ < 0:
            return walk, next_
        walk.append(next_)
        previous, current = current, next_


def _ring_routes_from_graph(
    G: nx.Graph, R: int, T: int
) -> list[tuple[int, list[int], int]]:
    """Extract canonical ring routes from either a topology or routeset graph."""
    routes: list[tuple[int, list[int], int]] = []
    visited_feeders: set[tuple[int, int]] = set()
    for root in range(-R, 0):
        for subroot in sorted(G[root]):
            if (root, subroot) in visited_feeders:
                continue
            walk, close = _walk_ring(G, root, subroot)
            visited_feeders.add((root, walk[0]))
            visited_feeders.add((close, walk[-1]))
            open_ = root

            terminal_positions = [i for i, node in enumerate(walk) if node < T]
            terminal_count = len(terminal_positions)
            if terminal_count == 0:
                raise ValueError('ring route contains no terminals')
            if terminal_count == 1 and close != open_:
                if G[open_][walk[0]].get('load') == 0:
                    open_, close = close, open_
                    walk.reverse()
            elif terminal_count > 1:
                open_boundaries = []
                for boundary, (left, right) in enumerate(pairwise(terminal_positions)):
                    segment = zip(walk[left:right], walk[left + 1 : right + 1])
                    if all(G[u][v].get('load') == 0 for u, v in segment):
                        open_boundaries.append(boundary)
                if len(open_boundaries) != 1:
                    raise ValueError(
                        'ring must have one open terminal-to-terminal link; '
                        f'found {len(open_boundaries)}'
                    )
                if open_boundaries[0] != math.ceil(terminal_count / 2) - 1:
                    walk.reverse()
                    open_, close = close, open_
            routes.append((open_, walk, close))
    return routes


def _parent_links_from_graph(G: nx.Graph, T: int, B: int, clone_count: int):
    targets: list[int | None] = [None] * (T + clone_count)
    for u, v, edge_data in G.edges(data=True):
        reverse = edge_data.get('reverse')
        if reverse is None:
            raise ValueError('reverse must not be None')
        u, v = (u, v) if u < v else (v, u)
        source, target = (u, v) if reverse else (v, u)
        slot = source if source < T else source - B
        if not 0 <= slot < len(targets):
            raise ValueError(f'parent source {source} is outside the encoded node set')
        if targets[slot] is not None:
            raise ValueError(f'node {source} has more than one parent')
        targets[slot] = int(target)
    missing = [i for i, target in enumerate(targets) if target is None]
    if missing:
        raise ValueError(f'nodes without an encoded parent: {missing}')
    return tuple(target for target in targets if target is not None)


@dataclass(frozen=True, slots=True)