Merge pull request #17 from ZachKeskinen/main

v0.1.4

build/lib/tests/__init__.py

@@ -0,0 +1 @@
+"""Unit test package for uavsar_pytools."""

build/lib/tests/download_pytest.py

@@ -0,0 +1,31 @@
+"""Tests for `uavsar_pytools` downloading functionality."""
+import sys
+sys.path.append('../')
+
+from uavsar_pytools.download.download import download_image
+import pandas as pd
+from os.path import join, basename, isfile
+import shutil
+import pytest
+
+@pytest.fixture
+def img_download():
+    out_dir = './data/imgs/'
+    in_csv = './data/urls'
+    url = pd.read_csv(in_csv, header = None).sample(1).values[0][0]
+    download_image(url, out_dir, ann = False)
+    return join(out_dir, basename(url))
+
+def test_one(img_download):
+    out_dir = './data/imgs/'
+    in_csv = './data/urls'
+    url = pd.read_csv(in_csv, header = None).sample(1).values[0][0]
+    assert isfile(img_download)
+
+@pytest.fixture
+def clear_dir():
+    tmp_dir = './data/imgs'
+    shutil.rmtree(tmp_dir)
+
+def test_two(clear_dir):
+    assert 1 == 1

build/lib/uavsar_pytools/UavsarImage.py

@@ -0,0 +1,98 @@
+import matplotlib.pyplot as plt
+import os
+import numpy as np
+import logging
+
+from uavsar_pytools.download.download import download_image
+from uavsar_pytools.convert.tiff_conversion import grd_tiff_convert
+
+logging.basicConfig()
+
+class UavsarImage():
+    """
+    Class to handle individual uavsar images. Methods include downloading and converting images.
+
+    Args:
+        url (str) - ASF or JPL url to a single uavsar image
+        work_dir (str) = directory to download images into
+        overwrite (bool) = Do you want to overwrite pre-existing files [Default = False]
+        debug (str) = level of logging (not yet implemented)
+
+    Attributes:
+        binary_fp (str): filepath of downloaded images. Created automatically after downloading.
+        ann_fp = file path to annotation file. Created automatically after downloading.
+        arr (array) = processed numpy array of the image
+        desc (dict) = description of image from annotation file.
+    """
+    binary_fp = None
+    ann_fp = None
+    tiff_dir = None
+    arr = None
+    desc = None
+    type = None
+
+    def __init__(self, url, work_dir, debug = False):
+        self.url = url
+        self.work_dir = work_dir
+        self.debug = debug
+
+    def download(self, sub_dir = 'bin_imgs/', ann = True):
+        """
+        Download an uavsar image from a ASF or JPL url.
+        Args:
+            download_dir (str): directory to download image to. Will be created if it doesn't exists.
+            ann (bool): download associated annotation file? [default = True]
+        """
+        out_dir = os.path.join(self.work_dir,sub_dir)
+        if not os.path.exists(out_dir):
+            os.makedirs(out_dir)
+        self.binary_fp, self.ann_fp = download_image(self.url, output_dir= out_dir, ann = ann)
+
+    def convert_to_tiff(self, binary_fp = None, sub_dir = None, ann_fp = None, overwrite = True):
+        """
+        Converts a binary image file with an associated annotation file to WGS84 geotiff.
+        Args:
+            binary_fp (str): path to input binary file
+            out_dir (str): directory to save geotiff in
+            ann_fp (str): path to UAVSAR annotation file
+            overwrite (bool): overwrite exisiting file [default = False]
+        Returns:
+            self.arr: array of image values
+            self.desc: description of image file from annotation
+            self.type: type of file from binary file path
+        """
+        if sub_dir:
+            out_dir = os.path.join(self.work_dir, sub_dir)
+        else:
+            out_dir = self.work_dir
+
+        if not binary_fp:
+            binary_fp = self.binary_fp
+
+        if not ann_fp:
+            ann_fp = self.ann_fp
+
+        result = grd_tiff_convert(in_fp = binary_fp, out_dir = out_dir, ann_fp = ann_fp, overwrite = overwrite)
+        if len(result) == 3:
+            self.desc, self.arr, self.type = result
+
+    def show(self):
+        """Convenience function to check converted array."""
+        if self.arr != None:
+            if len(self.arr.dtype) == 1:
+                d = self.arr['real']
+            else:
+                d = (self.arr['real']**2 + self.arr['imaginary']**2)**0.5
+            std_low = np.nanmedian(d) - np.nanstd(d)
+            std_high = np.nanmedian(d) + np.nanstd(d)
+            plt.imshow(d, vmin = std_low ,vmax = std_high)
+            plt.title(os.path.basename(self.url))
+            plt.colorbar()
+            plt.show()
+
+
+    def url_to_tiff(self, down_dir = 'bin_imgs/'):
+        """Download binary file from url and convert to WGS84 geotiff."""
+        self.download(sub_dir = down_dir)
+        if self.ann_fp:
+            self.convert_to_tiff()

build/lib/uavsar_pytools/UavsarScene.py

@@ -0,0 +1,138 @@
+import matplotlib.pyplot as plt
+import os
+import numpy as np
+import logging
+
+from uavsar_pytools.download.download import download_image, download_zip
+from uavsar_pytools.convert.file_control import unzip
+from uavsar_pytools.convert.tiff_conversion import grd_tiff_convert
+from uavsar_pytools.UavsarImage import UavsarImage
+
+log = logging.getLogger(__name__)
+logging.basicConfig()
+log.setLevel(logging.DEBUG)
+
+class UavsarScene():
+    """
+    Class to handle uavsar zip directories. Methods include downloading and converting images.
+
+    Args:
+        url (str): ASF or JPL url to a zip uavsar directory
+        work_dir (str): directory to download images into
+        overwrite (bool): Do you want to overwrite pre-existing files [Default = False]
+        debug (str): level of logging (not yet implemented)
+
+    Attributes:
+        zipped_fp (str): filepath to downloaded zip directory. Created automatically after downloading.
+        binary_fps (str): filepaths of downloaded binary images. Created automatically after unzipping.
+        ann_fp: file path to annotation file. Created automatically after unzipping.
+        arr (array): processed numpy array of the image
+        desc (dict): description of image from annotation file.
+    """
+
+    zipped_fp = None
+    ann_fp = None
+    binary_fps = []
+    images = []
+
+    def __init__(self, url, work_dir, debug = False):
+        self.url = url
+        self.work_dir = work_dir
+        self.debug = debug
+
+
+    def download(self, sub_dir = 'tmp/', ann = True):
+        """
+        Download an uavsar image or zip file from a ASF or JPL url.
+        Args:
+            download_dir (str): directory to download image to. Will be created if it doesn't exists.
+            ann (bool): download associated annotation file? [default = True]
+        """
+        out_dir = os.path.join(self.work_dir,sub_dir)
+        if not os.path.exists(out_dir):
+            os.makedirs(out_dir)
+
+        if self.url.split('.')[-1] == 'zip':
+            self.zipped_fp = download_zip(self.url, out_dir)
+        else:
+            log.warning('UavsarScene for zip files. Using UavsarImage for single images.')
+
+    def unzip(self, in_dir = None, sub_dir = 'tmp/bin_imgs/'):
+        """
+        Unpack a zipped directory.
+        Args:
+            in_dir (str): directory to unzip frin
+            sub_dir (str): sub-directory in working directory to unzip into
+        """
+        if not in_dir:
+            if not self.zipped_fp:
+                log.warning('No known zip file for this scene. Please provide.')
+            else:
+                in_dir = self.zipped_fp
+
+        out_dir = os.path.join(self.work_dir, sub_dir)
+
+        if not os.path.exists(out_dir):
+            os.makedirs(out_dir)
+
+        self.binary_fps = unzip(in_dir, out_dir)
+
+    def binary_to_tiffs(self, sub_dir = 'tiffs/', binary_dir = None, ann_fp = None):
+        """
+        Convert a set of binary images to WGS84 geotiffs.
+        Args:
+            sub_dir (str): sub-directory in working directory to put tiffs
+            binary_dir (str): directory containing binary files. Autogenerated from unzipping.
+        """
+        if not binary_dir:
+            if self.binary_fps:
+                binary_fps = self.binary_fps
+            if not self.binary_fps:
+                Exception('No binary files or directory known')
+        else:
+            binary_fps = os.listdir(binary_dir)
+
+        out_dir = os.path.join(self.work_dir, sub_dir)
+
+        if not os.path.exists(out_dir):
+            os.makedirs(out_dir)
+
+        if not ann_fp:
+            ann_fp = [a for a in binary_fps if '.ann' in a][0]
+            if not ann_fp:
+                log.warning('No annotation file found for binary files.')
+
+        binary_img_fps = [f for f in binary_fps if '.ann' not in f]
+        for f in binary_img_fps:
+            desc, array, type = grd_tiff_convert(f, out_dir, ann_fp = ann_fp, overwrite = True)
+            self.images.append({'description': desc, 'array':  array, 'type': type})
+
+    def url_to_tiffs(self):
+        self.download()
+        self.unzip()
+        self.binary_to_tiffs()
+
+
+    def show(self, i):
+        """
+        Convenience function for checking a few images within the zip file for successful conversion.
+        """
+        if len(self.images) > i:
+            array =self.images[i]['array']
+            if len(array.dtype) == 1:
+                d = array['real']
+            else:
+                d = (array['real']**2 + array['imaginary']**2)**0.5
+            std_low = np.nanmedian(d) - np.nanstd(d)
+            std_high = np.nanmedian(d) + np.nanstd(d)
+            plt.imshow(d, vmin = std_low ,vmax = std_high)
+            plt.title(os.path.basename(self.url))
+            plt.colorbar()
+            plt.show()
+
+
+
+
+
+
+

build/lib/uavsar_pytools/convert/file_control.py

@@ -0,0 +1,30 @@
+"""
+File control functions for uavsar_pytools.
+"""
+
+from zipfile import ZipFile
+from tqdm import tqdm
+from os.path import exists, join
+
+def unzip(dir_path, out_dir):
+    """
+    Function to extract zipped directory with tqdm progress bar.
+    From: https://stackoverflow.com/questions/4006970/monitor-zip-file-extraction-python.
+
+    Args:
+        dir_path (string) - path to zipped directory to unpack
+        out_dir (string) - path to directory to extract files to.
+    """
+    assert exists(dir_path), f'Zipped directory at {dir_path} not found.'
+
+    # Open your .zip file
+    with ZipFile(file=dir_path) as zip_file:
+
+        checked_list = [f for f in zip_file.namelist() if not exists(join(out_dir, f))]
+        # Loop over each file
+        if checked_list:
+            for file in tqdm(iterable=checked_list, total=len(zip_file.namelist()), unit = 'file', desc='Unzipping'):
+                # Extract each file to another directory
+                zip_file.extract(member=file, path=out_dir)
+
+    return [join(out_dir, fp) for fp in zip_file.namelist()]