grib2io.utils

  1"""
  2Collection of utility functions to assist in the encoding and decoding
  3of GRIB2 Messages.
  4"""
  5
  6import datetime
  7import struct
  8from typing import Union, Type, Dict, List
  9
 10import numpy as np
 11from numpy.typing import ArrayLike
 12
 13from .. import tables
 14from .. import templates
 15
 16def int2bin(i: int, nbits: int=8, output: Union[Type[str], Type[List]]=str):
 17    """
 18    Convert integer to binary string or list
 19
 20    The struct module unpack using ">i" will unpack a 32-bit integer from a
 21    binary string.
 22
 23    Parameters
 24    ----------
 25    i
 26        Integer value to convert to binary representation.
 27    nbits : default=8
 28        Number of bits to return.  Valid values are 8 [DEFAULT], 16, 32, and
 29        64.
 30    output : default=str
 31        Return data as `str` [DEFAULT] or `list` (list of ints).
 32
 33    Returns
 34    -------
 35    int2bin
 36        `str` or `list` (list of ints) of binary representation of the integer
 37        value.
 38    """
 39    i = int(i) if not isinstance(i,int) else i
 40    assert nbits in [8,16,32,64]
 41    bitstr = "{0:b}".format(i).zfill(nbits)
 42    if output is str:
 43        return bitstr
 44    elif output is list:
 45        return [int(b) for b in bitstr]
 46
 47
 48def ieee_float_to_int(f):
 49    """
 50    Convert an IEEE 754 32-bit float to a 32-bit integer.
 51
 52    Parameters
 53    ----------
 54    f : float
 55        Floating-point value.
 56
 57    Returns
 58    -------
 59    ieee_float_to_int
 60        `numpy.int32` representation of an IEEE 32-bit float.
 61    """
 62    i = struct.unpack('>i',struct.pack('>f',np.float32(f)))[0]
 63    return np.int32(i)
 64
 65
 66def ieee_int_to_float(i):
 67    """
 68    Convert a 32-bit integer to an IEEE 32-bit float.
 69
 70    Parameters
 71    ----------
 72    i : int
 73        Integer value.
 74
 75    Returns
 76    -------
 77    ieee_int_to_float
 78        `numpy.float32` representation of a 32-bit int.
 79    """
 80    f = struct.unpack('>f',struct.pack('>i',np.int32(i)))[0]
 81    return np.float32(f)
 82
 83
 84def get_leadtime(pdtn: int, pdt: ArrayLike) -> datetime.timedelta:
 85    """
 86    Compute lead time as a datetime.timedelta object.
 87
 88    Using information from GRIB2 Product Definition Template
 89    Number, and Product Definition Template (Section 4).
 90
 91    Parameters
 92    ----------
 93    pdtn
 94        GRIB2 Product Definition Template Number
 95    pdt
 96        Sequence containing GRIB2 Product Definition Template (Section 4).
 97
 98    Returns
 99    -------
100    leadTime
101        datetime.timedelta object representing the lead time of the GRIB2 message.
102    """
103    lt = tables.get_value_from_table(pdt[templates.UnitOfForecastTime._key[pdtn]], 'scale_time_seconds')
104    lt *= pdt[templates.ValueOfForecastTime._key[pdtn]]
105    return datetime.timedelta(seconds=int(lt))
106
107
108def get_duration(pdtn: int, pdt: ArrayLike) -> datetime.timedelta:
109    """
110    Compute a time duration as a datetime.timedelta.
111
112    Uses information from Product Definition Template Number, and Product
113    Definition Template (Section 4).
114
115    Parameters
116    ----------
117    pdtn
118        GRIB2 Product Definition Template Number
119    pdt
120        Sequence containing GRIB2 Product Definition Template (Section 4).
121
122    Returns
123    -------
124    get_duration
125        datetime.timedelta object representing the time duration of the GRIB2
126        message.
127    """
128    if pdtn in templates._timeinterval_pdtns:
129        ntime = pdt[templates.NumberOfTimeRanges._key[pdtn]]
130        duration_unit = tables.get_value_from_table(
131            pdt[templates.UnitOfTimeRangeOfStatisticalProcess._key[pdtn]],
132            'scale_time_seconds')
133        d = ntime * duration_unit * pdt[
134            templates.TimeRangeOfStatisticalProcess._key[pdtn]]
135    else:
136        d = 0
137    return datetime.timedelta(seconds=int(d))
138
139
140def decode_wx_strings(lus: bytes) -> Dict[int, str]:
141    """
142    Decode GRIB2 Local Use Section to obtain NDFD/MDL Weather Strings.
143
144    The decode procedure is defined
145    [here](https://vlab.noaa.gov/web/mdl/nbm-gmos-grib2-wx-info).
146
147    Parameters
148    ----------
149    lus
150        GRIB2 Local Use Section containing NDFD weather strings.
151
152    Returns
153    -------
154    decode_wx_strings
155        Dict of NDFD/MDL weather strings. Keys are an integer value that
156        represent the sequential order of the key in the packed local use
157        section and the value is the weather key.
158    """
159    assert lus[0] == 1
160    # Unpack information related to the simple packing method
161    # the packed weather string data.
162    ngroups = struct.unpack('>H',lus[1:3])[0]
163    nvalues = struct.unpack('>i',lus[3:7])[0]
164    refvalue = struct.unpack('>i',lus[7:11])[0]
165    dsf = struct.unpack('>h',lus[11:13])[0]
166    nbits = lus[13]
167    datatype = lus[14]
168    if datatype == 0: # Floating point
169        refvalue = np.float32(ieee_int_to_float(refvalue)*10**-dsf)
170    elif datatype == 1: # Integer
171        refvalue = np.int32(ieee_int_to_float(refvalue)*10**-dsf)
172    # Upack each byte starting at byte 15 to end of the local use
173    # section, create a binary string and append to the full
174    # binary string.
175    b = ''
176    for i in range(15,len(lus)):
177        iword = struct.unpack('>B',lus[i:i+1])[0]
178        b += bin(iword).split('b')[1].zfill(8)
179    # Iterate over the binary string (b). For each nbits
180    # chunk, convert to an integer, including the refvalue,
181    # and then convert the int to an ASCII character, then
182    # concatenate to wxstring.
183    wxstring = ''
184    for i in range(0,len(b),nbits):
185        wxstring += chr(int(b[i:i+nbits],2)+refvalue)
186    # Return string as list, split by null character.
187    #return list(filter(None,wxstring.split('\0')))
188    return {n:k for n,k in enumerate(list(filter(None,wxstring.split('\0'))))}
189
190
191def get_wgrib2_prob_string(
192    probtype: int,
193    sfacl: int,
194    svall: int,
195    sfacu: int,
196    svalu: int,
197) -> str:
198    """
199    Return a wgrib2-styled string of probabilistic threshold information.
200
201    Logic from wgrib2 source,
202    [Prob.c](https://github.com/NOAA-EMC/NCEPLIBS-wgrib2/blob/develop/wgrib2/Prob.c),
203    is replicated here.
204
205    Parameters
206    ----------
207    probtype
208        Type of probability (Code Table 4.9).
209    sfacl
210        Scale factor of lower limit.
211    svall
212        Scaled value of lower limit.
213    sfacu
214        Scale factor of upper limit.
215    svalu
216        Scaled value of upper limit.
217
218    Returns
219    -------
220    get_wgrib2_prob_string
221        wgrib2-formatted string of probability threshold.
222    """
223    probstr = ''
224    if sfacl == -127: sfacl = 0
225    if sfacu == -127: sfacu = 0
226    lower = svall/(10**sfacl)
227    upper = svalu/(10**sfacu)
228    if probtype == 0:
229        probstr = 'prob <%g' % (lower)
230    elif probtype == 1:
231        probstr = 'prob >%g' % (upper)
232    elif probtype == 2:
233        if lower == upper:
234            probstr = 'prob =%g' % (lower)
235        else:
236            probstr = 'prob >=%g <%g' % (lower,upper)
237    elif probtype == 3:
238        probstr = 'prob >%g' % (lower)
239    elif probtype == 4:
240        probstr = 'prob <%g' % (upper)
241    else:
242        probstr = ''
243    return probstr