|
3 | 3 | """ |
4 | 4 |
|
5 | 5 | import json |
6 | | -import re |
7 | 6 | from datetime import datetime, timezone |
8 | 7 |
|
9 | 8 | import importlib_metadata |
10 | 9 | import netCDF4 |
11 | 10 |
|
12 | | -import concatenator |
13 | | - |
14 | 11 | # Values needed for history_json attribute |
15 | 12 | HISTORY_JSON_SCHEMA = "https://harmony.earthdata.nasa.gov/schemas/history/0.1.0/history-v0.1.0.json" |
16 | 13 | PROGRAM = "stitchee" |
17 | 14 | PROGRAM_REF = "https://cmr.earthdata.nasa.gov:443/search/concepts/S2940253910-LARC_CLOUD" |
18 | 15 | VERSION = importlib_metadata.distribution("stitchee").version |
19 | 16 |
|
20 | 17 |
|
21 | | -def regroup_coordinate_attribute(attribute_string: str) -> str: |
22 | | - """ |
23 | | - Examples |
24 | | - -------- |
25 | | - >>> coord_att = "__Time_and_Position__time __Time_and_Position__instrument_fov_latitude __Time_and_Position__instrument_fov_longitude" |
26 | | - >>> flatten_string_with_groups(coord_att) |
27 | | - Time_and_Position/time Time_and_Position/instrument_fov_latitude Time_and_Position/instrument_fov_longitude |
28 | | -
|
29 | | - Parameters |
30 | | - ---------- |
31 | | - attribute_string : str |
32 | | -
|
33 | | - Returns |
34 | | - ------- |
35 | | - str |
36 | | - """ |
37 | | - # Use the separator that's in the attribute string only if all separators in the string are the same. |
38 | | - # Otherwise, we will use our own default separator. |
39 | | - whitespaces = re.findall(r"\s+", attribute_string) |
40 | | - if len(set(whitespaces)) <= 1: |
41 | | - new_sep = whitespaces[0] |
42 | | - else: |
43 | | - new_sep = concatenator.coord_delim |
44 | | - |
45 | | - return new_sep.join( |
46 | | - "/".join(c.split(concatenator.group_delim))[1:] |
47 | | - for c in attribute_string.split() # split on any whitespace |
48 | | - ) |
49 | | - |
50 | | - |
51 | | -def flatten_coordinate_attribute_paths( |
52 | | - dataset: netCDF4.Dataset, var: netCDF4.Variable, variable_name: str |
53 | | -) -> None: |
54 | | - """Flatten the paths of variables referenced in the 'coordinates' attribute.""" |
55 | | - if "coordinates" in var.ncattrs(): |
56 | | - coord_att = var.getncattr("coordinates") |
57 | | - |
58 | | - new_coord_att = flatten_string_with_groups(coord_att) |
59 | | - |
60 | | - dataset.variables[variable_name].setncattr("coordinates", new_coord_att) |
61 | | - |
62 | | - |
63 | | -def flatten_string_with_groups(str_with_groups: str) -> str: |
64 | | - """Determine separator and flatten string specifying group membership via "/". |
65 | | -
|
66 | | - Applies to variable paths or attributes, even for the root level. |
67 | | -
|
68 | | - Examples |
69 | | - -------- |
70 | | - >>> coord_att = "Time_and_Position/time Time_and_Position/instrument_fov_latitude Time_and_Position/instrument_fov_longitude" |
71 | | - >>> flatten_string_with_groups(coord_att) |
72 | | - __Time_and_Position__time __Time_and_Position__instrument_fov_latitude __Time_and_Position__instrument_fov_longitude |
73 | | -
|
74 | | - Parameters |
75 | | - ---------- |
76 | | - str_with_groups : str |
77 | | -
|
78 | | - Returns |
79 | | - ------- |
80 | | - str |
81 | | - """ |
82 | | - # Use the separator that's in the attribute string only if all separators in the string are the same. |
83 | | - # Otherwise, we will use our own default separator. |
84 | | - whitespaces = re.findall(r"\s+", str_with_groups) |
85 | | - if len(set(whitespaces)) == 0: |
86 | | - new_sep = "" |
87 | | - elif len(set(whitespaces)) == 1: |
88 | | - new_sep = whitespaces[0] |
89 | | - else: |
90 | | - new_sep = concatenator.coord_delim |
91 | | - |
92 | | - # A new string is constructed. |
93 | | - return new_sep.join( |
94 | | - f"{concatenator.group_delim}{c.replace('/', concatenator.group_delim)}" |
95 | | - for c in str_with_groups.split() # split on any whitespace |
96 | | - ) |
97 | | - |
98 | | - |
99 | 18 | def retrieve_history(dataset: netCDF4.Dataset) -> dict: |
100 | 19 | """ |
101 | 20 | Retrieve history_json field from NetCDF dataset, if it exists |
|
0 commit comments