diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index e22f5eff..aeb6b27a 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -53,12 +53,6 @@ jobs: with: python-version: ${{ matrix.python-version }} - - name: Set up Java - uses: actions/setup-java@v6.0.1 - with: - distribution: 'temurin' - java-version: 22 - - name: Download package uses: actions/download-artifact@v8 with: @@ -71,5 +65,7 @@ jobs: uv pip install --system "$(ls dist/*.whl)[dev]" - name: Test with pytest + env: + MPLBACKEND: Agg run: | pytest diff --git a/LICENSES/Apache-2.0.txt b/LICENSES/Apache-2.0.txt deleted file mode 100644 index 137069b8..00000000 --- a/LICENSES/Apache-2.0.txt +++ /dev/null @@ -1,73 +0,0 @@ -Apache License -Version 2.0, January 2004 -http://www.apache.org/licenses/ - -TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - -1. Definitions. - -"License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document. - -"Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License. - -"Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity. - -"You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License. - -"Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files. - -"Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types. - -"Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below). - -"Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof. - -"Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution." - -"Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work. - -2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form. - -3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed. - -4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions: - - (a) You must give any other recipients of the Work or Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License. - - You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License. - -5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions. - -6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file. - -7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License. - -8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages. - -9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability. - -END OF TERMS AND CONDITIONS - -APPENDIX: How to apply the Apache License to your work. - -To apply the Apache License to your work, attach the following boilerplate notice, with the fields enclosed by brackets "[]" replaced with your own identifying information. (Don't include the brackets!) The text should be enclosed in the appropriate comment syntax for the file format. We also recommend that a file or class name and description of purpose be included on the same "printed page" as the copyright notice for easier identification within third-party archives. - -Copyright [yyyy] [name of copyright owner] - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - -http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. diff --git a/MANIFEST.in b/MANIFEST.in index 0ea07afa..e2ef371a 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -3,6 +3,4 @@ # SPDX-License-Identifier: MIT include powerplantmatching/package_data/* -include powerplantmatching/package_data/duke_binaries/* include README.md LICENSE -include requirements.yaml diff --git a/README.md b/README.md index 843efbe5..fe5ef8ab 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,6 @@ and/or the current release stored on Zenodo with a release-specific DOI: `powerplantmatching` is released as free software under the [MIT](LICENSES/MIT.txt) license. The default output data [powerplants.csv](powerplants.csv) generated by the package is released under [CC BY 4.0](LICENSES/CC-BY-4.0.txt). -Parts of the repository may be licensed under different licenses, especially dependent package binaries for `duke` being licensed under [Apache 2.0 license](https://github.com/PyPSA/powerplantmatching/tree/master/LICENSES/Apache-2.0.txt). This repository uses the [REUSE](https://reuse.software/) conventions to indicate the licenses that apply to individual files and parts of the repository. For details on the licenses that apply, see the the header information of the respective files and [REUSE.toml](REUSE.toml) for details. diff --git a/REUSE.toml b/REUSE.toml index a0f8a6e3..744de935 100644 --- a/REUSE.toml +++ b/REUSE.toml @@ -32,11 +32,3 @@ path = [ ] SPDX-FileCopyrightText = "Contributors to powerplantmatching " SPDX-License-Identifier = "CC0-1.0" - - -[[annotations]] -path = [ - "powerplantmatching/package_data/duke_binaries/*.jar", -] -SPDX-FileCopyrightText = "Lars Marius Garshol " -SPDX-License-Identifier = "Apache-2.0" diff --git a/docs/api-core.md b/docs/api-core.md index 8f13f3f7..8b2dcc19 100644 --- a/docs/api-core.md +++ b/docs/api-core.md @@ -8,6 +8,6 @@ SPDX-License-Identifier: MIT ::: powerplantmatching.core -::: powerplantmatching.duke +::: powerplantmatching.linkage ::: powerplantmatching.accessor diff --git a/docs/license.md b/docs/license.md index d88059b1..1b4784d8 100644 --- a/docs/license.md +++ b/docs/license.md @@ -8,7 +8,6 @@ SPDX-License-Identifier: MIT `powerplantmatching` is released as free software under the [MIT](LICENSES/MIT.txt) license. The default output data [powerplants.csv](powerplants.csv) generated by the package is released under [CC BY 4.0](LICENSES/CC-BY-4.0.txt). -Parts of the repository may be licensed under different licenses, especially dependent package binaries for `duke` being licensed under [Apache 2.0 license](https://github.com/PyPSA/powerplantmatching/tree/master/LICENSES/Apache-2.0.txt). This repository uses the [REUSE](https://reuse.software/) conventions to indicate the licenses that apply to individual files and parts of the repository. For details on the licenses that apply, see the the header information of the respective files and [REUSE.toml](REUSE.toml) for details. diff --git a/docs/release-notes.md b/docs/release-notes.md index 1f1f48a0..d20355c9 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -8,7 +8,10 @@ SPDX-License-Identifier: MIT ## Upcoming Version -* +* Replace Java DUKE with a blocked Python matcher using rapidfuzz. Java, the bundled jars, XML matching files and `powerplantmatching.duke` are removed. Matching results change. +* Score plant names symmetrically and select fuzzy links with sparse maximum-score one-to-one assignment. Exact EIC matching still runs first and accepts only isolated one-to-one pairs. +* Preserve EICs as sorted unique lists throughout unit aggregation, reduction and cached reload. Missing EICs are `[]`. +* Rename `parallel_duke_processes` to `parallel_processes`. Update custom configuration keys; no compatibility alias is provided. ## [v0.9.0](https://github.com/PyPSA/powerplantmatching/releases/tag/v0.9.0) (3rd October 2026) diff --git a/powerplantmatching/accessor.py b/powerplantmatching/accessor.py index 05d50145..93140dd2 100644 --- a/powerplantmatching/accessor.py +++ b/powerplantmatching/accessor.py @@ -88,12 +88,12 @@ def set_name(self, name): def get_name(self): return self._obj.columns.name - def match_with(self, df, labels=None, config=None, reduced=True, **dukeargs): + def match_with(self, df, labels=None, config=None, reduced=True, **matchargs): from .matching import combine_multiple_datasets, reduce_matched_dataframe from .utils import to_list_if_other dfs = [self._obj] + to_list_if_other(df) - res = combine_multiple_datasets(dfs, labels, config=config, **dukeargs) + res = combine_multiple_datasets(dfs, labels, config=config, **matchargs) if reduced: return res.pipe(reduce_matched_dataframe, config=config) return res diff --git a/powerplantmatching/cleaning.py b/powerplantmatching/cleaning.py index fbe80c74..08aa544b 100644 --- a/powerplantmatching/cleaning.py +++ b/powerplantmatching/cleaning.py @@ -16,8 +16,8 @@ from deprecation import deprecated from .core import PANDAS_V3, get_config, get_obj_if_Acc -from .duke import duke -from .utils import get_name, set_column_name +from .linkage import match +from .utils import collect_eic_codes, get_name, set_column_name logger = logging.getLogger(__name__) @@ -44,7 +44,7 @@ def mode(x): "DateOut": "max", "File": mode, "projectID": set, - "EIC": set, + "EIC": collect_eic_codes, "Duration": "sum", # note this is weighted sum "Volume_Mm3": "sum", "DamHeight_m": "sum", @@ -505,11 +505,11 @@ def aggregate_units( country_query = "Country == @c" query = " and ".join(filter(None, [agg_query, block_query, country_query])) duplicates = pd.concat( - [duke(df.query(query), threads=threads) for c in countries] + [match(df.query(query), threads=threads) for c in countries] ) else: query = " and ".join(filter(None, [agg_query, block_query])) - duplicates = duke(df.query(query) if query else df, threads=threads) + duplicates = match(df.query(query) if query else df, threads=threads) df = cliques(df, duplicates) df = df.groupby("grouped").agg(props_for_groups) diff --git a/powerplantmatching/collection.py b/powerplantmatching/collection.py index a677ae52..9ea230d8 100644 --- a/powerplantmatching/collection.py +++ b/powerplantmatching/collection.py @@ -31,7 +31,7 @@ def collect( update=False, reduced=True, config=None, - **dukeargs, + **matchargs, ): """ Return the collection for a given list of datasets in matched or @@ -47,7 +47,7 @@ def collect( Switch as to return the reduced (True) or matched (False) dataset. config : dict Configuration file of powerplantmatching - **dukeargs : keyword-args for duke + **matchargs : keyword arguments for matching """ from . import data @@ -88,7 +88,7 @@ def df_by_name(name): if update: dfs = parmap(df_by_name, datasets) - matched = combine_multiple_datasets(dfs, datasets, config=config, **dukeargs) + matched = combine_multiple_datasets(dfs, datasets, config=config, **matchargs) ( matched.assign(projectID=lambda df: df.projectID.astype(str)).to_csv( outfn_matched, index_label="id" diff --git a/powerplantmatching/core.py b/powerplantmatching/core.py index 993cfaab..174da8db 100644 --- a/powerplantmatching/core.py +++ b/powerplantmatching/core.py @@ -110,6 +110,11 @@ def get_config(filename=None, **overrides): config.update(yaml.load(f, Loader=yaml.FullLoader)) config.update(overrides) + if "parallel_duke_processes" in config: + raise ValueError( + "Rename parallel_duke_processes to parallel_processes in your configuration" + ) + sha1digest = sha1(cPickle.dumps(overrides)).digest() if len(dict(**overrides)) == 0: config["hash"] = "default" diff --git a/powerplantmatching/duke.py b/powerplantmatching/duke.py deleted file mode 100644 index 104fa6c1..00000000 --- a/powerplantmatching/duke.py +++ /dev/null @@ -1,159 +0,0 @@ -# SPDX-FileCopyrightText: Contributors to powerplantmatching -# -# SPDX-License-Identifier: MIT - -import logging -import os -import shutil -import subprocess as sub -import tempfile - -import numpy as np -import pandas as pd - -from .core import _package_data - -logger = logging.getLogger(__name__) - - -def add_geoposition_for_duke(df): - """ - Returns the same pandas.Dataframe with an additional column "Geoposition" - which concats the latitude and longitude of the powerplant in a string - - """ - if not df.loc[:, ["lat", "lon"]].isnull().all().all(): - return df.assign( - Geoposition=df[["lat", "lon"]] - .apply(lambda s: ",".join(map(str, s)), axis=1) - .replace("nan,nan", np.nan) - ) - else: - return df.assign(Geoposition=np.nan) - - -def duke( - datasets, - labels=["one", "two"], - singlematch=False, - showmatches=False, - keepfiles=False, - showoutput=False, - threads=1, -): - """ - Run duke in different modes (Deduplication or Record Linkage Mode) to - either locate duplicates in one database or find the similar entries in two - different datasets. In RecordLinkagesMode (match two databases) please - set singlematch=True and use best_matches() afterwards - - Parameters - ---------- - - datasets : pd.DataFrame or [pd.DataFrame] - A single dataframe is run in deduplication mode, while multiple ones - are linked - labels : [str], default ['one', 'two'] - Labels for the linked dataframe - singlematch: boolean, default False - Only in Record Linkage Mode. Only report the best match for each entry - of the first named dataset. This does not guarantee a unique match in - the second named dataset. - keepfiles : boolean, default False - If true, do not delete temporary files - """ - - try: - sub.run(["java", "-version"], check=True, capture_output=True) - except sub.CalledProcessError: - err = "Java is not installed or not in the system's PATH. Please install Java and ensure it is in your system's PATH, then try again." - logger.error(err) - raise FileNotFoundError(err) - - dedup = isinstance(datasets, pd.DataFrame) - if dedup: - # Deduplication mode - duke_config = "Deleteduplicates.xml" - datasets = [datasets] - else: - duke_config = "Comparison.xml" - - duke_bin_dir = _package_data("duke_binaries") - - os.environ["CLASSPATH"] = os.pathsep.join( - [os.path.join(duke_bin_dir, r) for r in os.listdir(duke_bin_dir)] - ) - tmpdir = tempfile.mkdtemp() - - try: - shutil.copyfile( - os.path.join(_package_data(duke_config)), os.path.join(tmpdir, "config.xml") - ) - - logger.debug("Comparing files: %s", ", ".join(labels)) - - for n, df in enumerate(datasets): - df = add_geoposition_for_duke(df) - # due to index unity (see https://github.com/larsga/Duke/issues/236) - if n == 1: - shift_by = datasets[0].index.max() + 1 - df.index += shift_by - df.to_csv(os.path.join(tmpdir, f"file{n + 1}.csv"), index_label="id") - if n == 1: - df.index -= shift_by - - args = [ - "java", - "-Dfile.encoding=UTF-8", - "no.priv.garshol.duke.Duke", - "--linkfile=linkfile.txt", - f"--threads={threads}", - ] - if singlematch: - args.append("--singlematch") - if showmatches: - args.append("--showmatches") - stdout = sub.PIPE - else: - stdout = None - args.append("config.xml") - - run = sub.Popen( - args, - stderr=sub.PIPE, - cwd=tmpdir, - stdout=stdout, - universal_newlines=True, - ) - _, stderr = run.communicate() - - if showmatches: - print(_) - - logger.debug(f"Stderr: {stderr}") - if any(word in stderr.lower() for word in ["error", "fehler"]): - raise RuntimeError(f"duke failed: {stderr}") - - if dedup: - return pd.read_csv( - os.path.join(tmpdir, "linkfile.txt"), - encoding="utf-8", - usecols=[1, 2], - names=labels, - ) - else: - res = pd.read_csv( - os.path.join(tmpdir, "linkfile.txt"), - usecols=[1, 2, 3], - names=labels + ["scores"], - ) - res.iloc[:, 1] -= shift_by - res["scores"] = res.scores.astype(float) - return res - - finally: - if keepfiles: - logger.debug(f"Files of the duke run are kept in {tmpdir}") - else: - shutil.rmtree(tmpdir) - logger.debug(f"Files of the duke run have been deleted in {tmpdir}") diff --git a/powerplantmatching/linkage.py b/powerplantmatching/linkage.py new file mode 100644 index 00000000..b3c5ad02 --- /dev/null +++ b/powerplantmatching/linkage.py @@ -0,0 +1,314 @@ +# SPDX-FileCopyrightText: Contributors to powerplantmatching +# +# SPDX-License-Identifier: MIT + +""" +Vectorised record-linkage and deduplication engine. + +``match`` takes a list of two frames for record linkage or a single frame for +deduplication and returns the matched index pairs. Scoring is a Fellegi-Sunter +belief update over a 0.5 prior: every field maps its similarity linearly onto +its ``[low, high]`` probability bounds and updates the running belief. The +comparators are vectorised: symmetric best-token Jaro-Winkler for names, a factorised +q-gram Dice for categorical fields, a min/max ratio for capacity and a haversine +linear falloff for position (5 km cutoff). + +The bounds and thresholds originate in PR #301's GEO/GPD benchmark. Symmetric +name scoring and one-to-one assignment change its results and require separate +dataset validation before claiming the benchmark's precision or recall. +""" + +from collections.abc import Callable, Sequence +from dataclasses import dataclass + +import numpy as np +import pandas as pd +from rapidfuzz import process +from rapidfuzz.distance import JaroWinkler +from scipy.sparse import coo_matrix +from scipy.sparse.csgraph import min_weight_full_bipartite_matching + +GEO_MAX_DISTANCE_M = 5000.0 +BLOCK_CELLS = 2_000_000 +EARTH_RADIUS_M = 6_371_000.0 +UNMATCHED_COST = 2.0 + +Comparison = tuple[np.ndarray, np.ndarray] +Comparator = Callable[[pd.DataFrame, pd.DataFrame, str, int], Comparison] + + +def _strings(frame: pd.DataFrame, column: str) -> tuple[np.ndarray, np.ndarray]: + values = frame[column].fillna("").astype(str).str.lower().to_numpy() + return values, values != "" + + +def _bigrams(s: str) -> set[str]: + return {s[i : i + 2] for i in range(len(s) - 1)} if len(s) > 1 else {s} + + +def _qgram_matrix( + left: pd.DataFrame, right: pd.DataFrame, column: str, threads: int +) -> Comparison: + """Dice coefficient over character bigrams, computed on unique values only.""" + av, present_a = _strings(left, column) + bv, present_b = _strings(right, column) + codes_a, uniq_a = pd.factorize(av) + codes_b, uniq_b = pd.factorize(bv) + grams_a, grams_b = [_bigrams(x) for x in uniq_a], [_bigrams(x) for x in uniq_b] + table = np.empty((len(uniq_a), len(uniq_b))) + for i, ga in enumerate(grams_a): + for j, gb in enumerate(grams_b): + table[i, j] = 2 * len(ga & gb) / (len(ga) + len(gb)) + return table[codes_a[:, None], codes_b[None, :]], present_a[:, None] & present_b[ + None, : + ] + + +def _token_codes(values: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]: + """Pad the per-record token lists into a (records, width) vocabulary index.""" + tokens = pd.Series(values).str.split().explode().dropna() + records = tokens.rename_axis("record").reset_index(name="token").drop_duplicates() + flat_codes, vocabulary = pd.factorize(records.token) + row_indices = records.record.to_numpy(dtype=np.intp) + positions = records.groupby("record").cumcount().to_numpy(dtype=np.intp) + counts = np.bincount(row_indices, minlength=len(values)) + width = max(int(counts.max(initial=0)), 1) + codes = np.full((len(values), width), len(vocabulary), dtype=np.intp) + codes[row_indices, positions] = flat_codes + return codes, counts, np.append(vocabulary, "") + + +def _name_matrix( + left: pd.DataFrame, right: pd.DataFrame, column: str, threads: int +) -> Comparison: + """Use the lower directional token total over the longer token list. + + Character-level ratios cannot resolve unit designators -- ``token_set_ratio`` + scores "Doel 1" against "Doel 4" at 0.83 and "Neurath" against "Neurath F" at + 1.0 -- which merges the units of a station into a single record. Aligning + token by token scores the mismatched designator at 0 instead. + """ + av, present_a = _strings(left, column) + bv, present_b = _strings(right, column) + codes_a, counts_a, vocabulary_a = _token_codes(av) + codes_b, counts_b, vocabulary_b = _token_codes(bv) + tokens = process.cdist( + vocabulary_a, vocabulary_b, scorer=JaroWinkler.similarity, workers=threads + ) + tokens[-1, :] = tokens[:, -1] = 0.0 + forward = _token_totals(tokens, codes_a, counts_a, codes_b) + reverse = _token_totals(tokens.T, codes_b, counts_b, codes_a).T + total = np.minimum(forward, reverse) + width = np.maximum(counts_a[:, None], counts_b[None, :]) + sim = np.divide(total, width, out=np.zeros_like(total), where=width > 0) + return sim, present_a[:, None] & present_b[None, :] + + +def _token_totals( + similarities: np.ndarray, + left_codes: np.ndarray, + left_counts: np.ndarray, + right_codes: np.ndarray, +) -> np.ndarray: + """Sum best token matches in one direction without iterating record pairs.""" + best = np.zeros((similarities.shape[0], len(right_codes))) + for token in range(right_codes.shape[1]): + np.maximum(best, similarities[:, right_codes[:, token]], out=best) + total = np.zeros((len(left_codes), len(right_codes))) + for token in range(left_codes.shape[1]): + total += np.where(left_counts[:, None] > token, best[left_codes[:, token]], 0.0) + return total + + +def _numeric_matrix( + left: pd.DataFrame, right: pd.DataFrame, column: str, threads: int +) -> Comparison: + av = left[column].to_numpy(dtype=float)[:, None] + bv = right[column].to_numpy(dtype=float)[None, :] + lo, hi = np.minimum(av, bv), np.maximum(av, bv) + with np.errstate(invalid="ignore", divide="ignore"): + sim = np.where(hi > 0, lo / hi, 1.0) + return sim, np.isfinite(av) & np.isfinite(bv) & (av >= 0) & (bv >= 0) + + +def _geo_matrix( + left: pd.DataFrame, right: pd.DataFrame, column: str = "geo", threads: int = -1 +) -> Comparison: + """Haversine falloff on ``lat``/``lon``; ``column`` is a label, not a column.""" + la1 = np.radians(left["lat"].to_numpy(dtype=float))[:, None] + lo1 = np.radians(left["lon"].to_numpy(dtype=float))[:, None] + la2 = np.radians(right["lat"].to_numpy(dtype=float))[None, :] + lo2 = np.radians(right["lon"].to_numpy(dtype=float))[None, :] + h = ( + np.sin((la2 - la1) / 2) ** 2 + + np.cos(la1) * np.cos(la2) * np.sin((lo2 - lo1) / 2) ** 2 + ) + dist = 2 * EARTH_RADIUS_M * np.arcsin(np.sqrt(np.clip(h, 0, 1))) + return np.clip(1 - dist / GEO_MAX_DISTANCE_M, 0.0, None), np.isfinite(dist) + + +@dataclass(frozen=True) +class FieldSpec: + column: str + compare: Comparator + low: float + high: float + + +LINKAGE_FIELDS = [ + FieldSpec("Name", _name_matrix, 0.09, 0.99), + FieldSpec("Fueltype", _qgram_matrix, 0.09, 0.7), + FieldSpec("Country", _qgram_matrix, 0.0, 0.53), + FieldSpec("Capacity", _numeric_matrix, 0.3, 0.75), + FieldSpec("geo", _geo_matrix, 0.1, 0.8), +] +LINKAGE_THRESHOLD = 0.85 + +DEDUP_FIELDS = [ + FieldSpec("Name", _name_matrix, 0.09, 0.99), + FieldSpec("Fueltype", _qgram_matrix, 0.05, 0.65), + FieldSpec("Technology", _qgram_matrix, 0.25, 0.51), + FieldSpec("Country", _qgram_matrix, 0.05, 0.51), + FieldSpec("Capacity", _numeric_matrix, 0.49, 0.51), + FieldSpec("geo", _geo_matrix, 0.05, 0.75), +] +DEDUP_THRESHOLD = 0.96 + + +def _accepted_pairs( + left: pd.DataFrame, + right: pd.DataFrame, + fields: Sequence[FieldSpec], + threshold: float, + threads: int, + upper_triangle: bool, +) -> list[tuple[np.ndarray, np.ndarray, np.ndarray]]: + """Fellegi-Sunter belief update per row block, kept below ``BLOCK_CELLS`` cells.""" + rows_per_block = max(1, BLOCK_CELLS // max(len(right), 1)) + column_index = np.arange(len(right)) + blocks = [] + for start in range(0, len(left), rows_per_block): + block = left.iloc[start : start + rows_per_block] + prob = np.full((len(block), len(right)), 0.5) + for spec in fields: + sim, present = spec.compare(block, right, spec.column, threads) + p = np.where(present, spec.low + sim * (spec.high - spec.low), 0.5) + prob = (prob * p) / (prob * p + (1 - prob) * (1 - p)) + keep = prob >= threshold + if upper_triangle: + keep &= column_index[None, :] > (start + np.arange(len(block)))[:, None] + li, ri = np.nonzero(keep) + blocks.append((li + start, ri, prob[li, ri])) + return blocks + + +def _stack(parts: list[np.ndarray], dtype: type) -> np.ndarray: + return np.concatenate(parts) if parts else np.empty(0, dtype=dtype) + + +def select_one_to_one(links: pd.DataFrame) -> pd.DataFrame: + """Maximize total accepted score with sparse assignment and unmatched rows. + + Sort source labels and identifiers before assignment so source reversal and + input row order cannot change tie resolution. Dummy columns allow every row + to remain unmatched without adding absent candidate edges. + """ + if links.empty: + return links.copy() + labels = sorted(links.columns.difference(["scores"])) + if len(labels) != 2: + raise ValueError("One-to-one selection requires two source columns") + candidates = links.groupby(labels, as_index=False, sort=True).scores.max() + scores = candidates.scores.to_numpy(dtype=float) + if not (np.isfinite(scores) & (scores >= 0) & (scores <= 1)).all(): + raise ValueError("Matching scores must be finite and between zero and one") + rows, left_ids = pd.factorize(candidates[labels[0]], sort=True) + cols, right_ids = pd.factorize(candidates[labels[1]], sort=True) + unmatched = np.arange(len(left_ids)) + graph = coo_matrix( + ( + np.concatenate( + [UNMATCHED_COST - scores, np.full(len(left_ids), UNMATCHED_COST)] + ), + ( + np.concatenate([rows, unmatched]), + np.concatenate([cols, len(right_ids) + unmatched]), + ), + ), + shape=(len(left_ids), len(right_ids) + len(left_ids)), + ).tocsr() + chosen_rows, chosen_cols = min_weight_full_bipartite_matching(graph) + matched = chosen_cols < len(right_ids) + pairs = pd.DataFrame( + { + labels[0]: left_ids[chosen_rows[matched]], + labels[1]: right_ids[chosen_cols[matched]], + } + ) + return pairs.merge(candidates, on=labels, validate="one_to_one")[links.columns] + + +def _deduplicate( + df: pd.DataFrame, labels: Sequence[str], threshold: float, threads: int +) -> pd.DataFrame: + blocks = _accepted_pairs( + df, df, DEDUP_FIELDS, threshold, threads, upper_triangle=True + ) + idx = df.index.to_numpy() + a = idx[_stack([li for li, _, _ in blocks], int)] + b = idx[_stack([ri for _, ri, _ in blocks], int)] + return pd.DataFrame( + {labels[0]: np.concatenate([a, b]), labels[1]: np.concatenate([b, a])} + ) + + +def match( + datasets: pd.DataFrame | Sequence[pd.DataFrame], + labels: Sequence[str] = ("one", "two"), + singlematch: bool = False, + threshold: float | None = None, + threads: int = -1, +) -> pd.DataFrame: + """ + Link two datasets or deduplicate one. + + A single frame is deduplicated (returns reciprocal index pairs as + ``cliques()`` requires); a list of two frames is linked and additionally + carries a ``scores`` column. ``singlematch=True`` selects a maximum-weight + one-to-one assignment. ``threshold`` overrides the inherited + acceptance probability (``DEDUP_THRESHOLD`` / ``LINKAGE_THRESHOLD``); + ``threads`` is the rapidfuzz worker count, ``-1`` meaning all cores. + """ + if len(labels) != 2 or labels[0] == labels[1] or "scores" in labels: + raise ValueError( + "Matching requires two distinct source labels other than scores" + ) + frames = [datasets] if isinstance(datasets, pd.DataFrame) else list(datasets) + if any(not frame.index.is_unique for frame in frames): + raise ValueError("Matching requires a unique index in each source") + if threshold is not None and not 0 <= threshold <= 1: + raise ValueError("Matching threshold must be between zero and one") + if isinstance(datasets, pd.DataFrame): + cut = DEDUP_THRESHOLD if threshold is None else threshold + return _deduplicate(datasets, labels, cut, threads) + + left, right = datasets + cut = LINKAGE_THRESHOLD if threshold is None else threshold + empty = left.empty or right.empty + blocks = ( + [] + if empty + else _accepted_pairs( + left, right, LINKAGE_FIELDS, cut, threads, upper_triangle=False + ) + ) + res = pd.DataFrame( + { + labels[0]: left.index.to_numpy()[_stack([li for li, _, _ in blocks], int)], + labels[1]: right.index.to_numpy()[_stack([ri for _, ri, _ in blocks], int)], + "scores": _stack([s for _, _, s in blocks], float), + } + ) + if singlematch: + res = select_one_to_one(res) + return res.reset_index(drop=True) diff --git a/powerplantmatching/matching.py b/powerplantmatching/matching.py index e1933d9f..29dfcd66 100644 --- a/powerplantmatching/matching.py +++ b/powerplantmatching/matching.py @@ -7,6 +7,7 @@ """ import logging +from collections.abc import Hashable, Sequence from itertools import combinations import numpy as np @@ -14,34 +15,88 @@ from .cleaning import clean_technology from .core import get_config, get_obj_if_Acc -from .duke import duke -from .utils import get_name, parmap, read_csv_if_string +from .linkage import match, select_one_to_one +from .utils import collect_eic_codes, get_name, parmap, read_csv_if_string logger = logging.getLogger(__name__) -def best_matches(links): +def _match_by_eic( + df0: pd.DataFrame, df1: pd.DataFrame, labels: Sequence[str] +) -> tuple[pd.DataFrame, set[Hashable], set[Hashable]]: """ - Subsequent to duke() with singlematch=True. Returns reduced list of - matches on the base of the highest score for each duplicated entry. + Deterministic matching of two datasets by EIC (Energy Identification Code). + + Accept only isolated one-to-one links in the shared-code graph. Scheme + identifiers shared by several stations remain available for fuzzy matching. Parameters ---------- - links : pd.DataFrame - Links as returned by duke + df0, df1 : pd.DataFrame + Source dataframes with an 'EIC' column containing scalar or collected codes + (as produced by ``aggregate_units``). + labels : list of str + Two-element list of dataset names for the output columns. + + Returns + ------- + matches : pd.DataFrame + DataFrame with columns ``labels``, containing matched index pairs. + matched_idx0 : set + Indices from df0 that were matched. + matched_idx1 : set + Indices from df1 that were matched. """ - labels = links.columns.difference({"scores"}) + empty: tuple[pd.DataFrame, set[Hashable], set[Hashable]] = ( + pd.DataFrame(columns=list(labels)), + set(), + set(), + ) + + if len(labels) != 2 or labels[0] == labels[1] or "EIC" in labels: + raise ValueError( + "EIC matching requires two distinct source labels other than EIC" + ) + if not df0.index.is_unique or not df1.index.is_unique: + raise ValueError("EIC matching requires a unique index in each source") + if "EIC" not in df0.columns or "EIC" not in df1.columns: + return empty + + def codes(df: pd.DataFrame, label: str) -> pd.DataFrame: + expanded = df["EIC"].explode().dropna() + expanded = expanded[expanded.map(lambda value: isinstance(value, str))] + return expanded[expanded.ne("")].rename_axis(label).reset_index(name="EIC") + + links = pd.merge(codes(df0, labels[0]), codes(df1, labels[1]), on="EIC") + links = links[list(labels)].drop_duplicates() if links.empty: - return pd.DataFrame(columns=labels) - else: - scores = links["scores"].astype(float) - best_idx = scores.groupby(links.iloc[:, 1], sort=False).idxmax() - return links.loc[best_idx, labels].reset_index(drop=True) + return empty + + isolated = links.groupby(labels[0])[labels[1]].transform("size").eq( + 1 + ) & links.groupby(labels[1])[labels[0]].transform("size").eq(1) + matches = links.loc[isolated].reset_index(drop=True) + matched_idx0 = set(matches[labels[0]]) + matched_idx1 = set(matches[labels[1]]) + + logger.info( + "EIC matching: %d deterministic matches between `%s` and `%s`", + len(matches), + labels[0], + labels[1], + ) + return matches, matched_idx0, matched_idx1 -def compare_two_datasets(dfs, labels, country_wise=True, config=None, **dukeargs): + +def best_matches(links: pd.DataFrame) -> pd.DataFrame: + """Select a maximum-score one-to-one assignment of accepted links.""" + return select_one_to_one(links).drop(columns="scores") + + +def compare_two_datasets(dfs, labels, country_wise=True, config=None, **matchargs): """ - Duke-based horizontal match of two databases. Returns the matched + Fuzzy horizontal match of two databases. Returns the matched dataframe including only the matched entries in a multi-indexed pandas.Dataframe. Compares all properties of the given columns ['Name','Fueltype', 'Technology', 'Country', @@ -49,9 +104,7 @@ def compare_two_datasets(dfs, labels, country_wise=True, config=None, **dukeargs powerplant in different two datasets. The match is in one-to-one mode, that is every entry of the initial databases has maximally one link in order to obtain unique entries in the resulting - dataframe. Attention: When aborting this command, the duke - process will still continue in the background, wait until the - process is finished before restarting. + dataframe. Parameters ---------- @@ -66,44 +119,56 @@ def compare_two_datasets(dfs, labels, country_wise=True, config=None, **dukeargs config = get_config() deprecated_args = {"use_saved_matches", "use_saved_aggregation"} - used_deprecated_args = deprecated_args.intersection(dukeargs) + used_deprecated_args = deprecated_args.intersection(matchargs) if used_deprecated_args: for arg in used_deprecated_args: - dukeargs.pop(arg) + matchargs.pop(arg) msg = "The following arguments were deprecated and are being ignored: " logger.warning(msg + f"{used_deprecated_args}") dfs = list(map(read_csv_if_string, dfs)) - if "singlematch" not in dukeargs: - dukeargs["singlematch"] = True + if "singlematch" not in matchargs: + matchargs["singlematch"] = True + # Resolve isolated exact-EIC pairs before fuzzy matching. + eic_matches, matched_idx0, matched_idx1 = _match_by_eic(dfs[0], dfs[1], labels) + + # Remove EIC-matched rows from the fuzzy input + remaining = [ + dfs[0].drop(index=matched_idx0, errors="ignore"), + dfs[1].drop(index=matched_idx1, errors="ignore"), + ] + + # Compare only the residual records. def country_link(dfs, country): # country_selector for both dataframes sel_country_b = [df["Country"] == country for df in dfs] - # only append if country appears in both dataframse + # only append if country appears in both dataframes if all(sel.any() for sel in sel_country_b): - return duke( - [df[sel] for df, sel in zip(dfs, sel_country_b)], labels, **dukeargs + return match( + [df[sel] for df, sel in zip(dfs, sel_country_b)], labels, **matchargs ) else: return pd.DataFrame(columns=[*labels, "scores"]) if country_wise: countries = config["target_countries"] - links = [country_link(dfs, c) for c in countries] + links = [country_link(remaining, c) for c in countries] links = [link for link in links if not link.empty] if links: links = pd.concat(links, ignore_index=True) else: links = pd.DataFrame(columns=[*labels, "scores"]) else: - links = duke(dfs, labels=labels, **dukeargs) + links = match(remaining, labels=labels, **matchargs) if links.empty: - matches = pd.DataFrame(columns=labels) + fuzzy_matches = pd.DataFrame(columns=labels) else: - matches = best_matches(links) + fuzzy_matches = best_matches(links) + # Combine disjoint exact and fuzzy pairs. + matches = pd.concat([eic_matches, fuzzy_matches], ignore_index=True) return matches @@ -118,7 +183,7 @@ def cross_matches(sets_of_pairs, labels=None): ---------- sets_of_pairs : list list of pd.Dataframe's containing only the matches (without - scores), obtained from the linkfile (duke() and + scores), obtained from the linkfile (match() and best_matches()) labels : list of strings list of names of the databases, used for specifying the order @@ -167,10 +232,10 @@ def cross_matches(sets_of_pairs, labels=None): def link_multiple_datasets( - datasets, labels, use_saved_matches=False, config=None, **dukeargs + datasets, labels, use_saved_matches=False, config=None, **matchargs ): """ - Duke-based horizontal match of multiple databases. Returns the + Fuzzy horizontal match of multiple databases. Returns the matching indices of the datasets. Compares all properties of the given columns ['Name','Fueltype', 'Technology', 'Country', 'Capacity','lat', 'lon'] in order to determine the same @@ -197,7 +262,9 @@ def link_multiple_datasets( def comp_dfs(dfs_lbs): logger.info("Comparing data sources `{}` and `{}`".format(*dfs_lbs[2:])) - return compare_two_datasets(dfs_lbs[:2], dfs_lbs[2:], config=config, **dukeargs) + return compare_two_datasets( + dfs_lbs[:2], dfs_lbs[2:], config=config, **matchargs + ) mapargs = [[dfs[c], dfs[d], labels[c], labels[d]] for c, d in combs] all_matches = parmap(comp_dfs, mapargs) @@ -205,9 +272,9 @@ def comp_dfs(dfs_lbs): return cross_matches(all_matches, labels=labels) -def combine_multiple_datasets(datasets, labels=None, config=None, **dukeargs): +def combine_multiple_datasets(datasets, labels=None, config=None, **matchargs): """ - Duke-based horizontal match of multiple databases. Returns the + Fuzzy horizontal match of multiple databases. Returns the matched dataframe including only the matched entries in a multi-indexed pandas.Dataframe. Compares all properties of the given columns ['Name','Fueltype', 'Technology', 'Country', @@ -252,7 +319,7 @@ def combined_dataframe(cross_matches, datasets, config): .reset_index(drop=True) ) - crossmatches = link_multiple_datasets(datasets, labels, config=config, **dukeargs) + crossmatches = link_multiple_datasets(datasets, labels, config=config, **matchargs) return combined_dataframe(crossmatches, datasets, config).reindex( columns=config["target_columns"], level=0 ) @@ -287,12 +354,7 @@ def reduce_matched_dataframe(df, show_orig_names=False, config=None): "DateRetrofit": "max", "DateOut": "max", "projectID": lambda x: dict(x.droplevel(0).dropna()), - "EIC": lambda x: { - v - for val in x.dropna() - for v in (val if isinstance(val, set) else [val]) - if isinstance(v, str) - }, + "EIC": collect_eic_codes, } ) props_for_groups = pd.Series(props_for_groups)[cols].to_dict() diff --git a/powerplantmatching/package_data/Comparison.xml b/powerplantmatching/package_data/Comparison.xml deleted file mode 100644 index cfa11fbe..00000000 --- a/powerplantmatching/package_data/Comparison.xml +++ /dev/null @@ -1,100 +0,0 @@ - - - - - - - - - - - 0.965 - - - ID - - - NAME - no.priv.garshol.duke.comparators.JaroWinklerTokenized - 0.09 - 0.99 - - - FUELTYPE - no.priv.garshol.duke.comparators.QGramComparator - 0.09 - 0.7 - - - COUNTRY - no.priv.garshol.duke.comparators.QGramComparator - 0.0 - 0.53 - - - CAPACITY - no.priv.garshol.duke.comparators.NumericComparator - 0.3 - 0.75 - - - GEOPOSITION - GeoComparator - 0.1 - 0.8 - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/powerplantmatching/package_data/Deleteduplicates.xml b/powerplantmatching/package_data/Deleteduplicates.xml deleted file mode 100644 index b7490fa8..00000000 --- a/powerplantmatching/package_data/Deleteduplicates.xml +++ /dev/null @@ -1,89 +0,0 @@ - - - - - - - - - - - 0.96 - - - ID - - - NAME - no.priv.garshol.duke.comparators.JaroWinklerTokenized - 0.09 - 0.99 - - - FUELTYPE - no.priv.garshol.duke.comparators.QGramComparator - 0.05 - 0.65 - - - TECHNOLOGY - no.priv.garshol.duke.comparators.QGramComparator - 0.25 - 0.51 - - - COUNTRY - no.priv.garshol.duke.comparators.QGramComparator - 0.05 - 0.51 - - - CAPACITY - no.priv.garshol.duke.comparators.NumericComparator - 0.49 - 0.51 - - - GEOPOSITION - GeoComparator - 0.05 - 0.75 - - - - - - - - - - - - - - - - - - - - - - diff --git a/powerplantmatching/package_data/config.yaml b/powerplantmatching/package_data/config.yaml index 290fc401..8ec8dc76 100644 --- a/powerplantmatching/package_data/config.yaml +++ b/powerplantmatching/package_data/config.yaml @@ -54,7 +54,7 @@ fully_included_sources: aggregate_only_matching_sources: - MASTR # the matching process of very small units is not efficient -parallel_duke_processes: false +parallel_processes: false threads_extend_by_non_matched: 16 matched_data_url: https://raw.githubusercontent.com/PyPSA/powerplantmatching/{tag}/powerplants.csv diff --git a/powerplantmatching/package_data/duke_binaries/duke-core-1.3-SNAPSHOT-tests.jar b/powerplantmatching/package_data/duke_binaries/duke-core-1.3-SNAPSHOT-tests.jar deleted file mode 100644 index d39b8e78..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-core-1.3-SNAPSHOT-tests.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-core-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-core-1.3-SNAPSHOT.jar deleted file mode 100644 index b2ce083e..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-core-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-es-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-es-1.3-SNAPSHOT.jar deleted file mode 100644 index bb237b8d..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-es-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-json-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-json-1.3-SNAPSHOT.jar deleted file mode 100644 index dbd239ad..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-json-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-lucene-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-lucene-1.3-SNAPSHOT.jar deleted file mode 100644 index 3bbb42d9..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-lucene-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-mapdb-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-mapdb-1.3-SNAPSHOT.jar deleted file mode 100644 index ebd8104a..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-mapdb-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-mongodb-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-mongodb-1.3-SNAPSHOT.jar deleted file mode 100644 index 9838fb62..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-mongodb-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/package_data/duke_binaries/duke-server-1.3-SNAPSHOT.jar b/powerplantmatching/package_data/duke_binaries/duke-server-1.3-SNAPSHOT.jar deleted file mode 100644 index b294adee..00000000 Binary files a/powerplantmatching/package_data/duke_binaries/duke-server-1.3-SNAPSHOT.jar and /dev/null differ diff --git a/powerplantmatching/utils.py b/powerplantmatching/utils.py index 3ebb8b94..d157834a 100644 --- a/powerplantmatching/utils.py +++ b/powerplantmatching/utils.py @@ -28,6 +28,13 @@ cc = coco.CountryConverter() +def collect_eic_codes(values: pd.Series) -> list[str]: + """Collect nonempty string identifiers from scalar or collection values.""" + codes = values.explode().dropna() + strings = codes[codes.map(lambda value: isinstance(value, str))] + return sorted(strings[strings.ne("")].unique().tolist()) + + def lookup(df, keys=None, by="Country, Fueltype", exclude=None, unit="MW"): """ Returns a lookup table of the dataframe df with rounded numbers. @@ -346,7 +353,7 @@ def parmap(f, arg_list, config=None, threads=None): """ Parallel mapping function. Use this function to parallelly map function f onto arguments in arg_list. The maximum number of parallel threads is - taken from config.yaml:parallel_duke_processes. + taken from config.yaml:parallel_processes. Parameters --------- @@ -364,7 +371,7 @@ def parmap(f, arg_list, config=None, threads=None): config = get_config() if threads is None: - threads = config["parallel_duke_processes"] + threads = config["parallel_processes"] if isinstance(threads, bool): threads = config.get("process_limit", 1) diff --git a/pyproject.toml b/pyproject.toml index b7db4021..35367d92 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ requires-python = ">=3.11" dependencies = [ "numpy", + "rapidfuzz>=3,<4", "scipy", "pandas>=2.2.3,<4; python_version < '3.14'", "pandas>=2.3.3,<4; python_version >= '3.14'", diff --git a/test/test_duke.py b/test/test_duke.py deleted file mode 100644 index d1ea14d5..00000000 --- a/test/test_duke.py +++ /dev/null @@ -1,97 +0,0 @@ -# SPDX-FileCopyrightText: Contributors to powerplantmatching -# -# SPDX-License-Identifier: MIT - -import numpy as np -import pandas as pd - -from powerplantmatching.duke import add_geoposition_for_duke, duke - -TEST_DATA = { - "Name": [ - "Powerplant", - "an hydro powerplant", - " another powerplant with whitespaces", - " Power II coalition", - " Kraftwerk Besonders besonders '2' CHP", - ], - "Fueltype": [ - "", - "Run of-River", - "OCGT", - "Nuclear Power", - "", - ], - "Technology": [ - "Natural Gas", - "Run of-River", - "", - " Nuclear", - "", - ], - "Set": [ - np.nan, - "", - "", - "Powerplant", - "", - ], - "Country": ["", "", "", "", ""], - "Capacity": [120, 40, 80.008, 90.122, 400.2], - "lat": [51.5074, np.nan, 40.7128, 42.8765, 48.2031], - "lon": [-0.1278, np.nan, -74.0060, -60.9807, -1.0221], -} - - -def test_add_geoposition_for_duke(): - df = pd.DataFrame(TEST_DATA) - - expected = pd.DataFrame(TEST_DATA) - expected["Geoposition"] = [ - "51.5074,-0.1278", - np.nan, - "40.7128,-74.006", - "42.8765,-60.9807", - "48.2031,-1.0221", - ] - - output = add_geoposition_for_duke(df) - - pd.testing.assert_frame_equal(output, expected) - - -def test_duke_deduplication_mode(): - df = pd.DataFrame(TEST_DATA) - - expected = pd.DataFrame({"one": [0, 1], "two": [1, 0]}) - - output = duke(df) - - pd.testing.assert_frame_equal(output, expected) - - -def test_duke_record_linkage_mode(): - df1 = pd.DataFrame(TEST_DATA) - - df2 = pd.DataFrame(TEST_DATA) - df2.loc[0, "Name"] = "Plant" - df2.loc[0, "Technology"] = "Gas" - df2.loc[2, "Capacity"] = 44.5 - - expected = pd.DataFrame( - { - "one": {0: 0, 1: 1, 2: 2, 3: 3, 4: 4}, - "two": {0: 1, 1: 1, 2: 2, 3: 3, 4: 4}, - "scores": { - 0: 0.9769736842105264, - 1: 0.9985590778097984, - 2: 0.9992083247849796, - 3: 0.999639379733141, - 4: 0.9991589571068124, - }, - } - ) - - output = duke([df1, df2], labels=["one", "two"], singlematch=False) - - pd.testing.assert_frame_equal(output, expected) diff --git a/test/test_linkage.py b/test/test_linkage.py new file mode 100644 index 00000000..adcb8149 --- /dev/null +++ b/test/test_linkage.py @@ -0,0 +1,349 @@ +# SPDX-FileCopyrightText: Contributors to powerplantmatching +# +# SPDX-License-Identifier: MIT + +import numpy as np +import pandas as pd +import pytest + +from powerplantmatching import linkage as lk + +BASE = { + "Name": "Alpha Power Station", + "Fueltype": "Hard Coal", + "Technology": "Steam Turbine", + "Country": "Germany", + "Capacity": 100.0, + "lat": 51.0, + "lon": 7.0, +} +KM_IN_DEGREES = 1 / 111.19 + + +def record(**overrides: object) -> dict[str, object]: + return {**BASE, **overrides} + + +def frame( + records: list[dict[str, object]], index: list[int] | None = None +) -> pd.DataFrame: + return pd.DataFrame(records, index=index) + + +def scores_of(left: pd.DataFrame, right: pd.DataFrame) -> pd.DataFrame: + return lk.match([left, right], threshold=0.0) + + +@pytest.fixture +def left() -> pd.DataFrame: + return frame( + [ + record(), + record( + Name="Beta Hydro", Fueltype="Hydro", Capacity=40.0, lat=48.0, lon=11.0 + ), + ] + ) + + +@pytest.fixture +def right(left: pd.DataFrame) -> pd.DataFrame: + df = left.copy() + df.loc[0, "Name"] = "Alpha Power Stn" + df.loc[1, "Capacity"] = 41.0 + return df + + +def test_links_near_duplicates_and_reports_scores( + left: pd.DataFrame, right: pd.DataFrame +) -> None: + out = lk.match([left, right], labels=["one", "two"]) + + assert list(out.columns) == ["one", "two", "scores"] + assert out["scores"].between(0, 1).all() + assert set(zip(out["one"], out["two"])) == {(0, 0), (1, 1)} + + +@pytest.mark.parametrize( + "overrides, linked", + [ + pytest.param({"Name": "Alpha Power Stn"}, True, id="near-duplicate-name"), + pytest.param( + { + "Name": "Gamma Nuclear", + "Fueltype": "Nuclear", + "Country": "France", + "Capacity": 900.0, + "lat": 45.0, + "lon": 2.0, + }, + False, + id="unrelated-record", + ), + pytest.param( + {"Name": "Alpha Works", "Fueltype": "Nuclear", "Country": "France"}, + False, + id="fueltype-and-country-differ", + ), + pytest.param( + {"Name": "Delta Works", "Capacity": 5000.0}, + False, + id="different-name-and-50x-capacity-at-identical-position", + ), + pytest.param( + {"Name": "Alpha Works", "lat": 51.0 + 10 * KM_IN_DEGREES}, + False, + id="10-km-apart", + ), + ], +) +def test_dissimilar_records_are_rejected( + overrides: dict[str, object], linked: bool +) -> None: + out = lk.match([frame([record()]), frame([record(**overrides)])]) + + assert out.empty != linked + + +@pytest.mark.parametrize( + "first, second", + [("Kozloduy 1", "Kozloduy 5"), ("Doel 1", "Doel 4"), ("Neurath", "Neurath F")], +) +def test_units_of_one_station_stay_separate(first: str, second: str) -> None: + """Units share site, fueltype and technology -- only the designator separates them.""" + unit = record(Name=first, Capacity=440.0) + other = record(Name=second, Capacity=1040.0) + + assert lk.match(frame([unit, other], index=[0, 1])).empty + + +def test_linkage_prefers_the_matching_unit() -> None: + """Across sources the 1:1 reduction, not the threshold, resolves sibling units.""" + left = frame([record(Name="Kozloduy 1", Capacity=440.0)]) + right = frame( + [ + record(Name="Kozloduy 5", Capacity=1040.0), + record(Name="Kozloduy 1", Capacity=440.0), + ] + ) + + out = lk.match([left, right], singlematch=True) + + assert list(out["two"]) == [1] + + +def test_name_matching_ignores_word_order() -> None: + """Reordered names must still link, even 10 km apart -- token set, not plain ratio.""" + reordered = record(Name="Station Alpha Power", lat=51.0 + 10 * KM_IN_DEGREES) + + out = lk.match([frame([record()]), frame([reordered])]) + + assert len(out) == 1 + + +@pytest.mark.parametrize( + "counterpart", [np.nan, "", "Hard Coal"], ids=["nan", "empty", "present"] +) +def test_missing_categorical_is_neutral_not_a_bonus(counterpart: object) -> None: + """A field missing on one side must not score better or worse than any counterpart.""" + reference = scores_of( + frame([record(Fueltype=np.nan)]), frame([record(Name="Alpha Power")]) + ) + + out = scores_of( + frame([record(Fueltype=np.nan)]), + frame([record(Name="Alpha Power", Fueltype=counterpart)]), + ) + + assert out["scores"].to_list() == reference["scores"].to_list() + + +@pytest.mark.parametrize( + "missing", + [{"lat": np.nan, "lon": np.nan}, {"Capacity": np.nan}], + ids=["geo", "capacity"], +) +def test_missing_numeric_fields_do_not_block_a_link(missing: dict[str, float]) -> None: + out = lk.match( + [frame([record(**missing)]), frame([record(Name="Alpha Power", **missing)])] + ) + + assert len(out) == 1 + + +@pytest.mark.parametrize("singlematch, expected", [(True, 1), (False, 2)]) +def test_singlematch_keeps_only_the_best_candidate( + singlematch: bool, expected: int +) -> None: + left = frame([record()]) + right = frame( + [record(Capacity=130.0), record(Name="Alpha Power Stn")], index=[7, 8] + ) + + out = lk.match([left, right], singlematch=singlematch) + + assert len(out) == expected + assert out.loc[out["scores"].idxmax(), "two"] == 7 + + +def test_linkage_returns_index_labels_not_positions() -> None: + left = frame( + [record(), record(Name="Beta Hydro", Fueltype="Hydro", lat=48.0, lon=11.0)], + index=[10, 20], + ) + right = frame( + [record(Name="Beta Hydro", Fueltype="Hydro", lat=48.0, lon=11.0), record()], + index=[7, 8], + ) + + out = lk.match([left, right]) + + assert set(zip(out["one"], out["two"])) == {(10, 8), (20, 7)} + + +def test_dedup_returns_index_labels_not_positions() -> None: + df = frame( + [ + record(), + record(Name="Alpha Power Stn"), + record(Name="Beta Hydro", Fueltype="Hydro", lat=48.0, lon=11.0), + ], + index=[10, 20, 30], + ) + + out = lk.match(df) + + assert set(zip(out["one"], out["two"])) == {(10, 20), (20, 10)} + + +@pytest.mark.parametrize( + "kwargs", [{"showmatches": True}, {"singlmatch": True}, {"n_jobs": 2}] +) +def test_unknown_keyword_fails_fast( + left: pd.DataFrame, kwargs: dict[str, object] +) -> None: + with pytest.raises(TypeError): + lk.match(left, **kwargs) + + +def test_geo_contributes_nothing_beyond_the_cutoff() -> None: + near = scores_of( + frame([record()]), frame([record(lat=51.0 + 4 * KM_IN_DEGREES)]) + ).at[0, "scores"] + beyond = scores_of( + frame([record()]), frame([record(lat=51.0 + 5.1 * KM_IN_DEGREES)]) + ).at[0, "scores"] + far = scores_of( + frame([record()]), frame([record(lat=51.0 + 50 * KM_IN_DEGREES)]) + ).at[0, "scores"] + + assert beyond == far + assert beyond < near + + +def test_dedup_returns_reciprocal_pairs_for_duplicates_only() -> None: + df = frame( + [ + record(), + record(), + record( + Name="Beta Hydro", + Fueltype="Hydro", + Technology="Run-Of-River", + Capacity=40.0, + lat=48.0, + lon=11.0, + ), + ] + ) + + out = lk.match(df, labels=["one", "two"]) + + assert list(out.columns) == ["one", "two"] + forward = set(zip(out["one"], out["two"])) + assert forward == {(0, 1), (1, 0)} + + +def test_dedup_ignores_technology_that_is_missing_on_both_sides() -> None: + """aggregate_units fills missing strings with "" -- that must not count as agreement.""" + blank = frame( + [ + record(Technology=""), + record( + Name="Delta Works", + Technology="", + Capacity=5000.0, + lat=51.0 + 10 * KM_IN_DEGREES, + ), + ] + ) + + out = lk.match(blank) + + assert out.empty + + +@pytest.mark.parametrize("block_cells", [1, 3, 2_000_000]) +def test_results_are_invariant_to_row_blocking( + monkeypatch: pytest.MonkeyPatch, block_cells: int +) -> None: + df = frame( + [ + record(), + record(Name="Alpha Power Stn"), + record(Name="Alpha Power Station"), + record(Name="Beta Hydro", Fueltype="Hydro", lat=48.0, lon=11.0), + ], + index=[10, 20, 30, 40], + ) + dedup_reference, linkage_reference = lk.match(df), lk.match([df, df.iloc[::-1]]) + + monkeypatch.setattr(lk, "BLOCK_CELLS", block_cells) + + assert lk.match(df).equals(dedup_reference) + assert lk.match([df, df.iloc[::-1]]).equals(linkage_reference) + + +def test_empty_input_returns_empty_links(left: pd.DataFrame) -> None: + out = lk.match([left, left.iloc[0:0]], labels=["one", "two"]) + + assert out.empty + assert list(out.columns) == ["one", "two", "scores"] + + +def test_scores_and_acceptance_are_symmetric() -> None: + left = frame([record(Name="Alpha Alpho", lat=53.0)]) + right = frame([record(Name="Alpha Power", lat=53.028)]) + forward = scores_of(left, right) + reverse = scores_of(right, left) + np.testing.assert_allclose(forward.scores, reverse.scores, rtol=0, atol=1e-12) + assert lk.match([left, right]).empty == lk.match([right, left]).empty + + +def test_singlematch_uses_global_one_to_one_assignment() -> None: + left = frame([record(), record(Name="Alpha Power Stn")], index=[10, 20]) + right = frame([record(), record(Name="Alpha Station Power")], index=[30, 40]) + forward = lk.match([left, right], labels=["A", "B"], singlematch=True) + reverse = lk.match( + [right.iloc[::-1], left.iloc[::-1]], labels=["B", "A"], singlematch=True + ) + assert forward.A.is_unique and forward.B.is_unique + assert len(forward) == 2 + assert set(map(tuple, forward[["A", "B"]].to_numpy())) == set( + map(tuple, reverse[["A", "B"]].to_numpy()) + ) + + +def test_deduplication_does_not_depend_on_row_order() -> None: + units = frame( + [record(Name="Alpha Alpho"), record(Name="Alpha Power")], index=[10, 20] + ) + forward = lk.match(units) + reverse = lk.match(units.iloc[::-1]) + assert set(map(tuple, forward.to_numpy())) == set(map(tuple, reverse.to_numpy())) + + +def test_duplicate_index_cannot_collapse_fuzzy_matches() -> None: + units = frame([record(), record(Name="Beta Plant")], index=[0, 0]) + with pytest.raises(ValueError, match="unique index"): + lk.match([units, units], singlematch=True) diff --git a/test/test_matching.py b/test/test_matching.py new file mode 100644 index 00000000..0c14d180 --- /dev/null +++ b/test/test_matching.py @@ -0,0 +1,258 @@ +# SPDX-FileCopyrightText: Contributors to powerplantmatching +# +# SPDX-License-Identifier: MIT + +from io import StringIO + +import numpy as np +import pandas as pd +import pytest + +from powerplantmatching.cleaning import aggregate_units +from powerplantmatching.core import get_config +from powerplantmatching.matching import ( + _match_by_eic, + best_matches, + compare_two_datasets, + reduce_matched_dataframe, +) +from powerplantmatching.utils import parse_string_to_dict + + +@pytest.fixture +def df_entsoe(): + """ENTSOE-like dataset with EIC codes as sets.""" + return pd.DataFrame( + { + "Name": ["Eemshavencentrale", "Eemscentrale", "Maasvlakte"], + "Fueltype": ["Hard Coal", "Natural Gas", "Hard Coal"], + "Country": ["Netherlands", "Netherlands", "Netherlands"], + "Capacity": [1560.0, 2200.0, 1040.0], + "EIC": [ + {"49W000000000EMSA"}, + {"49W00000000008xG", "49W00000000008xK"}, + {"49W000000000MVSQ"}, + ], + "lat": [53.44, 53.44, 51.95], + "lon": [6.83, 6.84, 4.03], + } + ) + + +@pytest.fixture +def df_opsd(): + """OPSD-like dataset with EIC codes as sets.""" + return pd.DataFrame( + { + "Name": ["Eemshaven coal", "Eems gas", "Rijnmond"], + "Fueltype": ["Hard Coal", "Natural Gas", "Natural Gas"], + "Country": ["Netherlands", "Netherlands", "Netherlands"], + "Capacity": [1560.0, 2200.0, 800.0], + "EIC": [ + {"49W000000000EMSA"}, + {"49W00000000008xG"}, + set(), # Rijnmond has no EIC + ], + "lat": [53.44, 53.44, 51.88], + "lon": [6.83, 6.84, 4.50], + } + ) + + +def test_eic_matching_basic(df_entsoe, df_opsd): + """EIC matching correctly pairs plants sharing EIC codes.""" + labels = ["ENTSOE", "OPSD"] + matches, idx0, idx1 = _match_by_eic(df_entsoe, df_opsd, labels) + + # Eemshavencentrale (0) ↔ Eemshaven coal (0) via EMSA + # Eemscentrale (1) ↔ Eems gas (1) via 008xG + assert len(matches) == 2 + assert set(idx0) == {0, 1} + assert set(idx1) == {0, 1} + + # Maasvlakte (2) and Rijnmond (2) should NOT match (no shared EIC) + assert 2 not in idx0 + assert 2 not in idx1 + + +def test_eic_matching_no_eic_column(): + """Gracefully handles datasets without EIC column.""" + df0 = pd.DataFrame({"Name": ["Plant A"], "Capacity": [100]}) + df1 = pd.DataFrame({"Name": ["Plant B"], "Capacity": [200], "EIC": [{"CODE1"}]}) + + matches, idx0, idx1 = _match_by_eic(df0, df1, ["A", "B"]) + assert matches.empty + assert len(idx0) == 0 + + +def test_eic_matching_empty_sets(): + """No matches when all EIC sets are empty.""" + df0 = pd.DataFrame({"Name": ["A"], "EIC": [set()]}) + df1 = pd.DataFrame({"Name": ["B"], "EIC": [set()]}) + + matches, _, _ = _match_by_eic(df0, df1, ["X", "Y"]) + assert matches.empty + + +def test_eic_matching_nan_values(): + """Float nan inside EIC sets does not produce false matches.""" + df0 = pd.DataFrame({"Name": ["A", "B"], "EIC": [{np.nan}, {"CODE1"}]}) + df1 = pd.DataFrame({"Name": ["X", "Y"], "EIC": [{np.nan}, {"CODE1"}]}) + + matches, idx0, idx1 = _match_by_eic(df0, df1, ["L", "R"]) + # Only CODE1 should match, not nan + assert len(matches) == 1 + assert 0 not in idx0 # row with {nan} not matched + + +def test_eic_matching_nan_only(): + """All-NaN EIC column produces no matches.""" + df0 = pd.DataFrame({"Name": ["A"], "EIC": [None]}) + df1 = pd.DataFrame({"Name": ["B"], "EIC": [None]}) + + matches, _, _ = _match_by_eic(df0, df1, ["X", "Y"]) + assert matches.empty + + +def test_eic_matching_one_to_one(): + """A shared scheme identifier cannot select a station arbitrarily.""" + # Plant A has {C1, C2}; Plant X has {C1}, Plant Y has {C2} + df0 = pd.DataFrame({"Name": ["Plant A"], "EIC": [{"C1", "C2"}]}) + df1 = pd.DataFrame({"Name": ["Plant X", "Plant Y"], "EIC": [{"C1"}, {"C2"}]}) + + matches, idx0, idx1 = _match_by_eic(df0, df1, ["src0", "src1"]) + + assert matches.empty + assert not idx0 and not idx1 + + +def test_eic_matching_non_set_values(): + """Scalar identifiers participate in the same matching as collections.""" + df0 = pd.DataFrame({"Name": ["A", "B"], "EIC": ["CODE1", {"CODE2"}]}) + df1 = pd.DataFrame({"Name": ["X", "Y"], "EIC": [{"CODE1"}, {"CODE2"}]}) + + matches, idx0, idx1 = _match_by_eic(df0, df1, ["L", "R"]) + assert len(matches) == 2 + assert idx0 == idx1 == {0, 1} + + +@pytest.mark.parametrize("codes", [["C2", "C1", "C1"], {"C1", "C2"}, "C1"]) +def test_reduce_preserves_identifiers_across_sources(codes: object) -> None: + """A higher-priority source without EICs cannot erase known identifiers.""" + columns = pd.MultiIndex.from_product( + [["Name", "Fueltype", "Technology", "Set", "EIC"], ["ENTSOE", "GEM"]] + ) + frame = pd.DataFrame( + [ + [ + "Plant", + "Plant", + "Natural Gas", + "Natural Gas", + "CCGT", + "CCGT", + "PP", + "PP", + codes, + None, + ] + ], + columns=columns, + ) + config = { + "target_columns": ["Name", "Fueltype", "Technology", "Set", "EIC"], + "ENTSOE": {"reliability_score": 5}, + "GEM": {"reliability_score": 6}, + } + result = reduce_matched_dataframe(frame, config=config) + assert result.EIC.iloc[0] == (["C1"] if isinstance(codes, str) else ["C1", "C2"]) + + +def test_eic_matching_lists_and_ambiguous_scheme() -> None: + left = pd.DataFrame({"EIC": [["C1", "C2"], ["SCHEME"]]}, index=[10, 20]) + right = pd.DataFrame({"EIC": [["C1"], ["SCHEME"], ["SCHEME"]]}, index=[30, 40, 50]) + matches, idx0, idx1 = _match_by_eic(left, right, ["A", "B"]) + assert matches.to_dict("records") == [{"A": 10, "B": 30}] + assert idx0 == {10} and idx1 == {30} + + +def test_eic_matching_rejects_duplicate_index() -> None: + left = pd.DataFrame({"EIC": [["C1"], ["C2"]]}, index=[0, 0]) + right = pd.DataFrame({"EIC": [["C1"]]}) + with pytest.raises(ValueError, match="unique index"): + _match_by_eic(left, right, ["A", "B"]) + + +@pytest.mark.parametrize( + "codes, expected", + [(["C2", "C1", "C1", None, np.nan, "", 42], ["C1", "C2"]), (None, [])], +) +def test_aggregate_identifiers_survive_cache_roundtrip( + codes: object, expected: list[str] +) -> None: + """Unit aggregation and cached reload retain the same usable identifiers.""" + config = get_config() + units = pd.DataFrame( + [ + { + "Name": "Alpha Plant", + "Fueltype": "Natural Gas", + "Technology": "CCGT", + "Set": "PP", + "Country": "Netherlands", + "Capacity": 100.0, + "lat": 53.0, + "lon": 6.0, + "EIC": codes, + "projectID": "unit-1", + } + ] + ).reindex(columns=config["target_columns"]) + aggregated = aggregate_units(units, dataset_name="test", config=config) + assert aggregated.EIC.iloc[0] == expected + cached = pd.read_csv(StringIO(aggregated.to_csv(index=False))) + restored = parse_string_to_dict(cached, ["EIC"]) + assert restored.EIC.iloc[0] == expected + + +def test_fuzzy_selection_maximizes_total_score() -> None: + links = pd.DataFrame( + { + "A": [10, 10, 20, 20], + "B": [30, 40, 30, 40], + "scores": [0.99, 0.98, 0.97, 0.86], + } + ) + expected = {(10, 40), (20, 30)} + assert set(map(tuple, best_matches(links).to_numpy())) == expected + reversed_links = links.iloc[::-1][["B", "A", "scores"]] + assert ( + set(map(tuple, best_matches(reversed_links)[["A", "B"]].to_numpy())) == expected + ) + + +def test_eic_pairs_are_removed_before_python_fuzzy_matching() -> None: + base = { + "Fueltype": "Natural Gas", + "Technology": "CCGT", + "Country": "Netherlands", + "Capacity": 100.0, + "lat": 53.0, + "lon": 6.0, + } + left = pd.DataFrame( + [ + {**base, "Name": "Alpha", "EIC": ["C1"]}, + {**base, "Name": "Beta Plant", "EIC": []}, + ], + index=[10, 20], + ) + right = pd.DataFrame( + [ + {**base, "Name": "Unrelated Registry Name", "lat": 51.0, "EIC": ["C1"]}, + {**base, "Name": "Beta Plant", "EIC": []}, + ], + index=[30, 40], + ) + result = compare_two_datasets([left, right], ["A", "B"]) + assert set(map(tuple, result[["A", "B"]].to_numpy())) == {(10, 30), (20, 40)}