diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index bb9f95d5..32f9d5fd 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@v5 - with: - distribution: 'temurin' - java-version: 22 - - name: Download package uses: actions/download-artifact@v8 with: diff --git a/.gitignore b/.gitignore index e71eeaf1..b7d5d6c2 100644 --- a/.gitignore +++ b/.gitignore @@ -87,6 +87,7 @@ ENV/ # Rope project settings .ropeproject analysis/figures/ +analysis/data/ doc/_build doc/api 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..9f7261c3 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -3,6 +3,5 @@ # 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/analysis/benchmark_linkage.py b/analysis/benchmark_linkage.py new file mode 100644 index 00000000..eab51ba8 --- /dev/null +++ b/analysis/benchmark_linkage.py @@ -0,0 +1,307 @@ +# SPDX-FileCopyrightText: Contributors to powerplantmatching +# +# SPDX-License-Identifier: MIT + +""" +Tuning harness for ``powerplantmatching.linkage`` record linkage. + +Ground truth: the GEO<->GPD pairs encoded in the ``projectID`` column of the +production ``powerplants.csv``. Per-field similarity matrices are computed once +per country and cached, so a (low, high, threshold) configuration is scored by +pure arithmetic on the cached matrices. +""" + +import ast +import pickle +import tempfile +import time +from dataclasses import asdict, dataclass +from pathlib import Path + +import numpy as np +import pandas as pd +from rapidfuzz import fuzz, process + +import powerplantmatching as pm +from powerplantmatching.cleaning import clean_name +from powerplantmatching.linkage import ( + GEO_MAX_DISTANCE_M, + LINKAGE_FIELDS, + LINKAGE_THRESHOLD, +) +from powerplantmatching.linkage import _name_matrix as engine_name_matrix + +REPO = Path(__file__).resolve().parent.parent +POWERPLANTS_CSV = REPO / "powerplants.csv" +CACHE_DIR = Path(tempfile.gettempdir()) / "powerplantmatching_benchmark" +CACHE_FILE = CACHE_DIR / "benchmark_linkage_cache.pkl" + +NAME_SCORERS = { + "token_set_ratio": fuzz.token_set_ratio, + "token_sort_ratio": fuzz.token_sort_ratio, + "WRatio": fuzz.WRatio, +} +DEFAULT_NAME_MATRIX = "jw_tokenized" +QGRAM_FIELDS = ["Fueltype", "Country"] +STR_COLUMNS = ["Name", "Fueltype", "Technology", "Set", "Country"] + +Fields = list[tuple[str, float, float]] +BASELINE_FIELDS: Fields = [(f.column, f.low, f.high) for f in LINKAGE_FIELDS] + + +@dataclass +class CountryCache: + country: str + left_ids: np.ndarray + right_ids: np.ndarray + names: dict[str, np.ndarray] + sims: dict[str, np.ndarray] + present: dict[str, np.ndarray] + + +def load_ground_truth(path: Path = POWERPLANTS_CSV) -> set[tuple[str, str]]: + """GEO<->GPD projectID pairs that ended in the same production cluster.""" + ids = pd.read_csv(path, usecols=["projectID"]).projectID.map(ast.literal_eval) + pairs = set() + for entry in ids: + for left in entry.get("GEO", ()): + for right in entry.get("GPD", ()): + pairs.add((left, right)) + return pairs + + +def _preprocess(df: pd.DataFrame, config: dict) -> pd.DataFrame: + str_cols = [c for c in STR_COLUMNS if c in df.columns] + out = df.assign( + lat=df.lat.astype(float), + lon=df.lon.astype(float), + **df[str_cols].fillna("").astype(str), + ) + out = clean_name(out, config=config) + with pd.option_context("future.no_silent_downcasting", True): + out[str_cols] = out[str_cols].replace("", pd.NA).infer_objects(copy=False) + return out.reset_index(drop=True) + + +def load_frames(config: dict) -> tuple[pd.DataFrame, pd.DataFrame]: + """GEO and GPD with their ``matching_sources`` queries and production preprocessing.""" + queries = { + k: v + for source in config["matching_sources"] + if isinstance(source, dict) + for k, v in source.items() + } + geo = pm.data.GEO(update=False, config=config).query(queries["GEO"]) + gpd = pm.data.GPD(update=False, config=config).query(queries["GPD"]) + return _preprocess(geo, config), _preprocess(gpd, config) + + +def _qgram_matrix(a: pd.Series, b: pd.Series) -> np.ndarray: + """Dice coefficient over character bigrams, computed on unique values only.""" + codes_a, uniq_a = pd.factorize(a.fillna("").astype(str).str.lower()) + codes_b, uniq_b = pd.factorize(b.fillna("").astype(str).str.lower()) + + def grams(s: str) -> set[str]: + return {s[i : i + 2] for i in range(len(s) - 1)} if len(s) > 1 else {s} + + bg_a, bg_b = [grams(x) for x in uniq_a], [grams(x) for x in uniq_b] + table = np.empty((len(uniq_a), len(uniq_b)), dtype=np.float32) + for i, ga in enumerate(bg_a): + for j, gb in enumerate(bg_b): + table[i, j] = 2 * len(ga & gb) / (len(ga) + len(gb)) if (ga or gb) else 0.0 + return table[codes_a[:, None], codes_b[None, :]] + + +def _numeric_matrix(a: pd.Series, b: pd.Series) -> np.ndarray: + av = a.to_numpy(dtype=float)[:, None] + bv = b.to_numpy(dtype=float)[None, :] + lo, hi = np.minimum(av, bv), np.maximum(av, bv) + with np.errstate(invalid="ignore", divide="ignore"): + return np.where(hi > 0, lo / hi, 1.0).astype(np.float32) + + +def _geo_matrix( + left: pd.DataFrame, right: pd.DataFrame +) -> tuple[np.ndarray, np.ndarray]: + r = 6371000.0 + 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 * r * np.arcsin(np.sqrt(np.clip(h, 0, 1))) + sim = np.clip(1 - dist / GEO_MAX_DISTANCE_M, 0.0, None) + return sim.astype(np.float32), ~np.isnan(dist) + + +def _name_matrix(a: pd.Series, b: pd.Series, scorer) -> np.ndarray: + av = a.fillna("").astype(str).str.lower().to_numpy() + bv = b.fillna("").astype(str).str.lower().to_numpy() + return (process.cdist(av, bv, scorer=scorer, workers=-1) / 100.0).astype(np.float32) + + +def _jw_tokenized_matrix(a: pd.Series, b: pd.Series) -> np.ndarray: + """The production comparator, so the harness tunes what the engine runs.""" + sim, _ = engine_name_matrix(a.to_frame("Name"), b.to_frame("Name"), "Name", -1) + return sim.astype(np.float32) + + +def _present(a: pd.Series, b: pd.Series) -> np.ndarray: + return a.notna().to_numpy()[:, None] & b.notna().to_numpy()[None, :] + + +def _build_country_cache( + country: str, left: pd.DataFrame, right: pd.DataFrame +) -> CountryCache: + names = { + key: _name_matrix(left.Name, right.Name, scorer) + for key, scorer in NAME_SCORERS.items() + } + names[DEFAULT_NAME_MATRIX] = _jw_tokenized_matrix(left.Name, right.Name) + sims = {col: _qgram_matrix(left[col], right[col]) for col in QGRAM_FIELDS} + present = {col: _present(left[col], right[col]) for col in QGRAM_FIELDS} + sims["Name"], present["Name"] = ( + names[DEFAULT_NAME_MATRIX], + _present(left.Name, right.Name), + ) + sims["Capacity"], present["Capacity"] = ( + _numeric_matrix(left.Capacity, right.Capacity), + _present(left.Capacity, right.Capacity), + ) + sims["geo"], present["geo"] = _geo_matrix(left, right) + return CountryCache( + country, + left.projectID.to_numpy(), + right.projectID.to_numpy(), + names, + sims, + present, + ) + + +@dataclass +class Harness: + caches: list[CountryCache] + ground_truth: set[tuple[str, str]] + + def scores( + self, + cache: CountryCache, + fields: Fields, + name_scorer: str, + geo_curve: tuple[float, float], + ) -> np.ndarray: + log_odds = np.zeros(cache.sims["geo"].shape, dtype=np.float32) + for column, low, high in fields: + sim = cache.names[name_scorer] if column == "Name" else cache.sims[column] + if column == "geo" and geo_curve != (1.0, 0.0): + sim = np.clip(geo_curve[0] * sim + geo_curve[1], 0.0, 1.0) + p = np.clip(low + sim * (high - low), 1e-12, 1 - 1e-12) + log_odds += np.where( + cache.present[column], np.log(p / (1 - p)), 0.0 + ).astype(np.float32) + return log_odds + + def predict( + self, + fields: Fields = BASELINE_FIELDS, + threshold: float = LINKAGE_THRESHOLD, + name_scorer: str = DEFAULT_NAME_MATRIX, + geo_curve: tuple[float, float] = (1.0, 0.0), + ) -> tuple[set[tuple[str, str]], set[tuple[str, str]]]: + """Return the (raw, 1:1 reduced) predicted projectID pairs.""" + cut = np.log(threshold / (1 - threshold)) + raw, reduced = set(), set() + for cache in self.caches: + log_odds = self.scores(cache, fields, name_scorer, geo_curve) + li, ri = np.nonzero(log_odds >= cut) + if not len(li): + continue + raw.update(zip(cache.left_ids[li], cache.right_ids[ri])) + best_r = log_odds.argmax(axis=1) + best_s = log_odds[np.arange(len(best_r)), best_r] + kept_l = np.nonzero(best_s >= cut)[0] + kept_r, kept_s = best_r[kept_l], best_s[kept_l] + order = np.lexsort((-kept_s, kept_r)) + first = np.concatenate([[True], np.diff(kept_r[order]) != 0]) + winners = order[first] + reduced.update( + zip(cache.left_ids[kept_l[winners]], cache.right_ids[kept_r[winners]]) + ) + return raw, reduced + + def evaluate( + self, + fields: Fields = BASELINE_FIELDS, + threshold: float = LINKAGE_THRESHOLD, + **kwargs, + ) -> dict[str, float]: + raw, reduced = self.predict(fields, threshold, **kwargs) + truth = self.ground_truth + hits = len(reduced & truth) + precision = hits / len(reduced) if reduced else 0.0 + recall = hits / len(truth) + f1 = 2 * precision * recall / (precision + recall) if hits else 0.0 + return { + "precision": precision, + "recall": recall, + "f1": f1, + "pre_reduction_recall": len(raw & truth) / len(truth), + "n_predicted": len(reduced), + "n_raw": len(raw), + } + + +def prepare(config: dict | None = None, use_disk_cache: bool = True) -> Harness: + """Load data, build (or restore) the per-country similarity caches.""" + if use_disk_cache and CACHE_FILE.exists(): + with CACHE_FILE.open("rb") as f: + caches = [CountryCache(**entry) for entry in pickle.load(f)] + return Harness(caches, load_ground_truth()) + + config = config or pm.get_config() + left, right = load_frames(config) + caches = [] + for country in config["target_countries"]: + sub_left, sub_right = ( + left[left.Country == country], + right[right.Country == country], + ) + if sub_left.empty or sub_right.empty: + continue + caches.append(_build_country_cache(country, sub_left, sub_right)) + if use_disk_cache: + CACHE_DIR.mkdir(parents=True, exist_ok=True) + with CACHE_FILE.open("wb") as f: + pickle.dump([asdict(cache) for cache in caches], f) + return Harness(caches, load_ground_truth()) + + +def main() -> None: + t0 = time.perf_counter() + harness = prepare() + print(f"prepared in {time.perf_counter() - t0:.1f} s") + print(f"countries: {len(harness.caches)} {[c.country for c in harness.caches]}") + print(f"candidate pairs: {sum(c.sims['geo'].size for c in harness.caches):,}") + print(f"ground truth pairs: {len(harness.ground_truth)}") + + harness.evaluate() + t0 = time.perf_counter() + repeats = 5 + for _ in range(repeats): + result = harness.evaluate() + print(f"eval cost: {(time.perf_counter() - t0) / repeats * 1000:.0f} ms/config") + print("baseline:", {k: round(v, 3) for k, v in result.items()}) + + for scorer in [DEFAULT_NAME_MATRIX, *NAME_SCORERS]: + res = harness.evaluate(name_scorer=scorer) + print( + f" name scorer {scorer:17s} F1 {res['f1']:.3f} (P {res['precision']:.3f} / R {res['recall']:.3f})" + ) + + +if __name__ == "__main__": + main() diff --git a/analysis/compare-with-entsoe-stats.py b/analysis/compare-with-entsoe-stats.py index 0e67aa77..ba0fd195 100644 --- a/analysis/compare-with-entsoe-stats.py +++ b/analysis/compare-with-entsoe-stats.py @@ -8,82 +8,107 @@ import country_converter as cc import matplotlib.pyplot as plt -import numpy as np import pandas as pd -from bs4 import XMLParsedAsHTMLWarning -from entsoe import EntsoePandasClient +from bs4 import BeautifulSoup, XMLParsedAsHTMLWarning +from entsoe import EntsoeRawClient +from entsoe.mappings import PSRTYPE_MAPPINGS import powerplantmatching as pm from powerplantmatching.cleaning import gather_fueltype_info -warnings.simplefilter(action="ignore", category=(FutureWarning, XMLParsedAsHTMLWarning)) +warnings.simplefilter("ignore", category=FutureWarning) +warnings.simplefilter("ignore", category=XMLParsedAsHTMLWarning) + root = pathlib.Path(__file__).parent.absolute() figpath = root / "figures" +statspath = root / "data" -# whether to update the used dataset or use the precalculated data -UPDATE = False +# reference year of the ENTSO-E installed capacity statistics +YEAR = 2025 +# whether to rerun the matching or use the locally built dataset +UPDATE = False +# whether to requery the ENTSO-E statistics; the cache is fetched when missing +UPDATE_STATS = False config = pm.get_config() -powerplants = pm.powerplants(update=UPDATE, from_url=not UPDATE) - -powerplants = powerplants.powerplant.convert_country_to_alpha2() +def query_installed_capacity(client: EntsoeRawClient, country: str, year: int): + """Installed generation capacity per production type in MW. -client = EntsoePandasClient(api_key=config["entsoe_token"]) - -start = pd.Timestamp("20220101", tz="Europe/Berlin") -end = pd.Timestamp("20230101", tz="Europe/Berlin") - -kwargs = dict(start=start, end=end, psr_type=None) + entsoe-py's own parser drops yearly-resolution documents whose period does + not start at midnight UTC (all CET/EET zones), so the XML is read directly. + """ + start = pd.Timestamp(f"{year}0101", tz="UTC") + end = pd.Timestamp(f"{year + 1}0101", tz="UTC") + xml = client.query_installed_generation_capacity( + country, start=start, end=end, psr_type=None + ) + soup = BeautifulSoup(xml, "html.parser") + capacities = {} + for timeseries in soup.find_all("timeseries"): + fueltype = PSRTYPE_MAPPINGS[timeseries.find("psrtype").text] + capacities[fueltype] = float(timeseries.find("point").find("quantity").text) + return pd.Series(capacities, dtype=float) -def parse(c): +def query_statistics(countries: list[str], year: int) -> pd.DataFrame: + client = EntsoeRawClient(api_key=config["entsoe_token"]) rename = {"GB": "UK"} - for n in range(2): - try: - print(c, n) - return client.query_installed_generation_capacity( - rename.get(c, c), **kwargs - ).iloc[0] - except Exception as e: - print(f"Country {c} failed with {repr(e)}") - time.sleep(3) - return np.nan - - -stats = pd.DataFrame({c: parse(c) for c in powerplants.Country.unique()}) + capacities = {} + for c in countries: + for attempt in range(2): + try: + capacities[c] = query_installed_capacity(client, rename.get(c, c), year) + break + except Exception as e: + print(f"Country {c} failed with {repr(e)}") + time.sleep(3) + return pd.DataFrame(capacities) + + +powerplants = pm.powerplants(update=UPDATE).powerplant.convert_country_to_alpha2() + +statsfile = statspath / f"entsoe-installed-capacity-{YEAR}.csv" +if UPDATE_STATS or not statsfile.exists(): + stats = query_statistics(sorted(powerplants.Country.unique()), YEAR) + statspath.mkdir(parents=True, exist_ok=True) + stats.to_csv(statsfile) +else: + stats = pd.read_csv(statsfile, index_col=0) + fueltypes = gather_fueltype_info(pd.DataFrame({"Fueltype": stats.index}), ["Fueltype"]) stats = stats.groupby(fueltypes.Fueltype.values).sum().unstack() # Manual correction on the statistics - +# ENTSO-E reports no Swiss hydro capacity # https://de.wikipedia.org/wiki/Liste_von_Wasserkraftwerken_in_der_Schweiz?oldformat=true stats.loc["CH", "Hydro"] = 17038 # %% -query = "(DateOut > 2022 or DateOut != DateOut) and (DateIn < 2023 or DateIn != DateIn)" +query = f"(DateOut > {YEAR} or DateOut != DateOut) and (DateIn < {YEAR + 1} or DateIn != DateIn)" + + +def lookup(df): + # ENTSO-E reports biogas within its single Biomass category + df = df.assign(Fueltype=df.Fueltype.replace("Biogas", "Solid Biomass")) + return df.powerplant.lookup().fillna(0) + + powerplants = powerplants.query(query) -totals = powerplants.powerplant.lookup().fillna(0) +totals = lookup(powerplants) sources = [s if isinstance(s, str) else list(s)[0] for s in config["matching_sources"]] input_dbs = {} for s in sources: print(s.title()) - input_dbs[s.title()] = ( - getattr(pm.data, s)() - .powerplant.convert_country_to_alpha2() - .query(query) - .powerplant.lookup() - .fillna(0) - ) + db = getattr(pm.data, s)().powerplant.convert_country_to_alpha2().query(query) + input_dbs[s.title()] = lookup(db) output_dbs = { - s.title(): powerplants[ - powerplants.projectID.apply(lambda ds: s in ds) - ].powerplant.lookup() + s.title(): lookup(powerplants[powerplants.projectID.apply(lambda ds: s in ds)]) for s in sources } @@ -107,9 +132,9 @@ def parse(c): diff[s.title() + " (%)"] = ds.fillna(0) diff = diff[out_compare.Statistics != 0] -diff = diff.loc[ - :, list(set(out_compare.index.unique(1)) - {"Biogas", "Wind", "Solar"}), : -] +# Wind and Solar are not matched but extended separately, "Other" pools +# incomparable residual categories (ENTSO-E puts batteries and marine in there) +diff = diff[~diff.index.get_level_values(1).isin(["Wind", "Solar", "Other"])] diff.index = diff.index.get_level_values(0) + " " + diff.index.get_level_values(1) df = (diff[diff.Difference > 1]).sort_values("Difference", ascending=False) diff --git a/analysis/linkage_findings.md b/analysis/linkage_findings.md new file mode 100644 index 00000000..702bae1e --- /dev/null +++ b/analysis/linkage_findings.md @@ -0,0 +1,182 @@ + + +# Why the matching engine replaced DUKE + +`powerplantmatching` historically matched records with [DUKE](https://github.com/larsga/Duke), +a Java/JVM record-linkage engine invoked as a subprocess and configured via XML. +It was replaced by `powerplantmatching.linkage` — a pure-Python, vectorised +backend (`rapidfuzz` + `numpy`) — which needs no JVM and is an order of magnitude +faster. This note records the evidence, and is explicit about what the new engine +is *not* and about which of its numbers are unverified. + +## This is not a reimplementation of DUKE + +`linkage` keeps DUKE's overall shape — per-field similarity mapped onto +`[low, high]` probability bounds, combined by a Fellegi-Sunter belief update over +a 0.5 prior — but the scoring function is **different**. Comparing against +historical DUKE results, these divergences matter: + +| | DUKE | `linkage` | +|---|---|---| +| similarity → probability | `sim >= 0.5 ? 0.5 + (high-0.5)*sim² : low` (quadratic, hard floor below 0.5) | `low + sim*(high-low)` (linear ramp) | +| geoposition | `(1 - d/maxdist)*0.5 + 0.5` in `[0.5, 1]`, 0 beyond `maxdist` — i.e. always positive evidence | `clip(1 - d/maxdist, 0, None)` in `[0, 1]` | +| q-gram | overlap coefficient, `|A∩B| / min(|A|,|B|)` | Dice, `2|A∩B| / (|A|+|B|)` | +| numeric, both values 0 | 1.0 | 1.0 (matched after the port initially returned 0.0) | + +Measured on the GEO×GPD harness, ~10 % of accepted linkage pairs differ between +DUKE's scoring and this one. Neither the field weights nor the thresholds +therefore transfer, and none of them are inherited on faith: see below. + +Name comparison went the other way: `rapidfuzz.fuzz.token_set_ratio` was tried +first and had to be abandoned. No character-level ratio can resolve a unit +designator — *"Doel 1"* vs *"Doel 4"* differs in one character out of six — +and `token_set_ratio` additionally returns exactly 100 whenever one token set is a +subset of the other, so *"Neurath"* vs *"Neurath F"* is a perfect match. Both +failures merge the units of a station into a single record; see "Deduplication". +The engine therefore aligns token by token like DUKE's `JaroWinklerTokenized`, +scoring the mismatched designator at 0: + +| | `token_set_ratio` | mean best-token Jaro-Winkler | +|---|---|---| +| Kozloduy 1 / Kozloduy 5 | 0.90 | 0.50 | +| Neurath / Neurath F | 1.00 | 0.50 | +| Revin Pump 1 / Revin | 1.00 | 0.33 | +| Gravelines 1 / Gravelins 1 (typo) | 0.96 | 0.99 | + +The similarity is the mean over one record's tokens of the best Jaro-Winkler +match in the other, divided by the longer token count. Reducing over the shared +token vocabulary rather than over record pairs keeps it as fast as the character +kernel it replaces (6.5 s for a 6000-row dedup, unchanged). + +## Ground truth + +The production `powerplants.csv` records which source IDs ended in the same final +cluster. From its `projectID` column we recover **567 GEO↔GPD pairs** as a +cross-source linkage ground truth (`analysis/benchmark_linkage.py`). No dedup +target is built from it — see "Deduplication" below for why. + +It is not correctness: it is DUKE-era pipeline output. Agreement with it measures +agreement with DUKE-era clusters, so any *precision* cost reported below is an +upper bound. + +## Re-tuning the thresholds + +Because the scoring curve changed, the inherited constants had no justification. +They were re-tuned against the ground truth with a held-out protocol: the 24 +harness countries were sorted by ground-truth pair count (descending, ties broken +alphabetically) and dealt alternately into two folds of 12 — fold A the even +positions (337 pairs), fold B the odd (230) — then the threshold was fitted on +one fold by sweeping 0.50–0.995 in steps of 0.005 and evaluated on the other. + +- tune on A → `t* = 0.73` → held-out B **F1 0.845** vs 0.799 at the old 0.965 +- tune on B → `t* = 0.86` → held-out A **F1 0.907** vs 0.875 at the old 0.965 +- pooled held-out (each fold scored at the threshold fitted on the *other*): + **F1 0.881 / P 0.915 / R 0.850**, against 0.965's F1 0.844 / P 0.934 / R 0.771 + +Both directions agree in sign and magnitude. The gain is pure recall +(0.771 → 0.847) for ~1.4 pp of precision. Shipped threshold is **0.85**: it lies +inside both folds' plateaus (0.75–0.86 within 0.003 F1), near its +higher-precision end, and is the more conservative move away from 0.965 — which +sat on the edge of the one genuine cliff. Field bounds were left as they are; no re-weighting +transferred across folds. + +Full-harness confirmation at the shipped settings (all 567 pairs): + +| threshold | precision | recall | F1 | +|---|---|---|---| +| 0.965 (old) | 0.934 | 0.771 | 0.844 | +| **0.85 (shipped)** | 0.920 | **0.847** | **0.882** | + +Swapping the name comparator moves this operating point (F1 0.856, P 0.948 / +R 0.780 at the same 0.85) and the sweep was repeated: the pooled optimum shifts to +0.675 for +0.008 F1, the two folds fit 0.675 and 0.605, and every held-out gain is +under 0.01 — inside the noise of 567 pairs. **0.85 is kept**, at the +higher-precision end of a flat region, because over-merging is the failure mode +with real cost (below). Lowering the Capacity bound to make a size mismatch veto a +merge was searched and rejected: F1 falls monotonically (0.856 at `low = 0.30`, +0.832 at 0.15) on both folds. DUKE's quadratic curve was re-tested against the new +comparator too, and remains worse than the linear ramp (F1 0.819 vs 0.864). + +Note the direction of the bias: this ground truth is DUKE-era output, and the new +comparator is the DUKE-like one, so its scores here flatter it. The independent +check is the statistics comparison below. + +## Earlier DUKE-vs-`linkage` comparison — historical, unreproduced + +The original decision to drop DUKE was taken on an earlier evaluation setup that +reported: + +| backend | precision | recall | F1 | time | +|---|---|---|---|---| +| DUKE | 0.349 | 0.348 | 0.348 | 7.0 s | +| `linkage` | 0.389 | 0.400 | 0.394 | 0.4 s | + +**These numbers could not be reproduced.** `analysis/benchmark_linkage.py` — which +is bit-exact with the production path (identical pairs to +`compare_two_datasets`) — puts the same shipped config at P 0.934 / R 0.771 / +F1 0.844 on the same 567 pairs, and no production-faithful variant tried +(dropping the `matching_sources` queries, dropping preprocessing, row-level +rather than pair-level scoring) lands anywhere near 0.39. The "12 countries" the +original run cites matches nothing here either: 24 countries have rows in both +frames and 20 carry ground-truth pairs. + +The DUKE row was never re-measured on this harness. (DUKE itself is still +runnable: `git worktree` at the merge-base restores the jars, which is how the +baseline dataset under "Deduplication" was built.) Treat this +table, and the derived claims that once accompanied it (~17× speedup, raw recall +0.75, "98 % of true pairs lie within 5 km", capacity median ratio 0.44), as +historical and unverified. The speed advantage is not in doubt; the quality +comparison is. + +## Deduplication — validated against statistics, not against pairs + +A naive scoreboard against the intra-GPD clusters in `powerplants.csv` *looked* +like a DUKE win (DUKE F1 0.79 at recall 1.00 vs new-engine F1 0.71), but that +ground truth is **circular**: production used DUKE for dedup, so the clusters are +DUKE's own output and its recall is 1.00 by construction. No non-circular dedup +ground truth exists, so the dedup weights and threshold **could not be validated** +and were left at their inherited values. `DEDUP_FIELDS` and `DEDUP_THRESHOLD` are +byte-identical to the deleted `Deleteduplicates.xml`, so with the comparator now +matching DUKE's semantics they are back inside the design they were fitted for. + +The gap was closed from the other side instead: **ENTSO-E installed capacity** +(`analysis/compare-with-entsoe-stats.py`) is an independent reference, and +aggregating away a station's operating units is visible in it. Summed over the +country × fueltype cells ENTSO-E reports, for 2025, excluding Wind/Solar (extended +separately) and Other (incomparable residual): + +| dedup name comparator | signed gap | Σ|gap| | RMSE | +|---|---|---|---| +| DUKE (rebuilt from the merge-base, same inputs) | −17.6 GW | 104.2 GW | 1.234 | +| `token_set_ratio` | −53.2 GW | 123.2 GW | 1.511 | +| **mean best-token Jaro-Winkler (shipped)** | **−5.5 GW** | **101.9 GW** | **1.227** | + +`token_set_ratio` collapsed each station's units into one record. Capacity was +conserved, but `AGGREGATION_FUNCTIONS["DateOut"] = "max"` skips NaN, so the merged +station inherited a retired unit's shutdown year and the whole site dropped out of +the operating fleet: Kozloduy 1–6 became one 3840 MW record retired in 2006, and +Belgium, Bulgaria and the Netherlands lost their entire operating nuclear +capacity. Nuclear was 20.3 GW short of the statistics; it is now 2.9 GW over, +against DUKE's 4.8 GW over. + +Making `DateOut` aggregate to NaN whenever a unit has no shutdown year was tried +and **rejected**: sources omit `DateOut` for plants that are in fact closed, so it +resurrects them (Σ|gap| 101.9 → 121.1 GW, Polish hard coal 18.6 → 26.6 GW against +19.0 reported). `"max"` is the better estimator once units are no longer merged. + +- Across all GPD countries, 95 % of the engine's dedup merges have high name + similarity *and* lie within 5 km *and* share a fueltype. + +## Outcome + +DUKE — its Java dependency, bundled `.jar` binaries, and XML configs — was +removed. `linkage` is the sole matching engine, selected automatically with no +configuration. `rapidfuzz` is a hard dependency; Java is no longer required. + +> Note: an earlier prototype used the `recordlinkage` library, but its per-pair +> Python comparison was both slower and the quality bottleneck. The final engine +> implements record linkage with `rapidfuzz` + numpy directly (vectorised, C++ +> string kernels). 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 8a405895..8a657fbc 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -8,6 +8,8 @@ SPDX-License-Identifier: MIT ## Upcoming Version +* Replaced the Java-based DUKE matching engine with a pure-Python, vectorised record-linkage backend (`powerplantmatching.linkage`, built on `rapidfuzz` + `numpy`). Matching no longer requires a Java installation or the bundled DUKE binaries, and is substantially faster. **Breaking:** the `powerplantmatching.duke` module is gone and can no longer be imported, and `add_geoposition_for_duke` has been removed. The `parallel_duke_processes` config key is renamed to `parallel_processes`; the old key still works but emits a deprecation warning. Note that `linkage` is not a reimplementation of DUKE — it uses a different scoring curve, and its thresholds were re-tuned against a GEO/GPD ground truth (see `analysis/linkage_findings.md`), so matching results differ from previous versions. +* The `projectID` and `EIC` columns now hold sorted lists instead of sets, so that two runs over the same inputs produce byte-identical output. Set iteration order depends on the interpreter's hash seed, which made the serialised identifiers differ between otherwise identical builds. **Breaking:** a missing `EIC` is now an empty list rather than `{nan}`, and set operations on these values need an explicit `set(...)`. * OSM dataset upgraded from a Europe-only snapshot (`osm_europe.csv`) to a global snapshot (`osm_global.csv.gz` taken from [`osm-powerplants`](https://github.com/open-energy-transition/osm-powerplants). ## [v0.8.1](https:://github.com/PyPSA/powerplantmatching/releases/tag/v0.8.1) (11th February 2026) diff --git a/powerplantmatching/accessor.py b/powerplantmatching/accessor.py index 05d50145..d5b9a78a 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, **kwargs): 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, **kwargs) if reduced: return res.pipe(reduce_matched_dataframe, config=config) return res diff --git a/powerplantmatching/cleaning.py b/powerplantmatching/cleaning.py index fbe80c74..f0001813 100644 --- a/powerplantmatching/cleaning.py +++ b/powerplantmatching/cleaning.py @@ -16,7 +16,7 @@ from deprecation import deprecated from .core import PANDAS_V3, get_config, get_obj_if_Acc -from .duke import duke +from .linkage import match from .utils import get_name, set_column_name logger = logging.getLogger(__name__) @@ -29,6 +29,17 @@ def mode(x): return x.mode(dropna=False).at[0] +def unique_sorted(x): + """ + Get the unique values of a series in a reproducible order. + + Sets iterate in an order that depends on the interpreter's hash seed, which + makes the serialised identifier columns differ between otherwise identical + runs. + """ + return sorted(set(x.dropna()), key=str) + + AGGREGATION_FUNCTIONS = { "Name": mode, "Fueltype": mode, @@ -43,8 +54,8 @@ def mode(x): "DateMothball": "min", "DateOut": "max", "File": mode, - "projectID": set, - "EIC": set, + "projectID": unique_sorted, + "EIC": unique_sorted, "Duration": "sum", # note this is weighted sum "Volume_Mm3": "sum", "DamHeight_m": "sum", @@ -431,7 +442,7 @@ def aggregate_units( pre_clean_name=False, country_wise=True, config=None, - threads=1, + threads=-1, **kwargs, ): """ @@ -450,8 +461,8 @@ def aggregate_units( Whether to clean the 'Name'-column before aggregating. country_wise : Boolean, default True Whether to aggregate only entries with a identical country. - threads : int, default 1 - Number of threads to use + threads : int, default -1 + Number of worker threads passed to the linkage engine, -1 uses all cores. """ deprecated_args = {"use_saved_aggregation", "save_aggregation"} used_deprecated_args = deprecated_args.intersection(kwargs) @@ -500,16 +511,14 @@ def aggregate_units( if with_blocks := config["clean_name"].get("fuel_type_with_blocks", []): # noqa block_query = "Fueltype in @with_blocks" + query = " and ".join(filter(None, [agg_query, block_query])) + selection = df.query(query) if query else df + if country_wise: - countries = df.Country.unique() - 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] - ) + per_country = [selection[selection.Country == c] for c in df.Country.unique()] + duplicates = pd.concat([match(sel, threads=threads) for sel in per_country]) else: - query = " and ".join(filter(None, [agg_query, block_query])) - duplicates = duke(df.query(query) if query else df, threads=threads) + duplicates = match(selection, 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 8169affa..deba9c29 100644 --- a/powerplantmatching/collection.py +++ b/powerplantmatching/collection.py @@ -31,7 +31,7 @@ def collect( update=False, reduced=True, config=None, - **dukeargs, + **kwargs, ): """ 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 + **kwargs : keyword-args for the matcher """ from . import data @@ -67,7 +67,7 @@ def df_by_name(name): if not conf.get("aggregated_units", False): return aggregate_units(df, dataset_name=name, config=config) else: - return df.assign(projectID=df.projectID.map(lambda x: {x})) + return df.assign(projectID=df.projectID.map(lambda x: [x])) # Deal with the case that only one dataset is requested if isinstance(datasets, str): @@ -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, **kwargs) ( 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..a8ccfc55 100644 --- a/powerplantmatching/core.py +++ b/powerplantmatching/core.py @@ -110,6 +110,17 @@ def get_config(filename=None, **overrides): config.update(yaml.load(f, Loader=yaml.FullLoader)) config.update(overrides) + if "parallel_duke_processes" in config: + config["parallel_processes"] = config.pop("parallel_duke_processes") + logger.warning( + "The configuration key `parallel_duke_processes` was renamed to " + "`parallel_processes` when the Java-based DUKE matching engine was " + "replaced. Its value is used for `parallel_processes`, but please rename " + f"the key in your custom config file at `{custom_config}`. For more " + "information, see the release notes at " + "https://powerplantmatching.readthedocs.io/en/latest/release-notes.html." + ) + 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/heuristics.py b/powerplantmatching/heuristics.py index 08bf2b1f..e8940908 100644 --- a/powerplantmatching/heuristics.py +++ b/powerplantmatching/heuristics.py @@ -55,7 +55,7 @@ def extend_by_non_matched( if config is None: config = get_config() - threads = config.get("threads_extend_by_non_matched", 1) + threads = config.get("threads_extend_by_non_matched", -1) if isinstance(extend_by, str): label = extend_by diff --git a/powerplantmatching/linkage.py b/powerplantmatching/linkage.py new file mode 100644 index 00000000..86a2516e --- /dev/null +++ b/powerplantmatching/linkage.py @@ -0,0 +1,247 @@ +# 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: mean 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 are tuned against the GEO/GPD ground truth in +``analysis/benchmark_linkage.py``; they are properties of this scoring curve and +carry no meaning outside it. +""" + +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 + +GEO_MAX_DISTANCE_M = 5000.0 +BLOCK_CELLS = 2_000_000 + +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 = [list(dict.fromkeys(v.split())) for v in values] + vocabulary = pd.unique(np.array([t for ts in tokens for t in ts] or [""])) + position = {token: i for i, token in enumerate(vocabulary)} + width = max(max((len(ts) for ts in tokens), default=1), 1) + codes = np.full((len(tokens), width), len(vocabulary), dtype=np.intp) + counts = np.zeros(len(tokens), dtype=np.intp) + for i, ts in enumerate(tokens): + codes[i, : len(ts)] = [position[t] for t in ts] + counts[i] = len(ts) + return codes, counts, np.append(vocabulary, "") + + +def _name_matrix( + left: pd.DataFrame, right: pd.DataFrame, column: str, threads: int +) -> Comparison: + """Mean best-token Jaro-Winkler similarity 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 + best = np.zeros((len(vocabulary_a), len(bv))) + for j in range(codes_b.shape[1]): + np.maximum(best, tokens[:, codes_b[:, j]], out=best) + total = np.zeros((len(av), len(bv))) + for k in range(codes_a.shape[1]): + total += np.where(counts_a[:, None] > k, best[codes_a[:, k]], 0.0) + 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 _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.isnan(av) & ~np.isnan(bv) + + +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.""" + r = 6371000.0 + 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 * r * np.arcsin(np.sqrt(np.clip(h, 0, 1))) + return np.clip(1 - dist / GEO_MAX_DISTANCE_M, 0.0, None), ~np.isnan(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 _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. In record linkage pass ``singlematch=True`` and + reduce with ``best_matches()`` afterwards. ``threshold`` overrides the tuned + acceptance probability (``DEDUP_THRESHOLD`` / ``LINKAGE_THRESHOLD``); + ``threads`` is the rapidfuzz worker count, ``-1`` meaning all cores. + """ + 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 and not res.empty: + res = res.loc[res.groupby(labels[0])["scores"].idxmax()] + return res.reset_index(drop=True) diff --git a/powerplantmatching/matching.py b/powerplantmatching/matching.py index 0d950185..06132ec7 100644 --- a/powerplantmatching/matching.py +++ b/powerplantmatching/matching.py @@ -14,7 +14,7 @@ from .cleaning import clean_technology from .core import get_config, get_obj_if_Acc -from .duke import duke +from .linkage import match from .utils import get_name, parmap, read_csv_if_string logger = logging.getLogger(__name__) @@ -22,13 +22,13 @@ def best_matches(links): """ - Subsequent to duke() with singlematch=True. Returns reduced list of + Subsequent to match() with singlematch=True. Returns reduced list of matches on the base of the highest score for each duplicated entry. Parameters ---------- links : pd.DataFrame - Links as returned by duke + Links as returned by match """ labels = links.columns.difference({"scores"}) if links.empty: @@ -39,9 +39,9 @@ def best_matches(links): return links.loc[best_idx, labels].reset_index(drop=True) -def compare_two_datasets(dfs, labels, country_wise=True, config=None, **dukeargs): +def compare_two_datasets(dfs, labels, country_wise=True, config=None, **kwargs): """ - Duke-based horizontal match of two databases. Returns the matched + 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 +49,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,24 +64,24 @@ 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(kwargs) if used_deprecated_args: for arg in used_deprecated_args: - dukeargs.pop(arg) + kwargs.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 kwargs: + kwargs["singlematch"] = True 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 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, **kwargs ) else: return pd.DataFrame(columns=[*labels, "scores"]) @@ -97,7 +95,7 @@ def country_link(dfs, country): else: links = pd.DataFrame(columns=[*labels, "scores"]) else: - links = duke(dfs, labels=labels, **dukeargs) + links = match(dfs, labels=labels, **kwargs) if links.empty: matches = pd.DataFrame(columns=labels) @@ -118,8 +116,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 - best_matches()) + scores), obtained from match() and best_matches() labels : list of strings list of names of the databases, used for specifying the order of the output @@ -167,10 +164,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, **kwargs ): """ - Duke-based horizontal match of multiple databases. Returns the + 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 +194,7 @@ 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, **kwargs) mapargs = [[dfs[c], dfs[d], labels[c], labels[d]] for c, d in combs] all_matches = parmap(comp_dfs, mapargs) @@ -205,9 +202,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, **kwargs): """ - Duke-based horizontal match of multiple databases. Returns the + 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 +249,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, **kwargs) return combined_dataframe(crossmatches, datasets, config).reindex( columns=config["target_columns"], level=0 ) 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 aced4f37..8afe6c72 100644 --- a/powerplantmatching/package_data/config.yaml +++ b/powerplantmatching/package_data/config.yaml @@ -53,7 +53,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 4b9396bc..8840c5a6 100644 --- a/powerplantmatching/utils.py +++ b/powerplantmatching/utils.py @@ -341,7 +341,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 --------- @@ -359,7 +359,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 63aeb040..090c8c8b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ dependencies = [ "deprecation", "tqdm", "openpyxl", + "rapidfuzz", ] [project.urls] 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..82431bbf --- /dev/null +++ b/test/test_linkage.py @@ -0,0 +1,326 @@ +# 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: + return {**BASE, **overrides} + + +def frame(records: list[dict], 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, right): + 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}, + True, + id="capacity-differs-50x-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, linked): + 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, second): + """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(): + """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(): + """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 + + +def test_field_specs_and_thresholds_are_pinned(): + """Tuned against the GEO/GPD harness in analysis/benchmark_linkage.py.""" + linkage = [(f.column, f.low, f.high) for f in lk.LINKAGE_FIELDS] + dedup = [(f.column, f.low, f.high) for f in lk.DEDUP_FIELDS] + + assert linkage == [ + ("Name", 0.09, 0.99), + ("Fueltype", 0.09, 0.7), + ("Country", 0.0, 0.53), + ("Capacity", 0.3, 0.75), + ("geo", 0.1, 0.8), + ] + assert dedup == [ + ("Name", 0.09, 0.99), + ("Fueltype", 0.05, 0.65), + ("Technology", 0.25, 0.51), + ("Country", 0.05, 0.51), + ("Capacity", 0.49, 0.51), + ("geo", 0.05, 0.75), + ] + assert (lk.LINKAGE_THRESHOLD, lk.DEDUP_THRESHOLD, lk.GEO_MAX_DISTANCE_M) == ( + 0.85, + 0.96, + 5000.0, + ) + + +@pytest.mark.parametrize( + "counterpart", [np.nan, "", "Hard Coal"], ids=["nan", "empty", "present"] +) +def test_missing_categorical_is_neutral_not_a_bonus(counterpart): + """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): + 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, expected): + 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(): + 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(): + 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, kwargs): + with pytest.raises(TypeError): + lk.match(left, **kwargs) + + +def test_geo_contributes_nothing_beyond_the_cutoff(): + 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(): + 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(): + """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, block_cells): + 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): + out = lk.match([left, left.iloc[0:0]], labels=["one", "two"]) + + assert out.empty + assert list(out.columns) == ["one", "two", "scores"] diff --git a/test/test_linkage_integration.py b/test/test_linkage_integration.py new file mode 100644 index 00000000..facd08e0 --- /dev/null +++ b/test/test_linkage_integration.py @@ -0,0 +1,75 @@ +# SPDX-FileCopyrightText: Contributors to powerplantmatching +# +# SPDX-License-Identifier: MIT + +"""Offline coverage of the two production callers of ``linkage.match``.""" + +import pandas as pd +import pytest + +import powerplantmatching as pm +from powerplantmatching.cleaning import aggregate_units +from powerplantmatching.matching import compare_two_datasets + +UNITS = [ + ("Alpha Power Station", "Hard Coal", "Steam Turbine", "Germany", 100.0, 51.0, 7.0), + ("Alpha Power Stn", "Hard Coal", "Steam Turbine", "Germany", 150.0, 51.0, 7.0), + ("Beta Hydro Plant", "Hydro", "Run-Of-River", "Germany", 40.0, 48.0, 11.0), + ("Alpha Power Station", "Hard Coal", "Steam Turbine", "France", 100.0, 45.0, 2.0), +] +COLUMNS = ["Name", "Fueltype", "Technology", "Country", "Capacity", "lat", "lon"] + + +@pytest.fixture(scope="module") +def config() -> dict: + return pm.get_config() + + +@pytest.fixture +def units(config) -> pd.DataFrame: + df = pd.DataFrame(UNITS, columns=COLUMNS) + extras = dict(Set="PP", DateIn=2000, Efficiency=0.4, Duration=0.0, EIC=None) + df = df.assign(projectID=[f"p{i}" for i in df.index], **extras) + return df.reindex(columns=df.columns.union(config["target_columns"], sort=False)) + + +def test_aggregate_units_merges_only_true_duplicates(units, config): + out = aggregate_units(units, dataset_name="test", config=config).sort_values( + "Capacity" + ) + + assert len(out) == 3 + assert out.Capacity.to_list() == [40.0, 100.0, 250.0] + + +def test_aggregated_identifiers_are_ordered(units, config): + """Set iteration order follows the hash seed, so identical runs would differ.""" + out = aggregate_units(units, dataset_name="test", config=config) + + assert all(ids == sorted(ids) for ids in out.projectID) + assert max(out.projectID, key=len) == ["p0", "p1"] + assert out.EIC.to_list() == [[], [], []] + + +def test_aggregate_units_does_not_merge_across_countries(units, config): + """The country-wise fan-out blocks on Country before matching.""" + single = aggregate_units( + units.query("Country == 'Germany'"), dataset_name="test", config=config + ) + both = aggregate_units(units, dataset_name="test", config=config) + + assert len(single) == 2 + assert len(both) == len(single) + 1 + + +def test_compare_two_datasets_returns_index_labels(units, config): + left = units.head(3) + right = units.head(3).iloc[::-1].set_axis([17, 18, 19]) + + matches = compare_two_datasets([left, right], ["A", "B"], config=config) + + assert set(map(tuple, matches[["A", "B"]].to_numpy())) == { + (0, 19), + (1, 18), + (2, 17), + }