5975837719

Committed 25 Aug 2023 12:24PM UTC coverage: 85.547% (+0.05%) from 85.502%

Build # 5975837719

Build Type

push

github

Committed by

web-flow

Commit Message

Tiny edit on index page regarding youtube (#1628)

* Update index.rst

* Update index.rst

Run Details

11441 of 13374 relevant lines covered (85.55%)

3.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.16

/oggm/shop/rgitopo.py

import logging

import os
import shutil
import warnings

try:
    import salem
except ImportError:
    pass
try:
    import rasterio
except ImportError:
    pass
import numpy as np

from oggm import utils, workflow
from oggm.exceptions import InvalidParamsError

# Module logger
log = logging.getLogger(__name__)

DEMS_URL = 'https://cluster.klima.uni-bremen.de/data/gdirs/dems_v2/default'
DEMS_HR_URL = 'https://cluster.klima.uni-bremen.de/data/gdirs/dems_v1/highres/'


def init_glacier_directories_from_rgitopo(rgidf=None, dem_source=None,
                                          resolution='default',
                                          keep_dem_folders=False,
                                          reset=False,
                                          force=True):
    """Initialize a glacier directory from an RGI-TOPO DEM.

    Wrapper around :func:`workflow.init_glacier_directories`, which selects
    a default source for you (if not set manually with `dem_source`).

    The default source is NASADEM for all latitudes between 60N and 56S,
    and COPDEM90 elsewhere.

    Parameters
    ----------
    rgidf : GeoDataFrame or list of ids, optional for pre-computed runs
        the RGI glacier outlines. If unavailable, OGGM will parse the
        information from the glacier directories found in the working
        directory. It is required for new runs.
    reset : bool
        delete the existing glacier directories if found.
    force : bool
        setting `reset=True` will trigger a yes/no question to the user. Set
        `force=True` to avoid this.
    dem_source : str
        the source to pick from (default: NASADEM and COPDEM90)
    keep_dem_folders : bool
        the default is to delete the other DEM directories to save space.
        Set this to True to prevent that (e.g. for sensitivity tests)
    Returns
    -------
    gdirs : list of :py:class:`oggm.GlacierDirectory` objects
        the initialised glacier directories
    """

    if resolution == 'default':
        base_url = DEMS_URL
    elif resolution == 'lr':
        base_url = DEMS_URL
    elif resolution == 'hr':
        log.warning('High-res DEMs not available yet in version 2 with COPDEM30')
        base_url = DEMS_HR_URL
    else:
        raise InvalidParamsError('`resolution` should be of `lr` or `hr`')

    gdirs = workflow.init_glacier_directories(rgidf, reset=reset, force=force,
                                              prepro_base_url=base_url,
                                              from_prepro_level=1,
                                              prepro_rgi_version='62')

    workflow.execute_entity_task(select_dem_from_dir, gdirs,
                                 dem_source=dem_source,
                                 keep_dem_folders=keep_dem_folders)

    return gdirs


@utils.entity_task(log, writes=['dem'])
def select_dem_from_dir(gdir, dem_source=None, keep_dem_folders=False):
    """Select a DEM from the ones available in an RGI-TOPO glacier directory.

    Parameters
    ----------
    gdir : :py:class:`oggm.GlacierDirectory`
        the glacier directory
    dem_source : str
        the source to pick from. If 'RGI', we assume that there is a
        `dem_source` attribute in the RGI file.
    keep_dem_folders : bool
        the default is to delete the other DEM directories to save space.
        Set this to True to prevent that (e.g. for sensitivity tests)
    """

    # Start by deleting noise
    for fn in os.listdir(gdir.dir):
        if fn in ['glacier_mask.tif', 'glacier_grid.json',
                  'outlines.tar.gz', 'intersects.tar.gz']:
            continue
        fn = os.path.join(gdir.dir, fn)
        if os.path.isfile(fn):
            os.remove(fn)

    if dem_source == 'RGI':
        dem_source = gdir.rgi_dem_source

    sources = [f.name for f in os.scandir(gdir.dir) if f.is_dir()
               and not f.name.startswith('.')]
    if dem_source not in sources:
        raise RuntimeError('source {} not in folder'.format(dem_source))

    gdir.add_to_diagnostics('dem_source', dem_source)
    shutil.copyfile(os.path.join(gdir.dir, dem_source, 'dem.tif'),
                    gdir.get_filepath('dem'))
    shutil.copyfile(os.path.join(gdir.dir, dem_source, 'dem_source.txt'),
                    gdir.get_filepath('dem_source'))

    if not keep_dem_folders:
        for source in sources:
            shutil.rmtree(os.path.join(gdir.dir, source))


def _fallback_dem_quality_check(gdir):
    return dict(rgi_id=gdir.rgi_id)


@utils.entity_task(log, writes=[], fallback=_fallback_dem_quality_check)
def dem_quality_check(gdir):
    """Run a simple quality check on the rgitopo DEMs

    Parameters
    ----------
    gdir : GlacierDirectory
        the glacier directory

    Returns
    -------
    a dict of DEMSOURCE:frac pairs, where frac is the percentage of
    valid DEM grid points on the glacier.
    """

    with rasterio.open(gdir.get_filepath('glacier_mask'), 'r',
                       driver='GTiff') as ds:
        mask = ds.read(1) == 1
    area = mask.sum()

    sources = [f.name for f in os.scandir(gdir.dir) if f.is_dir()
               and not f.name.startswith('.')]

    out = dict(rgi_id=gdir.rgi_id)
    for s in sources:
        try:
            with rasterio.open(os.path.join(gdir.dir, s, 'dem.tif'), 'r',
                               driver='GTiff') as ds:
                topo = ds.read(1).astype(rasterio.float32)
                topo[topo <= -999.] = np.NaN
                topo[ds.read_masks(1) == 0] = np.NaN
            valid_mask = np.isfinite(topo) & mask
            if np.all(~valid_mask):
                continue
            if np.nanmax(topo[valid_mask]) == 0:
                continue
            out[s] = valid_mask.sum() / area
        except BaseException:
            pass
    return out

1	import logging	8✔
2
3	import os	8✔
4	import shutil	8✔
5	import warnings	8✔
6
7	try:	8✔
8	import salem	8✔
9	except ImportError:
10	pass
11	try:	8✔
12	import rasterio	8✔
13	except ImportError:
14	pass
15	import numpy as np	8✔
16
17	from oggm import utils, workflow	8✔
18	from oggm.exceptions import InvalidParamsError	8✔
19
20	# Module logger
21	log = logging.getLogger(__name__)	8✔
22
23	DEMS_URL = 'https://cluster.klima.uni-bremen.de/data/gdirs/dems_v2/default'	8✔
24	DEMS_HR_URL = 'https://cluster.klima.uni-bremen.de/data/gdirs/dems_v1/highres/'	8✔
25
26
27	def init_glacier_directories_from_rgitopo(rgidf=None, dem_source=None,	8✔
28	resolution='default',
29	keep_dem_folders=False,
30	reset=False,
31	force=True):
32	"""Initialize a glacier directory from an RGI-TOPO DEM.
33
34	Wrapper around :func:`workflow.init_glacier_directories`, which selects
35	a default source for you (if not set manually with `dem_source`).
36
37	The default source is NASADEM for all latitudes between 60N and 56S,
38	and COPDEM90 elsewhere.
39
40	Parameters
41	----------
42	rgidf : GeoDataFrame or list of ids, optional for pre-computed runs
43	the RGI glacier outlines. If unavailable, OGGM will parse the
44	information from the glacier directories found in the working
45	directory. It is required for new runs.
46	reset : bool
47	delete the existing glacier directories if found.
48	force : bool
49	setting `reset=True` will trigger a yes/no question to the user. Set
50	`force=True` to avoid this.
51	dem_source : str
52	the source to pick from (default: NASADEM and COPDEM90)
53	keep_dem_folders : bool
54	the default is to delete the other DEM directories to save space.
55	Set this to True to prevent that (e.g. for sensitivity tests)
56	Returns
57	-------
58	gdirs : list of :py:class:`oggm.GlacierDirectory` objects
59	the initialised glacier directories
60	"""
61
62	if resolution == 'default':	1✔
63	base_url = DEMS_URL	1✔
64	elif resolution == 'lr':	×
65	base_url = DEMS_URL	×
66	elif resolution == 'hr':	×
67	log.warning('High-res DEMs not available yet in version 2 with COPDEM30')	×
68	base_url = DEMS_HR_URL	×
69	else:
70	raise InvalidParamsError('`resolution` should be of `lr` or `hr`')	×
71
72	gdirs = workflow.init_glacier_directories(rgidf, reset=reset, force=force,	1✔
73	prepro_base_url=base_url,
74	from_prepro_level=1,
75	prepro_rgi_version='62')
76
77	workflow.execute_entity_task(select_dem_from_dir, gdirs,	1✔
78	dem_source=dem_source,
79	keep_dem_folders=keep_dem_folders)
80
81	return gdirs	1✔
82
83
84	@utils.entity_task(log, writes=['dem'])	8✔
85	def select_dem_from_dir(gdir, dem_source=None, keep_dem_folders=False):	8✔
86	"""Select a DEM from the ones available in an RGI-TOPO glacier directory.
87
88	Parameters
89	----------
90	gdir : :py:class:`oggm.GlacierDirectory`
91	the glacier directory
92	dem_source : str
93	the source to pick from. If 'RGI', we assume that there is a
94	`dem_source` attribute in the RGI file.
95	keep_dem_folders : bool
96	the default is to delete the other DEM directories to save space.
97	Set this to True to prevent that (e.g. for sensitivity tests)
98	"""
99
100	# Start by deleting noise
101	for fn in os.listdir(gdir.dir):	1✔
102	if fn in ['glacier_mask.tif', 'glacier_grid.json',	1✔
103	'outlines.tar.gz', 'intersects.tar.gz']:
104	continue	1✔
105	fn = os.path.join(gdir.dir, fn)	1✔
106	if os.path.isfile(fn):	1✔
107	os.remove(fn)	1✔
108
109	if dem_source == 'RGI':	1✔
110	dem_source = gdir.rgi_dem_source	×
111
112	sources = [f.name for f in os.scandir(gdir.dir) if f.is_dir()	1✔
113	and not f.name.startswith('.')]
114	if dem_source not in sources:	1✔
115	raise RuntimeError('source {} not in folder'.format(dem_source))	×
116
117	gdir.add_to_diagnostics('dem_source', dem_source)	1✔
118	shutil.copyfile(os.path.join(gdir.dir, dem_source, 'dem.tif'),	1✔
119	gdir.get_filepath('dem'))
120	shutil.copyfile(os.path.join(gdir.dir, dem_source, 'dem_source.txt'),	1✔
121	gdir.get_filepath('dem_source'))
122
123	if not keep_dem_folders:	1✔
124	for source in sources:	1✔
125	shutil.rmtree(os.path.join(gdir.dir, source))	1✔
126
127
128	def _fallback_dem_quality_check(gdir):	8✔
129	return dict(rgi_id=gdir.rgi_id)	×
130
131
132	@utils.entity_task(log, writes=[], fallback=_fallback_dem_quality_check)	8✔
133	def dem_quality_check(gdir):	8✔
134	"""Run a simple quality check on the rgitopo DEMs
135
136	Parameters
137	----------
138	gdir : GlacierDirectory
139	the glacier directory
140
141	Returns
142	-------
143	a dict of DEMSOURCE:frac pairs, where frac is the percentage of
144	valid DEM grid points on the glacier.
145	"""
146
147	with rasterio.open(gdir.get_filepath('glacier_mask'), 'r',	1✔
148	driver='GTiff') as ds:
149	mask = ds.read(1) == 1	1✔
150	area = mask.sum()	1✔
151
152	sources = [f.name for f in os.scandir(gdir.dir) if f.is_dir()	1✔
153	and not f.name.startswith('.')]
154
155	out = dict(rgi_id=gdir.rgi_id)	1✔
156	for s in sources:	1✔
157	try:	1✔
158	with rasterio.open(os.path.join(gdir.dir, s, 'dem.tif'), 'r',	1✔
159	driver='GTiff') as ds:
160	topo = ds.read(1).astype(rasterio.float32)	1✔
161	topo[topo <= -999.] = np.NaN	1✔
162	topo[ds.read_masks(1) == 0] = np.NaN	1✔
163	valid_mask = np.isfinite(topo) & mask	1✔
164	if np.all(~valid_mask):	1✔
165	continue	×
166	if np.nanmax(topo[valid_mask]) == 0:	1✔
167	continue	×
168	out[s] = valid_mask.sum() / area	1✔
169	except BaseException:	×
170	pass	×
171	return out	1✔

OGGM / oggm / 5975837719

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous