initial code commit
ADD
.github/dependabot.yml
+
+version: 2
+updates:
+ - package-ecosystem: pip
+ directory: "/"
+ schedule:
+ interval: "daily"
+ - package-ecosystem: "github-actions"
+ directory: "/"
+ schedule:
+ interval: "daily"

ADD
.github/workflows/main.yaml
+name: CI
+
+on:
+ push:
+ branches: "*"
+ pull_request:
+ branches: master
+
+jobs:
+ lint-black:
+ name: black-formatter
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v1
+ - name: Set up Python 3.8
+ uses: actions/setup-python@v1
+ with:
+ python-version: 3.8
+ - name: Install Black
+ run: |
+ python -m pip install black
+ - name: Black Code Formatter
+ run: |
+ black --check --line-length 100 --skip-string-normalization .
+
+ lint-flake8:
+ name: flake8
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v1
+ - name: Set up Python 3.8
+ uses: actions/setup-python@v1
+ with:
+ python-version: 3.8
+ - name: Install Flake8
+ run: |
+ python -m pip install flake8
+ - name: Flake8 check
+ run: |
+ flake8 .
+
+ lint-isort:
+ name: isort
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v1
+ - name: Set up Python 3.8
+ uses: actions/setup-python@v1
+ with:
+ python-version: 3.8
+ - name: Install isort
+ run: |
+ python -m pip install isort
+ - name: isort check
+ run: |
+ isort --recursive --check-only .
+
+ test:
+ name: ${{ matrix.python-version }}-build
+ runs-on: ubuntu-latest
+ strategy:
+ matrix:
+ python-version: [3.7, 3.8]
+ steps:
+ - uses: actions/checkout@v2
+ - name: Setup Python
+ uses: actions/setup-python@v1
+ with:
+ python-version: ${{ matrix.python-version }}
+ architecture: x64
+ - uses: actions/cache@v1
+ with:
+ path: ~/.cache/pip
+ key: ${{ runner.os }}-pip-${{ hashFiles('**/dev-requirements.txt') }}
+ restore-keys: |
+ ${{ runner.os }}-pip-
+ - run: |
+ python -m pip install -r dev-requirements.txt
+ python -m pip install --no-deps -e .
+ python -m pip list
+ - name: Running Tests
+ run: |
+ python -m pytest --verbose --junitxml=test-reports/junit.xml --cov=xpublish --cov-config .coveragerc --cov-report term-missing
+ # - name: Upload coverage to Codecov
+ # uses: codecov/codecov-action@v1
+ # with:
+ # token: ${{ secrets.CODECOV_TOKEN }}
+ # file: test-reports/junit.xml

ADD
.pre-commit-config.yaml
+repos:
+
+- repo: https://github.com/pre-commit/pre-commit-hooks
+ rev: v2.5.0
+ hooks:
+ - id: trailing-whitespace
+ - id: end-of-file-fixer
+ - id: check-docstring-first
+ - id: check-json
+ - id: check-yaml
+ - id: pretty-format-json
+ args: ["--autofix", "--indent=2", "--no-sort-keys"]
+
+- repo: https://github.com/ambv/black
+ rev: 19.10b0
+ hooks:
+ - id: black
+ args: ["--line-length", "100"]
+
+- repo: https://gitlab.com/pycqa/flake8
+ rev: 3.8.0a2
+ hooks:
+ - id: flake8
+
+- repo: https://github.com/asottile/seed-isort-config
+ rev: v2.1.1
+ hooks:
+ - id: seed-isort-config
+- repo: https://github.com/pre-commit/mirrors-isort
+ rev: v4.3.21
+ hooks:
+ - id: isort
+
+- repo: https://github.com/deathbeds/prenotebook
+ rev: master
+ hooks:
+ - id: prenotebook
+
+- repo: https://github.com/myint/rstcheck
+ rev: master
+ hooks:
+ - id: rstcheck

ADD
examples/oisst-avhrr-v02r01-pipeline.py
+from pangeo_forge.pipeline import AbstractPipeline, XarrayPrefectPipelineMixin
+
+
+class Pipeline(AbstractPipeline, XarrayPrefectPipelineMixin):
+ def __init__(self, name="oisst-avhrr-v02r01"):
+ self.name = name
+ self.cache_location = f"gs://pangeo-scratch/rabernat/pangeo_smithy/{name}-cache/"
+ self.concat_dim = "time"
+ self.files_per_chunk = 5
+
+ @property
+ def sources(self):
+ import pandas as pd
+
+ keys = pd.date_range("1981-09-01", "1981-09-10", freq="D")
+ source_url_pattern = "https://www.ncei.noaa.gov/data/sea-surface-temperature-optimum-interpolation/v2.1/access/avhrr/{yyyymm}/oisst-avhrr-v02r01.{yyyymmdd}.nc"
+ source_urls = [
+ source_url_pattern.format(yyyymm=key.strftime("%Y%m"), yyyymmdd=key.strftime("%Y%m%d"))
+ for key in keys
+ ]
+ return source_urls
+
+ @property
+ def targets(self):
+ return f"gs://pangeo-scratch/rabernat/pangeo_smithy/{self.name}-target/"

ADD
examples/oisst-avhrr-v02r01.py
+import os
+
+import fsspec
+import pandas as pd
+import zarr
+from prefect import Flow, task
+
+from pangeo_forge.utils import chunked_iterable
+
+
+def src_url_from_key(key):
+ source_url_pattern = (
+ "https://www.ncei.noaa.gov/data/sea-surface-temperature-optimum-interpolation"
+ "/v2.1/access/avhrr/{yyyymm}/oisst-avhrr-v02r01.{yyyymmdd}.nc"
+ )
+ return source_url_pattern.format(yyyymm=key.strftime("%Y%m"), yyyymmdd=key.strftime("%Y%m%d"))
+
+
+cache_location = "gs://pangeo-scratch/rabernat/pangeo_smithy/oisst-avhrr-v02r01-cache/"
+
+
+@task
+def download(key):
+ source_url = src_url_from_key(key)
+ target_url = os.path.join(cache_location, str(hash(key)))
+
+ # there is probably a better way to do caching!
+ try:
+ fsspec.open(target_url).open()
+ return target_url
+ except FileNotFoundError:
+ pass
+
+ with fsspec.open(source_url, mode="rb") as source:
+ with fsspec.open(target_url, mode="wb") as target:
+ target.write(source.read())
+ return target_url
+
+
+@task
+def combine_and_write(sources, target, append_dim, first=True):
+ import xarray as xr
+
+ # while debugging this, I had itermittent fsspec / hdf5 read errors related to
+ # "trying to read from a closed file"
+ # but they seem to have gone away for now
+ double_open_files = [fsspec.open(url).open() for url in sources]
+ ds = xr.open_mfdataset(double_open_files, combine="nested", concat_dim=concat_dim)
+ # by definition, this should be a contiguous chunk
+ ds = ds.chunk({append_dim: len(sources)})
+
+ if first:
+ kwargs = dict(mode="w")
+ else:
+ kwargs = dict(mode="a", append_dim=append_dim)
+
+ mapper = fsspec.get_mapper(target)
+ ds.to_zarr(mapper, **kwargs)
+
+
+@task
+def consolidate_metadata(target):
+ mapper = fsspec.get_mapper(target)
+ zarr.consolidate_metadata(mapper)
+
+
+keys = pd.date_range("1981-09-01", "1981-09-10", freq="D")
+target_path = "gs://pangeo-scratch/rabernat/pangeo_smithy/oisst-avhrr-v02r01-target/"
+concat_dim = "time"
+files_per_chunk = 5
+
+with Flow("Pangeo-Forge") as flow:
+ sources = [download(k) for k in keys]
+ first = True
+ write_tasks = []
+ for source_group in chunked_iterable(sources, files_per_chunk):
+ write_task = combine_and_write(source_group, target_path, concat_dim, first=first)
+ write_tasks.append(write_task)
+ first = False
+ cm = consolidate_metadata(target_path)
+
+# create dependencies in imperative mode
+for n in range(1, len(write_tasks)):
+ write_tasks[n].set_upstream(write_tasks[n - 1], flow=flow)
+cm.set_upstream(write_tasks[-1], flow=flow)
+
+
+# DO IT! #
+flow.run()

ADD
pangeo_forge/README.md
+# Pangeo Forge
+
+Tool for building cloud optimized analysis ready datasets.

ADD
pangeo_forge/__init__.py
+from pkg_resources import DistributionNotFound, get_distribution
+
+try:
+ __version__ = get_distribution(__name__).version
+except DistributionNotFound: # noqa: F401
+ # package is not installed
+ pass

ADD
pangeo_forge/cli.py
+import click
+
+
+@click.command()
+@click.version_option()
+def main():
+ """Pangeo-Forge: A tool for building and publishing analysis ready datasets.
+ """
+ click.secho("Hello World!", fg="green")
+
+
+if __name__ == "__main__":
+ main()

ADD
pangeo_forge/pipeline.py
+import os
+from abc import ABC, abstractmethod
+from typing import List
+
+import fsspec
+import zarr
+from prefect import Flow, task
+
+from .utils import chunked_iterable
+
+
+class AbstractPipeline(ABC):
+ @property
+ @abstractmethod
+ def sources(self) -> List[str]:
+ pass
+
+ @property
+ @abstractmethod
+ def targets(self) -> List[str]:
+ pass
+
+ @abstractmethod
+ def run(self) -> None:
+ pass
+
+
+class XarrayPrefectPipelineMixin:
+ @task
+ def download(self, source_url):
+ target_url = os.path.join(self.cache_location, str(hash(source_url)))
+
+ # there is probably a better way to do caching!
+ try:
+ fsspec.open(target_url).open()
+ return target_url
+ except FileNotFoundError:
+ pass
+
+ with fsspec.open(source_url, mode="rb") as source:
+ with fsspec.open(target_url, mode="wb") as target:
+ target.write(source.read())
+ return target_url
+
+ @task
+ def combine_and_write(self, sources, target, append_dim, first=True):
+ import xarray as xr
+
+ # while debugging this, I had itermittent fsspec / hdf5 read errors related to
+ # "trying to read from a closed file"
+ # but they seem to have gone away for now
+ double_open_files = [fsspec.open(url).open() for url in sources]
+ ds = xr.open_mfdataset(double_open_files, combine="nested", concat_dim=self.concat_dim)
+ # by definition, this should be a contiguous chunk
+ ds = ds.chunk({append_dim: len(sources)})
+
+ if first:
+ kwargs = dict(mode="w")
+ else:
+ kwargs = dict(mode="a", append_dim=append_dim)
+
+ mapper = fsspec.get_mapper(target)
+ ds.to_zarr(mapper, **kwargs)
+
+ @task
+ def consolidate_metadata(self):
+ mapper = fsspec.get_mapper(self.targets)
+ zarr.consolidate_metadata(mapper)
+
+ @property
+ def flow(self) -> Flow:
+ with Flow("Pangeo-Forge") as flow:
+
+ cached_sources = [self.download(k) for k in self.sources]
+
+ first = True
+ write_tasks = []
+ for source_group in chunked_iterable(cached_sources, self.files_per_chunk):
+ write_task = self.combine_and_write(
+ source_group, self.targets, self.concat_dim, first=first
+ )
+ write_tasks.append(write_task)
+ first = False
+ cm = self.consolidate_metadata(self.targets)
+
+ # create dependencies in imperative mode
+ for n in range(1, len(write_tasks)):
+ write_tasks[n].set_upstream(write_tasks[n - 1], flow=flow)
+ cm.set_upstream(write_tasks[-1], flow=flow)
+
+ return flow
+
+ def run(self):
+ self.flow.run()

ADD
pangeo_forge/utils.py
+import itertools
+
+
+# https://alexwlchan.net/2018/12/iterating-in-fixed-size-chunks/
+def chunked_iterable(iterable, size):
+ it = iter(iterable)
+ while True:
+ chunk = tuple(itertools.islice(it, size))
+ if not chunk:
+ break
+ yield chunk

ADD
pyproject.toml
+[tool.black]
+line-length = 100

ADD
requirements.txt
+setuptools
+click
+dask
+distributed
+prefect
+xarray
+zarr
+fsspec

ADD
setup.cfg
+[flake8]
+max-line-length = 100
+
+[isort]
+known_first_party=pangeo_forge
+known_third_party=click,fsspec,pandas,pkg_resources,prefect,setuptools,zarr
+multi_line_output=3
+include_trailing_comma=True
+force_grid_wrap=0
+combine_as_imports=True
+line_length=100
+
+[tool:pytest]
+log_cli = True
+log_level = DEBUG

ADD
setup.py
+#!/usr/bin/env python
+
+"""The setup script."""
+
+from setuptools import find_packages, setup
+
+with open("requirements.txt") as f:
+ install_requires = f.read().strip().split("\n")
+
+with open("README.md") as f:
+ long_description = f.read()
+
+
+CLASSIFIERS = [
+ "Development Status :: 1 - Planning",
+ "License :: OSI Approved :: Apache Software License",
+ "Operating System :: OS Independent",
+ "Intended Audience :: Science/Research",
+ "Programming Language :: Python",
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.7",
+ "Programming Language :: Python :: 3.8",
+ "Topic :: Scientific/Engineering",
+]
+
+setup(
+ name="pangeo-forge",
+ description="Pipeline tools for building and publishing analysis ready datasets",
+ long_description=long_description,
+ long_description_content_type="text/markdown",
+ python_requires=">=3.7",
+ maintainer="Joe Hamman",
+ maintainer_email="jhamman@ucar.edu",
+ classifiers=CLASSIFIERS,
+ url="https://github.com/pangeo-forge/pangeo-forge",
+ packages=find_packages(),
+ package_dir={"pangeo_forge": "pangeo_forge"},
+ include_package_data=True,
+ install_requires=install_requires,
+ license="Apache",
+ zip_safe=False,
+ keywords=["pangeo", "data"],
+ use_scm_version={"version_scheme": "post-release", "local_scheme": "dirty-tag"},
+ setup_requires=["setuptools_scm", "setuptools>=30.3.0"],
+ entry_points=dict(console_scripts=["pangeo-forge = pangeo_forge.cli:main"]),
+)

ADD
tests/test_cli.py
+from click.testing import CliRunner
+
+from pangeo_forge import __version__
+from pangeo_forge.cli import main as cli
+
+
+def test_version():
+ runner = CliRunner()
+ result = runner.invoke(cli, ["--version"])
+ assert result.exit_code == 0
+ assert __version__ in result.output
+
+
+def test_help():
+ runner = CliRunner()
+ result = runner.invoke(cli, ["--help"])
+ assert result.exit_code == 0
+ assert cli.__doc__.strip() in result.output

ADD
tests/test_pipeline.py
tests/test_pipeline.py
initial code commit