helpers

collection of useful functions used across workflows

bbox_to_utm(bbox, *, epsg_src, epsg_dst)

Convert bounding box coordinates to UTM.

Parameters:

Name	Type	Description	Default
bbox	tuple	Tuple containing the lon/lat bounding box coordinates (left, bottom, right, top) in degrees	required
epsg_src	int	EPSG code identifying input bbox coordinate system	required
epsg_dst	int	EPSG code identifying output coordinate system	required

Returns:

Type	Description
tuple	Tuple containing the bounding box coordinates in UTM (meters) (left, bottom, right, top)

Source code in src/compass/utils/helpers.py

def bbox_to_utm(bbox, *, epsg_src, epsg_dst):
    """Convert bounding box coordinates to UTM.

    Parameters
    ----------
    bbox : tuple
        Tuple containing the lon/lat bounding box coordinates
        (left, bottom, right, top) in degrees
    epsg_src : int
        EPSG code identifying input bbox coordinate system
    epsg_dst : int
        EPSG code identifying output coordinate system

    Returns
    -------
    tuple
        Tuple containing the bounding box coordinates in UTM (meters)
        (left, bottom, right, top)
    """
    xmin, ymin, xmax, ymax = bbox
    xys = _convert_to_utm([(xmin, ymin), (xmax, ymax)], epsg_src, epsg_dst)
    return (*xys[0], *xys[1])

burst_bbox_from_db(burst_id, burst_db_file=None, burst_db_conn=None)

Find the bounding box of a burst in the database.

Parameters:

Name	Type	Description	Default
burst_id	str	JPL burst ID	required
burst_db_file	str	Location of burst database sqlite file, by default None	None
burst_db_conn	Connection	Connection object to burst database (If already connected) Alternative to providing burst_db_file, will be faster for multiply queries.	None

Returns:

Name	Type	Description
epsg	int	EPSG code(s) of burst bounding box(es)
bbox	tuple[float]	Bounding box of burst in EPSG coordinates. Bounding box given as tuple(xmin, ymin, xmax, ymax)

Raises:

Type	Description
ValueError	If burst_id is not found in burst database

Source code in src/compass/utils/helpers.py

def burst_bbox_from_db(burst_id, burst_db_file=None, burst_db_conn=None):
    """Find the bounding box of a burst in the database.

    Parameters
    ----------
    burst_id : str
        JPL burst ID
    burst_db_file : str
        Location of burst database sqlite file, by default None
    burst_db_conn : sqlite3.Connection
        Connection object to burst database (If already connected)
        Alternative to providing burst_db_file, will be faster
        for multiply queries.

    Returns
    -------
    epsg : int
        EPSG code(s) of burst bounding box(es)
    bbox : tuple[float]
        Bounding box of burst in EPSG coordinates. Bounding box given as
        tuple(xmin, ymin, xmax, ymax)

    Raises
    ------
    ValueError
        If burst_id is not found in burst database
    """
    # example burst db:
    # /home/staniewi/dev/burst_map_IW_000001_375887.OPERA-JPL.sqlite3
    if burst_db_conn is None:
        burst_db_conn = sqlite3.connect(burst_db_file)
    burst_db_conn.row_factory = sqlite3.Row  # return rows as dicts

    query = "SELECT epsg, xmin, ymin, xmax, ymax FROM burst_id_map WHERE burst_id_jpl = ?"
    cur = burst_db_conn.execute(query, (burst_id,))
    result = cur.fetchone()

    if not result:
        raise ValueError(f"Failed to find {burst_id} in {burst_db_file}")

    epsg = result["epsg"]
    bbox = (result["xmin"], result["ymin"], result["xmax"], result["ymax"])

    return epsg, bbox

burst_bboxes_from_db(burst_ids, burst_db_file=None, burst_db_conn=None)

Find the bounding box of bursts in the database.

Parameters:

Name	Type	Description	Default
burst_id	list[str]	list of JPL burst IDs.	required
burst_db_file	str	Location of burst database sqlite file, by default None	None
burst_db_conn	Connection	Connection object to burst database (If already connected) Alternative to providing burst_db_file, will be faster for multiply queries.	None

Returns:

Name	Type	Description
bboxes	dict	Burst bounding boxes as a dict with burst IDs as key and tuples of EPSG and bounding boxes (tuple[float]) as values. Bounding box given as tuple(xmin, ymin, xmax, ymax)

Raises:

Type	Description
ValueError	If no burst_ids are found in burst database

Source code in src/compass/utils/helpers.py

def burst_bboxes_from_db(burst_ids, burst_db_file=None, burst_db_conn=None):
    """Find the bounding box of bursts in the database.

    Parameters
    ----------
    burst_id : list[str]
        list of JPL burst IDs.
    burst_db_file : str
        Location of burst database sqlite file, by default None
    burst_db_conn : sqlite3.Connection
        Connection object to burst database (If already connected)
        Alternative to providing burst_db_file, will be faster
        for multiply queries.

    Returns
    -------
    bboxes : dict
        Burst bounding boxes as a dict with burst IDs as key and tuples of
        EPSG and bounding boxes (tuple[float]) as values. Bounding box given as
        tuple(xmin, ymin, xmax, ymax)

    Raises
    ------
    ValueError
        If no burst_ids are found in burst database
    """
    # example burst db:
    # /home/staniewi/dev/burst_map_IW_000001_375887.OPERA-JPL.sqlite3
    if burst_db_conn is None:
        burst_db_conn = sqlite3.connect(burst_db_file)
    burst_db_conn.row_factory = sqlite3.Row  # return rows as dicts

    # concatenate '?, ' with for each burst ID for IN query
    qs_in_query = ', '.join('?' for _ in burst_ids)
    query = f"SELECT * FROM burst_id_map WHERE burst_id_jpl IN ({qs_in_query})"
    cur = burst_db_conn.execute(query, burst_ids)
    results = cur.fetchall()

    if not results:
        raise ValueError(f"Failed to find {burst_ids} in {burst_db_file}")

    n_results = len(results)
    epsgs = [[]] * n_results
    bboxes = [[]] * n_results
    burst_ids = [[]] * n_results
    for i_result, result in enumerate(results):
        epsgs[i_result] = result["epsg"]
        bboxes[i_result] = (result["xmin"], result["ymin"],
                           result["xmax"], result["ymax"])
        burst_ids[i_result] = result["burst_id_jpl"]

    # TODO add warning if not all burst bounding boxes found
    return dict(zip(burst_ids, zip(epsgs, bboxes)))

bursts_grouping_generator(bursts)

Dict to group bursts with the same burst ID but different polarizations key: burst ID, value: list[S1BurstSlc]

Parameters:

Name	Type	Description	Default
bursts		List of bursts to grouped	required

Yields:

Name	Type	Description
k	S1BurstId	Burst ID of grouped list of bursts
v	list[Sentinel1BurstSlc]	List of bursts with the same burst ID

Source code in src/compass/utils/helpers.py

def bursts_grouping_generator(bursts):
    '''
    Dict to group bursts with the same burst ID but different polarizations
    key: burst ID, value: list[S1BurstSlc]

    Parameters
    ----------
    bursts: list[Sentinel1BurstSlc]
        List of bursts to grouped

    Yields
    ------
    k: S1BurstId
        Burst ID of grouped list of bursts
    v: list[Sentinel1BurstSlc]
        List of bursts with the same burst ID
    '''
    grouped_bursts = itertools.groupby(bursts, key=lambda b: str(b.burst_id))

    for k, v in grouped_bursts:
        yield k, list(v)

check_dem(dem_path)

Check if given path is a GDAL-compatible file; else raise error

Parameters:

Name	Type	Description	Default
dem_path	str	File path to DEM for which to check GDAL-compatibility	required

Source code in src/compass/utils/helpers.py

def check_dem(dem_path: str):
    """Check if given path is a GDAL-compatible file; else raise error

    Parameters
    ----------
    dem_path : str
        File path to DEM for which to check GDAL-compatibility
    """
    error_channel = journal.error('helpers.check_dem')
    try:
        gdal.Open(dem_path, gdal.GA_ReadOnly)
    except ValueError:
        err_str = f'{dem_path} cannot be opened by GDAL'
        error_channel.log(err_str)
        raise ValueError(err_str)

    epsg = isce3.io.Raster(dem_path).get_epsg()
    if not 1024 <= epsg <= 32767:
        err_str = f'DEM epsg of {epsg} out of bounds'
        error_channel.log(err_str)
        raise ValueError(err_str)

check_directory(file_path)

Check if directory in file_path exists else raise an error.

Parameters:

Name	Type	Description	Default
file_path	str	Path to directory to be checked	required

Source code in src/compass/utils/helpers.py

def check_directory(file_path: str) -> None:
    """Check if directory in file_path exists else raise an error.

    Parameters
    ----------
    file_path: str
       Path to directory to be checked
    """
    error_channel = journal.error('helpers.check_directory')
    if not os.path.isdir(file_path):
        err_str = f'{file_path} not found'
        error_channel.log(err_str)
        raise FileNotFoundError(err_str)

check_file_path(file_path)

Check if file_path exist else raise an error.

Parameters:

Name	Type	Description	Default
file_path	str	Path to file to be checked	required

Source code in src/compass/utils/helpers.py

def check_file_path(file_path: str) -> None:
    """Check if file_path exist else raise an error.

    Parameters
    ----------
    file_path : str
        Path to file to be checked
    """
    error_channel = journal.error('helpers.check_file_path')
    if not os.path.exists(file_path):
        err_str = f'{file_path} not found'
        error_channel.log(err_str)
        raise FileNotFoundError(err_str)

check_url(url)

Check if a resource exists at the given URL.

Parameters:

Name	Type	Description	Default
url		URL to the object	required

Returns:

Name	Type	Description
_	Bool	True if the resource exists in the URL provide; False otherwise

Source code in src/compass/utils/helpers.py

def check_url(url):
    '''
    Check if a resource exists at the given URL.

    Parameters
    ----------
    url: str
        URL to the object

    Returns
    -------
    _: Bool
        `True` if the resource exists in the URL provide; False otherwise
    '''
    error_channel = journal.error('helpers.check_url')
    info_channel = journal.info('helpers.check_url')

    try:
        response = requests.head(url, allow_redirects=True, timeout=30)
        # A 200 OK or 30x redirect status code means the resource exists.
        if response.status_code in range(200, 400):
            info_channel.log(f"Got response {response.status_code}. "
                             f"Resource exists at: {url}")
            return True
        else:
            info_channel.log(f"Got response {response.status_code}. "
                             f"Resource does not exist at: {url}")
            return False
    except requests.exceptions.RequestException as err:
        error_channel.log(f"An error occurred: {err}")
        return False

check_write_dir(dst_path)

Check if given directory is writeable; else raise error.

Parameters:

Name	Type	Description	Default
dst_path	str	File path to directory for which to check writing permission	required

Source code in src/compass/utils/helpers.py

def check_write_dir(dst_path: str):
    """Check if given directory is writeable; else raise error.

    Parameters
    ----------
    dst_path : str
        File path to directory for which to check writing permission
    """
    if not dst_path:
        dst_path = '.'

    error_channel = journal.error('helpers.check_write_dir')

    # check if scratch path exists
    dst_path_ok = os.path.isdir(dst_path)

    if not dst_path_ok:
        try:
            os.makedirs(dst_path, exist_ok=True)
        except OSError:
            err_str = f"Unable to create {dst_path}"
            error_channel.log(err_str)
            raise OSError(err_str)

    # check if path writeable
    write_ok = os.access(dst_path, os.W_OK)
    if not write_ok:
        err_str = f"{dst_path} scratch directory lacks write permission."
        error_channel.log(err_str)
        raise PermissionError(err_str)

deep_update(original, update)

Update default runconfig dict with user-supplied dict.

Parameters:

Name	Type	Description	Default
original	dict	Dict with default options to be updated	required
update		Dict with user-defined options used to update original/default	required

Returns:

Name	Type	Description
original	dict	Default dictionary updated with user-defined options

References

https://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth

Source code in src/compass/utils/helpers.py

def deep_update(original, update):
    """Update default runconfig dict with user-supplied dict.

    Parameters
    ----------
    original : dict
        Dict with default options to be updated
    update: dict
        Dict with user-defined options used to update original/default

    Returns
    -------
    original: dict
        Default dictionary updated with user-defined options

    References
    ----------
    https://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth
    """
    for key, val in update.items():
        if isinstance(val, dict):
            original[key] = deep_update(original.get(key, {}), val)
        else:
            original[key] = val

    # return updated original
    return original

download_url(url, out_file_path)

Download a file from a given URL.

Parameters:

Name	Type	Description	Default
url	str	URL to the object	required
out_file_path	str	Path to the file where the object is to be downloaded	required

Returns:

Name	Type	Description
_	Bool	True if the resource exists in the URL provide; False otherwise

Source code in src/compass/utils/helpers.py

def download_url(url:str, out_file_path:str):
    '''
    Download a file from a given URL.

    Parameters
    ----------
    url: str
        URL to the object
    out_file_path: str
        Path to the file where the object is to be downloaded

    Returns
    -------
    _: Bool
        `True` if the resource exists in the URL provide; `False` otherwise
    '''
    error_channel = journal.error('helpers.download_url')
    info_channel = journal.info('helpers.download_url')

    if not check_url(url):
        error_channel.log(f"Resource does not exist at: {url}")
        return False

    info_channel.log(f"Downloading {url} to {out_file_path}")

    r = requests.get(url, stream=True)
    r.raise_for_status()
    with open(out_file_path, "wb") as f:
        for chunk in r.iter_content(1024*1024):
            f.write(chunk)
    return True

get_file_polarization_mode(file_path)

Check polarization mode from file name

Taking PP from SAFE file name with following format: MMM_BB_TTTR_LFPP_YYYYMMDDTHHMMSS_YYYYMMDDTHHMMSS_OOOOOO_DDDDDD_CCCC.SAFE

Parameters:

Name	Type	Description	Default
file_path	str	SAFE file name to parse	required

Returns:

Name	Type	Description
original	dict	Default dictionary updated with user-defined options

References

https://sentinel.esa.int/web/sentinel/user-guides/sentinel-1-sar/naming-conventions

Source code in src/compass/utils/helpers.py

def get_file_polarization_mode(file_path: str) -> str:
    '''Check polarization mode from file name

    Taking PP from SAFE file name with following format:
    MMM_BB_TTTR_LFPP_YYYYMMDDTHHMMSS_YYYYMMDDTHHMMSS_OOOOOO_DDDDDD_CCCC.SAFE

    Parameters
    ----------
    file_path : str
        SAFE file name to parse

    Returns
    -------
    original: dict
        Default dictionary updated with user-defined options

    References
    ----------
    https://sentinel.esa.int/web/sentinel/user-guides/sentinel-1-sar/naming-conventions
    '''
    # index split tokens from rear to account for R in TTTR being possibly
    # replaced with '_'
    safe_pol_mode = os.path.basename(file_path).split('_')[-6][2:]

    return safe_pol_mode

get_time_delta_str(t_prev)

Helper function that computes difference between current time and a given time object and returns it as a str

Parameters:

Name	Type	Description	Default
t_prev	time	Date and time where a difference is to be computed from	required
_		Difference from current time and t_prev represented as a string	required

Source code in src/compass/utils/helpers.py

def get_time_delta_str(t_prev: time) -> str:
    '''
    Helper function that computes difference between current time and a given
    time object and returns it as a str

    Parameters
    ----------
    t_prev: time
        Date and time where a difference is to be computed from

    _: str
        Difference from current time and t_prev represented as a string
    '''
    return str(timedelta(seconds=time.perf_counter()
                         - t_prev)).split(".", maxsplit=1)[0]

open_raster(filename, band=1)

Return band as numpy array from gdal-friendly raster

Parameters:

Name	Type	Description	Default
filename		Path where is stored GDAL raster to open	required
band		Band number to open	1

Returns:

Name	Type	Description
raster	ndarray	Numpy array containing the raster band to open

Source code in src/compass/utils/helpers.py

def open_raster(filename, band=1):
    '''
    Return band as numpy array from gdal-friendly raster

    Parameters
    ----------
    filename: str
        Path where is stored GDAL raster to open
    band: int
        Band number to open

    Returns
    -------
    raster: np.ndarray
        Numpy array containing the raster band to open
    '''
    error_channel = journal.error('helpers.open_raster')
    if not os.path.isfile(filename):
        err_str = f'{filename} '
        error_channel.log(err_str)
        raise FileNotFoundError(err_str)

    try:
        # The pythonic exception handling for GDAL can be turned on / off
        # The flag of which can be identified by `gdal.GetUseExceptions()`
        # In pythonic exception handling, `gdal.Open()` will raise `RuntimeError`
        # In traditional GDAL, the function does not raise exception buy will return `None`,
        # and the attempts to call GDAL methods will raise `AttributeError`

        ds = gdal.Open(filename, gdal.GA_ReadOnly)
        arr = ds.GetRasterBand(band).ReadAsArray()
        return arr

    except Exception:
        # GDAL reads 1st 2 bytes of ENVI binary to determine file type. If 1st
        # bytes of flat binary is that of a jpeg but the binary is not then
        # GDAL throws a libjpeg runtime error. Follow specifically tries to
        # load as an ENVI file.
        ds = gdal.OpenEx(filename, gdal.OF_VERBOSE_ERROR,
                         allowed_drivers=['ENVI'])
        arr = ds.GetRasterBand(band).ReadAsArray()
        return arr

polygon_to_utm(poly, *, epsg_src, epsg_dst)

Convert a shapely.Polygon's coordinates to UTM.

Parameters:

Name	Type	Description	Default
poly		Polygon object	required
epsg	int	EPSG code identifying output projection system	required

Returns:

Type	Description
tuple	Tuple containing the bounding box coordinates in UTM (meters) (left, bottom, right, top)

Source code in src/compass/utils/helpers.py

def polygon_to_utm(poly, *, epsg_src, epsg_dst):
    """Convert a shapely.Polygon's coordinates to UTM.

    Parameters
    ----------
    poly: shapely.geometry.Polygon
        Polygon object
    epsg : int
        EPSG code identifying output projection system

    Returns
    -------
    tuple
        Tuple containing the bounding box coordinates in UTM (meters)
        (left, bottom, right, top)
    """
    coords = np.array(poly.exterior.coords)
    xys = _convert_to_utm(coords, epsg_src, epsg_dst)
    return geometry.Polygon(xys)

write_raster(filename, data_list, descriptions, data_type=gdal.GDT_Float32, data_format='GTiff')

Write a multiband GDAL-friendly raster to disk. Each dataset allocated in the output file contains a description of the dataset allocated for that band

Parameters:

Name	Type	Description	Default
filename		File path where to store output dataset	required
data_list		List of numpy.ndarray to allocate for each raster band. All datasets within the list are assumed to have the same shape	required
descriptions		List of strings containing a description for the bands to allocate	required
data_type		GDAL dataset type	GDT_Float32
format		Format for GDAL output file	required

Source code in src/compass/utils/helpers.py

def write_raster(filename, data_list, descriptions,
                 data_type=gdal.GDT_Float32, data_format='GTiff'):
    '''
    Write a multiband GDAL-friendly raster to disk.
    Each dataset allocated in the output file contains
    a description of the dataset allocated for that band

    Parameters
    ----------
    filename: str
        File path where to store output dataset
    data_list: list[np.ndarray]
        List of numpy.ndarray to allocate for each
        raster band. All datasets within the list
        are assumed to have the same shape
    descriptions: list[str]
        List of strings containing a description
        for the bands to allocate
    data_type: gdal.dtype
        GDAL dataset type
    format: gdal.Format
        Format for GDAL output file
    '''

    error_channel = journal.error('helpers.write_raster')

    # Check number of datasets match number of descriptions
    if len(data_list) != len(descriptions):
        err_str = f'Number of datasets to write does not match' \
                  f'the number of descriptions ' \
                  f'{len(data_list)} != {len(descriptions)}'
        error_channel.log(err_str)
        raise ValueError(err_str)

    # Get the shape of a dataset within the list. All the datasets
    # are assumed to have the same shape
    length, width = data_list[0].shape
    nbands = len(data_list)

    driver = gdal.GetDriverByName(data_format)
    out_ds = driver.Create(filename, width, length, nbands, data_type)

    band = 0
    for data, description in zip(data_list, descriptions):
        band += 1
        raster_band = out_ds.GetRasterBand(band)
        raster_band.SetDescription(description)
        raster_band.WriteArray(data)

    out_ds.FlushCache()