added ipfs support as input source

.gitignore

@@ -11,4 +11,6 @@ site
 .ipynb_checkpoints/
 *.ipynb
 processed_data
-results
+results
+data_collection_scripts/google-service-account-key.json
+.ipfs_cache

config.yaml

@@ -62,5 +62,26 @@ plot_parameters:
 # List of paths that specify where to look for raw block data. Relative to the root directory of the repository.
 # The first item in the list is the directory that is used to write newly fetched data when using the
 #  `collect_block_data` script and is also the directory where tests expect the sample data to be found.
+# For this reason, entries that fetch data from IPFS (see below) should not be listed first.
+# Entries can also be IPFS references of the form `ipfs://<CID>` (optionally `ipfs://<CID>/<subpath>`). <CID>
+#  combined with <subpath> (if given) must point directly at a directory that contains files named
+#  <ledger>_raw_data.json (one or more of them, in the same format expected in a local input directory) - not at a
+#  parent/wrapper directory that merely contains such a directory. If the CID you were given only wraps the actual
+#  dataset directory (e.g. a single named subdirectory holding the raw data files), append
+#  that subdirectory's name as <subpath> so the reference resolves to where the .json files actually live. Such
+#  files are fetched via the configured `ipfs_gateway` and cached locally (in .ipfs_cache/) so that they are not
+#  re-fetched on subsequent runs.
 input_directories:
   - raw_block_data
+  #- ipfs://<CID>
+
+# The gateway(s) used to resolve `ipfs://<CID>` entries in input_directories. Can be a single URL (as below) or a
+#  list of URLs, in which case they are tried in order for each file, falling back to the next one on failure. This
+#  can be used e.g. to fall back to a local/self-hosted node's gateway (typically http://127.0.0.1:8080, requires
+#  running `ipfs daemon`) if a public gateway is unavailable or rate-limited:
+#  ipfs_gateway:
+#    - https://ipfs.io
+#    - http://127.0.0.1:8080
+ipfs_gateway:
+  - https://ipfs.io
+  - http://127.0.0.1:8080

consensus_decentralization/helper.py

@@ -7,15 +7,18 @@
 import datetime
 import calendar
 import argparse
+import logging
 from functools import lru_cache
 from collections import defaultdict
 
+import requests
 from yaml import safe_load
 
 ROOT_DIR = pathlib.Path(__file__).resolve().parent.parent
 INTERIM_DIR = ROOT_DIR / 'processed_data'
 MAPPING_INFO_DIR = ROOT_DIR / 'mapping_information'
 RESULTS_DIR = ROOT_DIR / 'results'
+IPFS_CACHE_DIR = ROOT_DIR / '.ipfs_cache'
 
 with open(ROOT_DIR / "config.yaml") as f:
     config = safe_load(f)
@@ -474,10 +477,71 @@ def get_output_filename(clustering_flag):
     return 'output_' + ('clustered' if clustering_flag else 'non_clustered') + '.csv'
 
 
+def get_ipfs_gateways():
+    """
+    Retrieves the IPFS gateway URLs used to resolve `ipfs://<CID>` entries in input_directories. If more than one is
+    configured, they are tried in order for each file, falling back to the next one on failure (e.g. a public
+    gateway followed by a local node's gateway, typically http://127.0.0.1:8080).
+    :returns: list of str, the base URLs of the gateways (defaults to a single public gateway if not set in the
+        config file)
+    """
+    config = get_config_data()
+    gateways = config.get('ipfs_gateway', 'https://ipfs.io')
+    if isinstance(gateways, str):
+        gateways = [gateways]
+    return [gateway.rstrip('/') for gateway in gateways]
+
+
+def fetch_ipfs_directory(cid, subpath=''):
+    """
+    Fetches the raw data files (one per ledger defined in the config file) from an IPFS directory, via the configured
+    IPFS gateway(s), and caches them locally. Files that have already been cached are not re-fetched. If a file can't
+    be fetched from any configured gateway, it is skipped (a warning is logged), so that other input directories may
+    still be used to look for the corresponding ledger's raw data.
+    :param cid: string, the IPFS CID of a (unixfs) directory that is expected to contain files named
+        <ledger>_raw_data.json for one or more ledgers
+    :param subpath: string, an optional path within the directory pointed to by the CID
+    :returns: pathlib.PosixPath object of the local directory where the fetched files are cached
+    """
+    gateways = get_ipfs_gateways()
+    cache_dir = IPFS_CACHE_DIR / cid / subpath if subpath else IPFS_CACHE_DIR / cid
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    gateway_path = f'{cid}/{subpath}' if subpath else cid
+
+    for ledger in get_ledgers():
+        filename = f'{ledger}_raw_data.json'
+        local_path = cache_dir / filename
+        if local_path.is_file():
+            continue
+        for gateway in gateways:
+            url = f'{gateway}/ipfs/{gateway_path}/{filename}'
+            try:
+                response = requests.get(url, timeout=60)
+                response.raise_for_status()
+            except requests.RequestException as e:
+                logging.warning(f'Could not fetch {filename} from IPFS at {url}: {e}')
+                continue
+            local_path.write_bytes(response.content)
+            break
+
+    return cache_dir
+
+
 def get_input_directories():
     """
-    Reads the config file and retrieves the directories to look for raw block data
+    Reads the config file and retrieves the directories to look for raw block data. Entries can be local paths,
+    relative to the root directory of the repository, or IPFS references of the form `ipfs://<CID>` (optionally
+    `ipfs://<CID>/<subpath>`), in which case the relevant raw data files are fetched (if not already cached) from the
+    configured IPFS gateway and the local cache directory is used in their place.
     :returns: a list of directories that may contain the raw block data
     """
     config = get_config_data()
-    return [ROOT_DIR / input_dir for input_dir in config['input_directories']]
+    input_dirs = []
+    for entry in config['input_directories']:
+        entry = str(entry)
+        if entry.startswith('ipfs://'):
+            cid, _, subpath = entry[len('ipfs://'):].partition('/')
+            input_dirs.append(fetch_ipfs_directory(cid, subpath))
+        else:
+            input_dirs.append(ROOT_DIR / entry)
+    return input_dirs

docs/data.md

@@ -125,3 +125,26 @@ There are also two command line arguments that can be used to customize the data
 - `--force-query` forces the collection of all raw data files, even if the corresponding files already
   exist. By default, this flag is set to False and the script only fetches block data for some blockchain if the
   corresponding file does not already exist.
+
+## Reading input data from IPFS
+
+In addition to local directories, entries in the `input_directories` list of the
+[configuration file](https://github.com/Blockchain-Technology-Lab/consensus-decentralization/blob/main/config.yaml)
+can be IPFS references of the form `ipfs://<CID>` (optionally `ipfs://<CID>/<subpath>`). `<CID>` combined with
+`<subpath>` (if given) must resolve directly to a (unixfs) directory that contains raw data files named
+`<ledger>_raw_data.json`, in the same format expected in a local input directory (one file per ledger, following the
+schemas described above) - not to a parent/wrapper directory that merely contains such a directory.
+
+For example, a dataset publisher might give you a CID whose root only contains a single named subdirectory (e.g.
+`solana-dataset/`) which in turn holds the actual `<ledger>_raw_data.json` file. In that case the CID alone is not
+enough - you need `ipfs://<CID>/solana-dataset` so that the reference points directly at the directory containing the
+`.json` file. You can check a CID's contents beforehand by browsing `https://<gateway>/ipfs/<CID>/` (e.g.
+`https://ipfs.io/ipfs/<CID>/`) in a browser.
+
+When such an entry is encountered, the relevant files are fetched from the gateway(s) configured via `ipfs_gateway` in
+the configuration file (a public gateway such as `https://ipfs.io` by default) and cached locally under
+`.ipfs_cache/` at the root of the repository, so that they are only fetched once. `ipfs_gateway` can also be a list
+of gateway URLs, in which case they are tried in order for each file, falling back to the next one on failure -
+useful for falling back to a local/self-hosted node's gateway (typically `http://127.0.0.1:8080`, requires running
+`ipfs daemon`) if a public gateway is unavailable or rate-limited. Note that the first item of `input_directories` is
+used as the destination for newly collected data (see above) and therefore should not be an IPFS reference.

requirements.txt

@@ -8,4 +8,5 @@ matplotlib>=3.4.3
 seaborn>=0.11.2
 colorcet>=3.0.1
 pandas>=1.3.4
-google>=3.0.0
+google>=3.0.0
+requests>=2.28.0

tests/test_helper.py

@@ -2,6 +2,9 @@
 import argparse
 import shutil
 import pytest
+from unittest.mock import patch, Mock
+import requests
+import consensus_decentralization.helper as hlp
 from consensus_decentralization.helper import get_pool_identifiers, get_pool_legal_links, get_known_addresses, \
     get_pool_clusters, write_blocks_per_entity_to_file, get_blocks_per_entity_from_file, get_timeframe_beginning, \
     get_timeframe_end, get_time_period, get_ledgers, valid_date, INTERIM_DIR, get_blocks_per_entity_filename, \
@@ -184,3 +187,59 @@ def test_get_representative_dates():
         ]
     representative_dates = get_representative_dates(time_chunks)
     assert representative_dates == ['2022-07-02', '2023-07-02', '2024-07-01']
+
+
+def test_get_input_directories(tmp_path, monkeypatch):
+    monkeypatch.setattr(hlp, 'IPFS_CACHE_DIR', tmp_path)
+    monkeypatch.setattr(hlp, 'config', {
+        'ledgers': ['bitcoin', 'ethereum'],
+        'input_directories': ['raw_block_data', 'ipfs://testcid'],
+    })
+
+    mock_response = Mock(content=b'{"number": 1}\n')
+    with patch('consensus_decentralization.helper.requests.get', return_value=mock_response) as mock_get:
+        dirs = hlp.get_input_directories()
+
+    ipfs_dir = tmp_path / 'testcid'
+    assert dirs == [hlp.ROOT_DIR / 'raw_block_data', ipfs_dir]
+    assert (ipfs_dir / 'bitcoin_raw_data.json').read_bytes() == b'{"number": 1}\n'
+    assert (ipfs_dir / 'ethereum_raw_data.json').is_file()
+    assert mock_get.call_count == 2
+    mock_get.assert_any_call('https://ipfs.io/ipfs/testcid/bitcoin_raw_data.json', timeout=60)
+
+    # Cached files should not be re-fetched on subsequent calls
+    with patch('consensus_decentralization.helper.requests.get', return_value=mock_response) as mock_get_again:
+        hlp.get_input_directories()
+    assert mock_get_again.call_count == 0
+
+
+def test_get_input_directories_skips_missing_ledger_data(tmp_path, monkeypatch):
+    monkeypatch.setattr(hlp, 'IPFS_CACHE_DIR', tmp_path)
+    monkeypatch.setattr(hlp, 'config', {
+        'ledgers': ['bitcoin'],
+        'input_directories': ['ipfs://testcid'],
+    })
+
+    with patch('consensus_decentralization.helper.requests.get', side_effect=requests.RequestException('not found')):
+        dirs = hlp.get_input_directories()
+
+    assert not (dirs[0] / 'bitcoin_raw_data.json').exists()
+
+
+def test_get_input_directories_falls_back_to_next_gateway(tmp_path, monkeypatch):
+    monkeypatch.setattr(hlp, 'IPFS_CACHE_DIR', tmp_path)
+    monkeypatch.setattr(hlp, 'config', {
+        'ledgers': ['bitcoin'],
+        'input_directories': ['ipfs://testcid'],
+        'ipfs_gateway': ['https://unreachable.example', 'http://127.0.0.1:8080'],
+    })
+
+    mock_response = Mock(content=b'{"number": 1}\n')
+    with patch('consensus_decentralization.helper.requests.get',
+               side_effect=[requests.RequestException('unreachable'), mock_response]) as mock_get:
+        dirs = hlp.get_input_directories()
+
+    assert (dirs[0] / 'bitcoin_raw_data.json').read_bytes() == b'{"number": 1}\n'
+    assert mock_get.call_count == 2
+    mock_get.assert_any_call('https://unreachable.example/ipfs/testcid/bitcoin_raw_data.json', timeout=60)
+    mock_get.assert_any_call('http://127.0.0.1:8080/ipfs/testcid/bitcoin_raw_data.json', timeout=60)