7187113635

Committed 12 Dec 2023 08:49PM UTC coverage: 52.879%. Remained the same

Build # 7187113635

Build Type

push

github

Committed by

ghiggi

Commit Message

Refactor full_product_level and product_level

Run Details

11 of 34 new or added lines in 5 files covered. (32.35%)

3 existing lines in 2 files now uncovered.

2920 of 5522 relevant lines covered (52.88%)

0.53 hits per line

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

0.0

/gpm_api/utils/collocation.py

#!/usr/bin/env python3
"""
Created on Mon Jul 31 17:18:10 2023

@author: ghiggi
"""
import datetime

import xarray as xr

import gpm_api


def _get_collocation_defaults_args(product, variables, groups, version, scan_modes):
    """Get collocation defaults arguments."""
    if scan_modes is None:
        scan_modes = gpm_api.available_scan_modes(product=product, version=version)
    if isinstance(scan_modes, str):
        scan_modes = [scan_modes]
    if product not in gpm_api.available_products(product_categories="PMW"):
        if len(scan_modes) > 1:
            raise ValueError("Multiple scan modes can be specified only for PMW products!")
    # PMW defaults
    if variables is None and groups is None:
        if product in gpm_api.available_products(product_levels="2A", product_categories="PMW"):
            variables = [
                "surfacePrecipitation",
                "mostLikelyPrecipitation",
                "cloudWaterPath",  # kg/m2
                "rainWaterPath",  # kg/m2
                "iceWaterPath",  # kg/m2
            ]
        elif product.startswith("1C"):
            variables = ["Tc"]
        elif product.startswith("1B"):
            variables = ["Tb"]
        else:
            pass
    return scan_modes, variables, groups


def collocate_product(
    ds,
    product,
    product_type="RS",
    version=None,
    scan_modes=None,
    variables=None,
    groups=None,
    verbose=True,
    chunks={},
):
    """Collocate a product on the provided dataset."""
    # Get default collocation arguments
    scan_modes, variables, groups = _get_collocation_defaults_args(
        product=product, variables=variables, groups=groups, version=version, scan_modes=scan_modes
    )

    # Define start_time, end_time around input dataset
    start_time = ds.gpm_api.start_time - datetime.timedelta(minutes=5)
    end_time = ds.gpm_api.end_time + datetime.timedelta(minutes=5)

    # Download PMW products (if necessary)
    gpm_api.download(
        product=product,
        product_type=product_type,
        start_time=start_time,
        end_time=end_time,
        version=version,
        force_download=False,
        verbose=verbose,
    )

    # Read datasets
    list_ds = [
        gpm_api.open_dataset(
            product=product,
            start_time=start_time,
            end_time=end_time,
            # Optional
            version=version,
            variables=variables,
            groups=groups,
            product_type=product_type,
            scan_mode=scan_mode,
            chunks={},
            decode_cf=True,
            prefix_group=False,
        )
        for scan_mode in scan_modes
    ]

    # Remap datasets
    list_remapped = [src_ds.gpm_api.remap_on(ds) for src_ds in list_ds]

    # Concatenate if necessary (PMW case)
    if len(list_remapped) > 1:
        output_ds = xr.concat(list_remapped, dim="pmw_frequency")
    else:
        output_ds = list_remapped[0]

    # Assign attributes
    output_ds.attrs = list_ds[0].attrs
    output_ds.attrs["ScanMode"] = scan_modes

    return output_ds

1	#!/usr/bin/env python3
2	"""	×
3	Created on Mon Jul 31 17:18:10 2023
4
5	@author: ghiggi
6	"""
7	import datetime	×
8
9	import xarray as xr	×
10
11	import gpm_api	×
12
13
14	def _get_collocation_defaults_args(product, variables, groups, version, scan_modes):	×
15	"""Get collocation defaults arguments."""
16	if scan_modes is None:	×
17	scan_modes = gpm_api.available_scan_modes(product=product, version=version)	×
18	if isinstance(scan_modes, str):	×
19	scan_modes = [scan_modes]	×
20	if product not in gpm_api.available_products(product_categories="PMW"):	×
21	if len(scan_modes) > 1:	×
22	raise ValueError("Multiple scan modes can be specified only for PMW products!")	×
23	# PMW defaults
24	if variables is None and groups is None:	×
NEW 25	if product in gpm_api.available_products(product_levels="2A", product_categories="PMW"):	×
UNCOV 26	variables = [	×
27	"surfacePrecipitation",
28	"mostLikelyPrecipitation",
29	"cloudWaterPath", # kg/m2
30	"rainWaterPath", # kg/m2
31	"iceWaterPath", # kg/m2
32	]
33	elif product.startswith("1C"):	×
34	variables = ["Tc"]	×
35	elif product.startswith("1B"):	×
36	variables = ["Tb"]	×
37	else:
38	pass	×
39	return scan_modes, variables, groups	×
40
41
42	def collocate_product(	×
43	ds,
44	product,
45	product_type="RS",
46	version=None,
47	scan_modes=None,
48	variables=None,
49	groups=None,
50	verbose=True,
51	chunks={},
52	):
53	"""Collocate a product on the provided dataset."""
54	# Get default collocation arguments
55	scan_modes, variables, groups = _get_collocation_defaults_args(	×
56	product=product, variables=variables, groups=groups, version=version, scan_modes=scan_modes
57	)
58
59	# Define start_time, end_time around input dataset
60	start_time = ds.gpm_api.start_time - datetime.timedelta(minutes=5)	×
61	end_time = ds.gpm_api.end_time + datetime.timedelta(minutes=5)	×
62
63	# Download PMW products (if necessary)
64	gpm_api.download(	×
65	product=product,
66	product_type=product_type,
67	start_time=start_time,
68	end_time=end_time,
69	version=version,
70	force_download=False,
71	verbose=verbose,
72	)
73
74	# Read datasets
75	list_ds = [	×
76	gpm_api.open_dataset(
77	product=product,
78	start_time=start_time,
79	end_time=end_time,
80	# Optional
81	version=version,
82	variables=variables,
83	groups=groups,
84	product_type=product_type,
85	scan_mode=scan_mode,
86	chunks={},
87	decode_cf=True,
88	prefix_group=False,
89	)
90	for scan_mode in scan_modes
91	]
92
93	# Remap datasets
94	list_remapped = [src_ds.gpm_api.remap_on(ds) for src_ds in list_ds]	×
95
96	# Concatenate if necessary (PMW case)
97	if len(list_remapped) > 1:	×
98	output_ds = xr.concat(list_remapped, dim="pmw_frequency")	×
99	else:
100	output_ds = list_remapped[0]	×
101
102	# Assign attributes
103	output_ds.attrs = list_ds[0].attrs	×
104	output_ds.attrs["ScanMode"] = scan_modes	×
105
106	return output_ds	×

ghiggi / gpm_api / 7187113635

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