cep_graph.py

# Copyright 2020 Jannis Harder
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
# AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
# LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
# OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
# PERFORMANCE OF THIS SOFTWARE.
#
# SPDX-License-Identifier: 0BSD

'''Read and write networkx Graphs in the PACE 2021 format.

See https://pacechallenge.org/2021/tracks/#appendix-input-format for a
description.
'''

from typing import Union
import networkx as nx
import os


def write_cep(G: nx.Graph, path: Union[str, os.PathLike]):
    '''Write a networkx Graph to a file at the given path.

    This uses the format used in the PACE 2021 challenge. See
    https://pacechallenge.org/2021/tracks/#appendix-input-format for a
    description.

    When the graph's vertices are not numbered 1..n, they are automatically
    relabeled using `G = nx.convert_node_labels_to_integers(G, 1)`.

    The the graph attribute `G.graph['comment']` is written to the output as
    comment.
    '''
    if not all(v in G for v in range(1, 1 + G.number_of_nodes())):
        G = nx.convert_node_labels_to_integers(G, 1)
    with open(path, 'w') as f:
        if 'comment' in G.graph:
            for line in G.graph['comment'].split('\n'):
                f.write(f'c {line}\n')
        f.write(f'p cep {G.number_of_nodes()} {G.number_of_edges()}\n')
        for a, b in G.edges:
            if a == b:
                raise ValueError('loops are forbidden')
            f.write(f'{a} {b}\n')


def read_cep(path: Union[str, os.PathLike]) -> nx.Graph:
    '''Read a networkx Graph from a file at the given path.

    This uses the format used in the PACE 2021 challenge. See
    https://pacechallenge.org/2021/tracks/#appendix-input-format for a
    description.

    Any comment lines are stored in the graph attribute `G.graph['comment']` of
    the resulting graph G.
    '''
    comments = []
    with open(path, 'r') as f:
        G, edge_count = None, None
        for line in f:
            line = line.rstrip('\n')
            if line.startswith('c'):
                comments.append(line[1:].lstrip(' '))
                continue
            parts = line.split(' ')
            if parts and parts[0] == 'p':
                if G is not None:
                    raise ValueError('duplicated problem descriptor')
                if len(parts) != 4 or parts[1] != 'cep':
                    raise ValueError('not a cluster editing problem instance')
                try:
                    vertex_count = int(parts[2])
                except ValueError:
                    vertex_count = -1
                if vertex_count < 0:
                    raise ValueError(f'invalid vertex count `{parts[2]}`')

                try:
                    edge_count = int(parts[3])
                except ValueError:
                    edge_count = -1
                if edge_count < 0:
                    raise ValueError(f'invalid edge count `{parts[3]}`')

                G = nx.Graph()
                G.add_nodes_from(range(1, 1 + vertex_count))
            else:
                if G is None:
                    raise ValueError('missing problem descriptor')
                if len(parts) != 2:
                    raise ValueError(f'expected an edge but got `{line}`')
                edge = []
                for part in parts:
                    try:
                        vertex = int(part)
                    except ValueError:
                        vertex = 0
                    if vertex not in G:
                        raise ValueError(f'invalid vertex `{part}`')
                    edge.append(vertex)
                if edge[0] == edge[1]:
                    raise ValueError(f'loops ({line}) are forbidden')
                if edge in G.edges:
                    raise ValueError(f'parallel edges ({line}) are forbidden')
                G.add_edge(*edge)
        if G is None:
            raise ValueError('missing problem descriptor')
        if G.number_of_edges() != edge_count:
            raise ValueError('wrong number of edges in problem descriptor')
        if comments:
            G.graph['comment'] = '\n'.join(comments)
        return G