数据源与读取器

数据源与读取器#

统一入口#

reki.from_source(name: str, *args, lazily: bool = False, **kwargs) → Source#

Create a source by name and mutate it into the most concrete source.

The source class is looked up by SourceMaker, instantiated with *args and **kwargs, and then mutate() is called repeatedly until the source no longer changes (fixed-point loop). For example, LocalSource mutates into FileSource once the data path is resolved.

After the loop, to_data_object() converts the final source into the unified data object: for data-bearing sources (e.g. file) this is the reader produced by the reki.readers dispatch; other sources (e.g. memory) simply return themselves.

Sources marked remote = True (e.g. url, cmadaas) perform remote I/O in their pipeline; for them from_source() returns a LazySource proxy so that calling from_source() alone never fires a remote request — the request happens on first use (e.g. to_xarray()).

参数:

name -- source name, e.g. "memory". Underscores and hyphens are interchangeable.
*args -- positional arguments passed to the source class.
lazily -- if True, return a LazySource proxy that defers the whole pipeline — including source construction — to first attribute access.
**kwargs -- keyword arguments passed to the source class.

返回:

the unified data object, or a lazy proxy for it.

返回类型:

Source or Reader or LazySource

reki.from_source_lazily(name: str, *args, **kwargs) → LazySource#

Lazy variant of from_source().

Returns a LazySource proxy; source construction, the mutate loop and the conversion to the data object all happen on first attribute access, not before.

reki.register(name: str, klass) → None#

参数:

name -- source name used in from_source(name, ...). Underscores and hyphens are interchangeable.
klass -- the source class. It is instantiated with the arguments passed to from_source.

数据源#

class reki.Source(**kwargs)#

Base class for all sources.

A source knows where the data comes from (local file, CMA HPC archive, CMADaaS service, memory object, ...). Sources are created through reki.from_source() and are transformed by the mutate() loop into the most concrete source before being parsed by a reader.

This class follows the design of earthkit.data.sources.Source.

mutate()#

Transform this source into a more concrete source.

Called repeatedly by from_source() until the returned source is identical to the previous one (fixed-point loop). For example, LocalSource mutates into FileSource once the path is resolved. The default implementation returns self, which terminates the loop.

返回:: the mutated source, or self if no mutation is needed.
返回类型:: Source

mutate_source()#

Hook for a reader to replace the source.

Gives the reader a chance to ask the source to transform itself (e.g. an archive source expanding into a multi source) before the reader gives up. Optional; the default implementation returns None which means no replacement.

返回:: a replacement source, or None.
返回类型:: Source or None

remote = False#: True for sources whose mutate() performs remote I/O (a download or a service request). from_source() defers the mutate loop and to_data_object() of such sources to first use by returning a lazy proxy, so calling from_source() alone never fires a remote request.

to_data_object()#

Convert this source into the unified data object.

Called by from_source() after the mutate fixed-point loop converges. Sources carrying parseable data (e.g. FileSource) override this to return the reader for their data; the default implementation returns the source itself (e.g. MemorySource already provides the conversion methods).

返回:: the unified data object (a reader or the source itself).
返回类型:: object

内置数据源：

class reki.sources.test.TestSource(dataset_name: str = 'gfs', output_dir: str | Path | None = None, domain: Literal['eastasia', 'global'] = 'eastasia', source: Literal['wis', 'music-dir'] = 'wis', storage_base: str | None = None, start_time: Timestamp | None = None, forecast_time: Timedelta | None = None, **kwargs)#

Fetch a test dataset file, then read it as a local file.

参数:

dataset_name -- which dataset to fetch: "ecmwf_ifs" (frozen ECMWF IFS subset from a GitHub release; documentation examples) or "cma_gfs" (rolling CMA-GFS from the WIS website or a mounted music-dir directory; tests only, not reproducible). The legacy name "gfs" is accepted as an alias of "cma_gfs".
domain -- for "ecmwf_ifs" only: which frozen asset to fetch, "eastasia" (default) or "global".
output_dir -- directory the data file is downloaded to. Defaults to a per-user temp directory. Downloads are idempotent: an existing file is reused (see reki.sources.url.download_file).
source -- for "cma_gfs" only: fetch backend, "wis" (HTTP download) or "music-dir" (copy from a mounted directory, requires storage_base).
storage_base -- storage base directory for source="music-dir".
start_time -- for "cma_gfs" only: model start time. Defaults to yesterday 00Z.
forecast_time -- for "cma_gfs" only: forecast time. Defaults to 24 hours.

mutate() → Source#

Transform this source into a more concrete source.

返回:: the mutated source, or self if no mutation is needed.
返回类型:: Source

remote = True#: fetching the dataset is remote I/O; defer it to first use.

class reki.sources.file.FileSource(path, reader=None, **kwargs)#

A source for a local data file, format auto-detected by readers.

参数:

path -- path of the data file.
reader -- optional explicit reader: a reader name (e.g. "grib") or a callable. When given, the reader auto-detection is skipped.
**kwargs -- extra options forwarded to the reader (e.g. engine="cfgrib" for the GRIB reader).

mutate()#

Transform this source into a more concrete source.

返回:: the mutated source, or self if no mutation is needed.
返回类型:: Source

to_data_object()#

Convert this source into the unified data object.

返回:: the unified data object (a reader or the source itself).
返回类型:: object

class reki.sources.url.UrlSource(url: str, download_dir: str | Path | None = None, reader=None, **kwargs)#

A source for a remote file identified by a URL.

mutate() downloads the URL to a local path and returns a FileSource for it, so the reader dispatch chain continues automatically:

reki.from_source("url", "https://example.com/data.grib2").to_xarray()

The download is remote I/O, so this source sets remote = True: from_source() returns a lazy proxy and the download only happens on first use (e.g. to_xarray()).

参数:

url -- the remote file URL.
download_dir -- directory to download into. Defaults to a shared temp directory (DEFAULT_DOWNLOAD_DIR). If the target file already exists the download is skipped.
reader -- optional explicit reader forwarded to the FileSource: a reader name (e.g. "grib") or a callable.
**kwargs -- extra options forwarded to the reader (e.g. engine="cfgrib" for the GRIB reader).

local_path() → Path#: The local path the URL is (or will be) downloaded to.

mutate()#

Transform this source into a more concrete source.

返回:: the mutated source, or self if no mutation is needed.
返回类型:: Source

remote = True#: True for sources whose mutate() performs remote I/O (a download or a service request). from_source() defers the mutate loop and to_data_object() of such sources to first use by returning a lazy proxy, so calling from_source() alone never fires a remote request.

class reki.sources.memory.MemorySource(buf, reader=None, **kwargs)#

Wrap an in-memory object as a source.

Supports xarray.DataArray, xarray.Dataset, pandas.DataFrame and numpy.ndarray. The source mutates to itself and only provides conversions between these representations.

When an explicit reader is given, buf may be any object the named memory reader knows how to handle (e.g. a CMADaaS response object with reader="cmadaas"); to_data_object() then dispatches to that reader instead of returning the source itself.

参数:

buf -- the in-memory object to wrap.
reader -- optional explicit memory reader: a reader name (e.g. "cmadaas") or a callable.

mutate()#

Transform this source into a more concrete source.

返回:: the mutated source, or self if no mutation is needed.
返回类型:: Source

to_data_object()#

Convert this source into the unified data object.

返回:: the unified data object (a reader or the source itself).
返回类型:: object

class reki.sources.local.LocalSource(data_type: str, start_time: str | Timestamp | datetime, forecast_time: str | Timedelta = '0', **kwargs)#

Resolve a data path on the CMA HPC file system.

The path is resolved from YAML configs and Jinja2 templates (the former reki.data_finder.local logic). mutate() resolves the path and returns a FileSource for it.

参数:

data_type -- data type, relative path of the config file without suffix, e.g. "cma_gfs_gmf/grib2/orig".
start_time -- start time of production. YYYYMMDDHH if str.
forecast_time -- forecast time of production. A string (such as "3h") will be parsed by pd.to_timedelta.
**kwargs -- data_level / data_class / config_dir / obs_time / debug and any extra variables used by the path template.

mutate()#

Transform this source into a more concrete source.

返回:: the mutated source, or self if no mutation is needed.
返回类型:: Source

resolve_path() → Path | None#: Resolve the local data path, or return None if not found.

读取器#

class reki.readers.Reader(source, path, **kwargs)#

Base class for all readers.

A reader is bound to a source and a path and provides conversions to unified data objects: to_xarray() / to_pandas() / to_numpy().

mutate()#: Give the reader a chance to replace itself after creation.

mutate_source()#: Give the reader a chance to replace the source, or None.

class reki.readers.grib.reader.GribReader(source, path, engine: str = 'eccodes', filters: Dict | None = None, **kwargs)#

Lazy query object for GRIB files.

参数:

source -- the source the file comes from.
path -- path of the GRIB file.
engine -- decoding engine, "eccodes" (default) or "cfgrib".
filters -- initial filter conditions, see sel().

property filters: Dict#: The accumulated filter conditions (a copy).

first() → GribField | None#

Scan sequentially and return the first matching field.

The filter conditions are passed to the engine kernel unchanged (each engine applies its own level type fixing), so the result is identical to calling the kernel's load_field_from_file.

返回:: the first matching field (file order), or None if not found.
返回类型:: GribField or None

Return a new query object with more filter conditions (no I/O).

参数:

parameter -- parameter name (shortName) or a dict of GRIB keys.
level_type -- level type, e.g. "pl", "sfc", "isobaricInhPa", or a dict of GRIB keys.
level -- level value(s). A list (e.g. [850, 500]) matches multiple messages.
count -- 1-based message index in the file; when set, all other conditions are ignored (eccodes engine only).
**kwargs -- extra GRIB keys used as filter conditions (eccodes engine), and engine options: level_dim, field_name, show_progress, lazy (eccodes), with_index (cfgrib).

to_xarray(lazy: bool = False, **kwargs)#

Execute the query and decode all matching fields.

参数:

lazy -- eccodes engine: defer values decoding to data access (see load_field_from_file). May also be given through sel() as an engine option. The cfgrib engine reads through its on-disk index on demand by nature and ignores this option.
**kwargs -- additional filter conditions (merged with those from sel()).

返回:

None -- if no message matches (eccodes engine; the cfgrib engine follows cfgrib/xarray behaviour for empty filters).
xr.DataArray -- if exactly one message matches.
xr.Dataset -- if multiple messages match a single hypercube.
list of xr.Dataset -- if the matches span multiple hypercubes (grouped by level type, following cfgrib.open_datasets).

数据处理#

extract region from gridded data array.

参数:

data
start_longitude
end_longitude
start_latitude
end_latitude
longitude_step
latitude_step

返回类型:

xr.DataArray

Extract a point from 2D field with interpolation.

参数:

data
latitude
longitude
scheme --
interpolate method.
- if engine="xarray", linear or nearest
- if engine="scipy", linear, nearest, splinef2d or rect_bivariate_spline
engine -- interpolate engine, xarray or scipy.
kwargs

返回类型:

xr.DataArray

reki.operator.interpolate_grid(data: DataArray, target: DataArray, scheme: str = 'linear', engine: Literal['scipy', 'xarray'] = 'xarray', **kwargs) → DataArray#

Interpolate grid data into a target grid.

参数:

data (xr.DataArray) -- intput data
target (xr.DataArray) -- target grid
scheme (str) --
interpolate method.
- if engine="xarray", linear or nearest
- if engine="scipy", linear, nearest, splinef2d or rect_bivariate_spline
engine (str) -- interpolate engine, xarray or scipy
kwargs -- key-value parameters to be passed to interpolator with _get_interpolator function.

返回类型:

xr.DataArray

reki.operator.sample_nearest(data: DataArray, longitude_step: float | int, latitude_step: float | int | None = None) → DataArray#

Sample gridded data to a coarser step by nearest (stride) selection.

Each dimension is strided by round(target_step / data_step), anchored at the first grid point, so the output grid is a subset of the input grid and values are kept pointwise (no interpolation). If the target step is not larger than the data step, the input is returned unchanged.

参数:

data -- gridded data on a regular latitude/longitude grid.
longitude_step -- target longitude step, unit degree.
latitude_step -- target latitude step, unit degree. Defaults to longitude_step.

返回类型:

xr.DataArray

数据源与读取器

目录

数据源与读取器#

统一入口#

数据源#

读取器#

数据处理#