sarlo.isce¶
Routines for interacting with the isce library
ALOS1_processing
¶
Downloads and writes files necessary for ISCE Stripmap processing
Attributes:
| Name | Type | Description |
|---|---|---|
self.sensor |
ALOS1 PALSAR |
|
self.file1 |
raw file to be downloaded |
|
self.file2 |
secondary raw file to be downloaded |
|
self.date1 |
acquisition date of raw reference |
|
self.date2 |
acquisition date of raw secondary |
|
self.azimuth_lks |
number of azimuth looks (default 2) |
|
self.range_lks |
number of range looks (default 10) |
|
self.bounds |
bounds for processing Stripmap |
Source code in sarlo/isce.py
class ALOS1_processing(object):
"""Downloads and writes files necessary for ISCE Stripmap processing
Attributes:
self.sensor: ALOS1 PALSAR
self.file1: raw file to be downloaded
self.file2: secondary raw file to be downloaded
self.date1: acquisition date of raw reference
self.date2: acquisition date of raw secondary
self.azimuth_lks: number of azimuth looks (default 2)
self.range_lks: number of range looks (default 10)
self.bounds: bounds for processing Stripmap
"""
def __init__(
self,
file1: str,
file2: str,
azimuth_lks: int = 2,
range_lks: int = 10,
lonMin: int = None,
lonMax: int = None,
latMin: int = None,
latMax: int = None,
):
self.sensor = "ALOS1 PALSAR"
self.file1 = file1
self.date1 = None
self.file2 = file2
self.date2 = None
self.azimuth_lks = azimuth_lks
self.range_lks = range_lks
self.bounds = [lonMin, lonMax, latMin, latMax]
def configure_inputs(outDir: str, self):
"""Write Configuration files for ISCE2 stripmapApp to process NISAR sample products"""
cmd_reference_config = """<component>
<property name="IMAGEFILE">
<value>[data/{0}/{1}/IMG-HH-{2}-H1.0__A]</value>
</property>
<property name="LEADERFILE">
<value>[data/{0}/{1}/LED-{2}-H1.0__A]</value>
</property>
<property name="OUTPUT">
<value>{0}</value>
</property>
</component>""".format(
self.date1, self.file1, self.file1[:-5]
)
print("writing reference.xml")
with open(os.path.join(outDir, "reference.xml"), "w") as fid:
fid.write(cmd_reference_config)
cmd_secondary_config = """<component>
<property name="IMAGEFILE">
<value>[data/{0}/{1}/IMG-HH-{2}-H1.0__A]</value>
</property>
<property name="LEADERFILE">
<value>[data/{0}/{1}/LED-{2}-H1.0__A]</value>
</property>
<property name="OUTPUT">
<value>{0}</value>
</property>
</component>""".format(
self.date2, self.file2, self.file2[:-5]
)
print("writing secondary.xml")
with open(os.path.join(outDir, "secondary.xml"), "w") as fid:
fid.write(cmd_secondary_config)
cmd_stripmap_config = """<?xml version="1.0" encoding="UTF-8"?>
<stripmapApp>
<component name="insar">
<property name="sensor name">ALOS</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do rubbersheetingAzimuth">True</property>
<property name="do rubbersheetingRange">True</property>
<property name="do denseoffsets">True</property>
</component>
</stripmapApp>"""
print("writing stripmapApp.xml")
with open(os.path.join(outDir, "stripmapApp.xml"), "w") as fid:
fid.write(cmd_stripmap_config)
def download_ALOS_raw(self, user: str = "", password: str = ""):
"""Download ALOS PALSAR Raw imagery from ASF
Args:
self.user: username for ASF download
self.password: password for ASF download
Return:
None. Downloads and formats two ALOS PALSAR raw files for interferometric computing
"""
self.user = user
self.password = password
home_dir = os.path.join(os.getenv("HOME"), "work")
process_dir = os.path.join(home_dir, "Tahoe_{0}").format(self.sensor)
data_dir = os.path.join(process_dir, "data")
# If ASF USER/PWD is not provided read it from the .netrc file
if (len(self.password) == 0 | len(self.user) == 0) & (
os.path.exists(os.path.join(os.getenv("HOME"), ".netrc"))
):
netrc_path = os.path.join(os.getenv("HOME"), ".netrc")
count = len(open(netrc_path).readlines())
if count == 1:
file = open(os.path.join(os.getenv("HOME"), ".netrc"), "r")
contents = file.read().split(" ")
self.user = contents[3]
self.password = contents[5]
file.close()
else:
self.user = np.loadtxt(
os.path.join(os.getenv("HOME"), ".netrc"),
skiprows=1,
usecols=1,
dtype=str,
)[0]
self.password = np.loadtxt(
os.path.join(os.getenv("HOME"), ".netrc"),
skiprows=1,
usecols=1,
dtype=str,
)[1]
if (len(self.password) == 0 | len(self.user) == 0) & (
not os.path.exists(os.path.join(os.getenv("HOME"), ".netrc"))
):
print(
"WARNING: The ASF User and Password needs to be included in ~/.netrc file."
)
print(" The ~/.netrc file does not exist or has not been setup properly")
sys.exit("System Exiting until ~/.netrc file setup properly")
# Now lets check if processing and data directories exist
# If they do not we will make the directories
if os.path.exists(process_dir) is False:
print("create ", process_dir)
os.makedirs(process_dir)
else:
print(process_dir, " exists")
if os.path.exists(data_dir) is False:
print("create ", data_dir)
os.makedirs(data_dir)
else:
print(data_dir, " exists")
os.chdir(data_dir)
# Now lets download the data to the data directory
if len(self.file1) == 0 | len(self.file2) == 0:
sys.exit("Data links do not exist")
else:
cmd1 = ("wget %s --user {0} --password {1}" % self.file1).format(
self.user, self.password
)
zip1 = os.path.basename(os.path.normpath(self.file1))
# isolating initial zip file
if not os.path.exists(os.path.join(data_dir, zip1)):
os.system(cmd1)
os.system("unzip %s" % zip1)
else:
print(zip1, " already exists")
cmd2 = ("wget %s --user {0} --password {1}" % self.file2).format(
self.user, self.password
)
zip2 = os.path.basename(
os.path.normpath(self.file2)
) # isolating initial zip file
if not os.path.exists(os.path.join(data_dir, zip2)):
os.system(cmd2)
os.system("unzip %s" % zip2)
else:
print(zip2, " already exists")
SCDT1 = "{0}/{1}".format(zip1[:-4], "%s.l0.workreport" % zip1[:-9])
with open(SCDT1, "r") as myfile:
data = myfile.readlines()
self.date1 = data[6]
self.date1 = self.date1[38:46]
print(self.date1)
SCDT2 = "{0}/{1}".format(zip2[:-4], "%s.l0.workreport" % zip2[:-9])
with open(SCDT2, "r") as myfile:
data = myfile.readlines()
self.date2 = data[6]
self.date2 = self.date2[38:46]
print(self.date2)
if not os.path.exists(os.path.join(data_dir, self.date1)):
os.mkdir(self.date1)
if not os.path.exists(os.path.join(data_dir, self.date2)):
os.mkdir(self.date2)
os.system("mv {0} {1}".format(zip1[:-4], self.date1))
os.system("mv {0} {1}".format(zip2[:-4], self.date2))
configure_inputs(outDir, self)
¶
Write Configuration files for ISCE2 stripmapApp to process NISAR sample products
Source code in sarlo/isce.py
def configure_inputs(outDir: str, self):
"""Write Configuration files for ISCE2 stripmapApp to process NISAR sample products"""
cmd_reference_config = """<component>
<property name="IMAGEFILE">
<value>[data/{0}/{1}/IMG-HH-{2}-H1.0__A]</value>
</property>
<property name="LEADERFILE">
<value>[data/{0}/{1}/LED-{2}-H1.0__A]</value>
</property>
<property name="OUTPUT">
<value>{0}</value>
</property>
</component>""".format(
self.date1, self.file1, self.file1[:-5]
)
print("writing reference.xml")
with open(os.path.join(outDir, "reference.xml"), "w") as fid:
fid.write(cmd_reference_config)
cmd_secondary_config = """<component>
<property name="IMAGEFILE">
<value>[data/{0}/{1}/IMG-HH-{2}-H1.0__A]</value>
</property>
<property name="LEADERFILE">
<value>[data/{0}/{1}/LED-{2}-H1.0__A]</value>
</property>
<property name="OUTPUT">
<value>{0}</value>
</property>
</component>""".format(
self.date2, self.file2, self.file2[:-5]
)
print("writing secondary.xml")
with open(os.path.join(outDir, "secondary.xml"), "w") as fid:
fid.write(cmd_secondary_config)
cmd_stripmap_config = """<?xml version="1.0" encoding="UTF-8"?>
<stripmapApp>
<component name="insar">
<property name="sensor name">ALOS</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do rubbersheetingAzimuth">True</property>
<property name="do rubbersheetingRange">True</property>
<property name="do denseoffsets">True</property>
</component>
</stripmapApp>"""
print("writing stripmapApp.xml")
with open(os.path.join(outDir, "stripmapApp.xml"), "w") as fid:
fid.write(cmd_stripmap_config)
download_ALOS_raw(self, user='', password='')
¶
Download ALOS PALSAR Raw imagery from ASF
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
self.user |
username for ASF download |
required | |
self.password |
password for ASF download |
required |
Returns:
| Type | Description |
|---|---|
None. Downloads and formats two ALOS PALSAR raw files for interferometric computing |
Source code in sarlo/isce.py
def download_ALOS_raw(self, user: str = "", password: str = ""):
"""Download ALOS PALSAR Raw imagery from ASF
Args:
self.user: username for ASF download
self.password: password for ASF download
Return:
None. Downloads and formats two ALOS PALSAR raw files for interferometric computing
"""
self.user = user
self.password = password
home_dir = os.path.join(os.getenv("HOME"), "work")
process_dir = os.path.join(home_dir, "Tahoe_{0}").format(self.sensor)
data_dir = os.path.join(process_dir, "data")
# If ASF USER/PWD is not provided read it from the .netrc file
if (len(self.password) == 0 | len(self.user) == 0) & (
os.path.exists(os.path.join(os.getenv("HOME"), ".netrc"))
):
netrc_path = os.path.join(os.getenv("HOME"), ".netrc")
count = len(open(netrc_path).readlines())
if count == 1:
file = open(os.path.join(os.getenv("HOME"), ".netrc"), "r")
contents = file.read().split(" ")
self.user = contents[3]
self.password = contents[5]
file.close()
else:
self.user = np.loadtxt(
os.path.join(os.getenv("HOME"), ".netrc"),
skiprows=1,
usecols=1,
dtype=str,
)[0]
self.password = np.loadtxt(
os.path.join(os.getenv("HOME"), ".netrc"),
skiprows=1,
usecols=1,
dtype=str,
)[1]
if (len(self.password) == 0 | len(self.user) == 0) & (
not os.path.exists(os.path.join(os.getenv("HOME"), ".netrc"))
):
print(
"WARNING: The ASF User and Password needs to be included in ~/.netrc file."
)
print(" The ~/.netrc file does not exist or has not been setup properly")
sys.exit("System Exiting until ~/.netrc file setup properly")
# Now lets check if processing and data directories exist
# If they do not we will make the directories
if os.path.exists(process_dir) is False:
print("create ", process_dir)
os.makedirs(process_dir)
else:
print(process_dir, " exists")
if os.path.exists(data_dir) is False:
print("create ", data_dir)
os.makedirs(data_dir)
else:
print(data_dir, " exists")
os.chdir(data_dir)
# Now lets download the data to the data directory
if len(self.file1) == 0 | len(self.file2) == 0:
sys.exit("Data links do not exist")
else:
cmd1 = ("wget %s --user {0} --password {1}" % self.file1).format(
self.user, self.password
)
zip1 = os.path.basename(os.path.normpath(self.file1))
# isolating initial zip file
if not os.path.exists(os.path.join(data_dir, zip1)):
os.system(cmd1)
os.system("unzip %s" % zip1)
else:
print(zip1, " already exists")
cmd2 = ("wget %s --user {0} --password {1}" % self.file2).format(
self.user, self.password
)
zip2 = os.path.basename(
os.path.normpath(self.file2)
) # isolating initial zip file
if not os.path.exists(os.path.join(data_dir, zip2)):
os.system(cmd2)
os.system("unzip %s" % zip2)
else:
print(zip2, " already exists")
SCDT1 = "{0}/{1}".format(zip1[:-4], "%s.l0.workreport" % zip1[:-9])
with open(SCDT1, "r") as myfile:
data = myfile.readlines()
self.date1 = data[6]
self.date1 = self.date1[38:46]
print(self.date1)
SCDT2 = "{0}/{1}".format(zip2[:-4], "%s.l0.workreport" % zip2[:-9])
with open(SCDT2, "r") as myfile:
data = myfile.readlines()
self.date2 = data[6]
self.date2 = self.date2[38:46]
print(self.date2)
if not os.path.exists(os.path.join(data_dir, self.date1)):
os.mkdir(self.date1)
if not os.path.exists(os.path.join(data_dir, self.date2)):
os.mkdir(self.date2)
os.system("mv {0} {1}".format(zip1[:-4], self.date1))
os.system("mv {0} {1}".format(zip2[:-4], self.date2))
Sentinel1_processing
¶
Downloads and writes files necessary for ISCE TOPSAR processing
Attributes:
| Name | Type | Description |
|---|---|---|
self.sensor |
S1A or S1B |
|
self.file1 |
slc file to be downloaded |
|
self.file2 |
secondary slc file to be downloaded |
|
self.date1 |
acquisition date of slc reference |
|
self.date2 |
acquisition date of slc secondary |
|
self.azimuth_lks |
number of azimuth looks (default 3) |
|
self.range_lks |
number of range looks (default 7) |
|
self.bounds |
bounds for processing TOPSAR |
|
self.flag |
set to False. Set to true after formation of amplitude to geocode |
Source code in sarlo/isce.py
class Sentinel1_processing(object):
"""Downloads and writes files necessary for ISCE TOPSAR processing
Attributes:
self.sensor: S1A or S1B
self.file1: slc file to be downloaded
self.file2: secondary slc file to be downloaded
self.date1: acquisition date of slc reference
self.date2: acquisition date of slc secondary
self.azimuth_lks: number of azimuth looks (default 3)
self.range_lks: number of range looks (default 7)
self.bounds: bounds for processing TOPSAR
self.flag: set to False. Set to true after formation of amplitude to geocode
"""
def __init__(
self,
query1: str,
query2: str,
lonMin: int,
lonMax: int,
latMin: int,
latMax: int,
azimuth_looks: int = 3,
range_looks: int = 7,
flag: bool = False,
):
"""Intitialze download of SLC files for ISCE processing.
Args:
query1: acquisition date of slc reference
query2: acquisition date of slc secondary
azimuth_lks: number of azimuth looks (default 3)
range_lks: number of range looks (default 7)
lonMin: longitude minimum extent
lonMax: longitude maximum extent
latMin: latitude minimum extent
latMax: latitude maximum extent
flag: set to False. Set to true after formation of amplitude to geocode
Return:
None.
"""
self.query1 = query1
self.query2 = query2
self.date1 = None
self.date2 = None
self.azimuth_lks = azimuth_looks
self.range_lks = range_looks
self.bounds = [latMin, latMax, lonMin, lonMax]
self.flag = flag
# Variables that will not be controlled by user inputs
self.user = None
self.password = None
self.title1 = None
self.title2 = None
self.id1 = None
self.id2 = None
self.mode1 = None
self.mode2 = None
self.frame1 = None
self.frame2 = None
self.relative_orbit1 = None
self.relative_orbit2 = None
self.Satellite_number1 = None
self.Satellite_number2 = None
self.polygon1 = None
self.polygon2 = None
self.center1 = None
self.center2 = None
def download_SLC(self, user: str = None, password: str = None):
"""Download S1 Single Look Complex imagery from ASF
Args:
self.user: string, optional
username for Copernicus download
self.password: string, optional
password for Copernicus download
Return:
None. Downloads and formats two S1 SLC files for interferometric computing
"""
self.user = user
self.password = password
if (len(self.password) is None | len(self.user) is None) & (
not os.path.exists(os.path.join(os.getenv("HOME"), ".netrc"))
):
print(
"WARNING: The Username and Password needs to be included in ~/.netrc file."
)
print(" The ~/.netrc file does not exist or has not been setup properly")
sys.exit("System Exiting until ~/.netrc file setup properly")
# Now lets download the data to the data directory
if len(self.query1) == 0 | len(self.query2) == 0:
sys.exit("Data links do not exist")
else:
bounds = write_polygon_geojson(
lonMin=self.lonMin,
lonMax=self.lonMax,
latMin=self.latMin,
latMax=self.latMax,
)
download_SLC_pair(
bounds,
self.date1,
self.date2,
slcpath=os.path.join(os.getenv("HOME"), "downloads"),
self=self,
username=self.username,
password=self.password,
)
home_dir = os.path.join(os.getenv("HOME"), "work")
self.process_dir = os.path.join(
home_dir,
"S1_{0}_{1}".format(round(self.center1[1]), round(self.center1[0])),
)
data_dir = os.path.join(self.process_dir, "data")
if os.path.exists(self.process_dir) is False:
print("create ", self.process_dir)
os.makedirs(self.process_dir)
else:
print(self.process_dir, " exists")
if os.path.exists(data_dir) is False:
print("create ", data_dir)
os.makedirs(data_dir)
else:
print(data_dir, " exists")
os.chdir(data_dir)
if not os.path.exists(os.path.join(data_dir, "orbits")):
os.mkdir("orbits")
shutil.move(
os.path.join(os.getenv("HOME"), "downloads", "{0}.zip".format(self.title1)),
os.path.join(data_dir, "SLC"),
)
shutil.move(
os.path.join(os.getenv("HOME"), "downloads", "{0}.zip".format(self.title2)),
os.path.join(data_dir, "SLC"),
)
download_eofs(
sentinel_file=os.path.join(data_dir),
save_dir=os.path.join(data_dir, "orbits"),
)
def configure_inputs(self):
"""Writes XML files for TOPSAR processing
Args:
outdir: Directory in which XML files will be written
Return:
None. XML files x3 are written
"""
cmd_reference_config = """<?xml version="1.0" encoding="UTF-8"?>
<component name="reference">
<property name="safe">data/SLC/{0}.zip</property>
<property name="output directory">referencedir</property>
<property name="orbit directory">data/orbits</property>
</component>
""".format(
self.title1
)
print("writing Reference.xml")
with open(os.path.join(self.process_dir, "Reference.xml"), "w") as fid:
fid.write(cmd_reference_config)
cmd_secondary_config = """<?xml version="1.0" encoding="UTF-8"?>
<component name="reference">
<property name="safe">data/SLC/{0}.zip</property>
<property name="output directory">secondarydir</property>
<property name="orbit directory">data/orbits</property>
</component>
""".format(
self.title2
)
print("writing secondary.xml")
with open(os.path.join(self.process_dir, "secondary.xml"), "w") as fid:
fid.write(cmd_secondary_config)
if None in self.bounds is False:
if self.flag is False:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="region of interest">{0}</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{1}</property>
<property name="azimuth looks">{2}</property>
<property name="geocode bounding box">{0}</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.bounds, self.range_lks, self.azimuth_lks
)
elif self.flag is True:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="region of interest">{0}</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{1}</property>
<property name="azimuth looks">{2}</property>
<property name="geocode bounding box">{0}</property>
<property name="geocode">[./merged/resampOnlyImage.amp]</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.bounds, self.range_lks, self.azimuth_lks
)
elif None in self.bounds is True:
if self.flag is False:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{0}</property>
<property name="azimuth looks">{1}</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.range_lks, self.azimuth_lks
)
elif self.flag is True:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{0}</property>
<property name="azimuth looks">{1}</property>
<property name="geocode">[./merged/resampOnlyImage.amp]</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.range_lks, self.azimuth_lks
)
print("writing topsApp.xml")
with open(os.path.join(self.process_dir, "topsApp.xml"), "w") as fid:
fid.write(cmd_topsApp_config)
return
def process_topsApp(self):
os.system(
"topsApp.py topsApp.xml --start=preprocess --end=geocodeoffsets | tee sentinel1.txt"
)
slc1 = os.path.join(self.process_dir, "merged", "reference.slc.full")
slc2 = os.path.join(self.process_dir, "merged", "secondary.slc.full")
# Single Look Complex imagery use real and imaginary numbers. Taking the absolute value of the reference and secondary SLC is how we create the amplitude file that is biweaved
os.system(
"imageMath.py -e 'abs(a);abs(b)' --a {0} --b {1} -o ./merged/topophase.amp.full -s BIP -t Float".format(
slc1, slc2
)
)
# Range and Azimuth looking are required to downsize the data to match topophase.cor.geo
os.system(
"looks.py -i ./merged/topophase.amp.full -o ./merged/topophase.amp -r {0} -a {1}".format(
self.range, self.azimuth
)
)
# Now we just rename topophase.amp to resampOnlyImage.amp
os.system(
"imageMath.py -e='a_0;a_1' --a ./merged/topophase.amp -o ./merged/resampOnlyImage.amp -s BIL -t FLOAT"
)
self.flag = True
Sentinel1_processing.configure_inputs(self)
os.system("topsApp.py topsApp.xml --dostep=geocode")
__init__(self, query1, query2, lonMin, lonMax, latMin, latMax, azimuth_looks=3, range_looks=7, flag=False)
special
¶
Intitialze download of SLC files for ISCE processing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query1 |
str |
acquisition date of slc reference |
required |
query2 |
str |
acquisition date of slc secondary |
required |
azimuth_lks |
number of azimuth looks (default 3) |
required | |
range_lks |
number of range looks (default 7) |
required | |
lonMin |
int |
longitude minimum extent |
required |
lonMax |
int |
longitude maximum extent |
required |
latMin |
int |
latitude minimum extent |
required |
latMax |
int |
latitude maximum extent |
required |
flag |
bool |
set to False. Set to true after formation of amplitude to geocode |
False |
Returns:
| Type | Description |
|---|---|
None. |
Source code in sarlo/isce.py
def __init__(
self,
query1: str,
query2: str,
lonMin: int,
lonMax: int,
latMin: int,
latMax: int,
azimuth_looks: int = 3,
range_looks: int = 7,
flag: bool = False,
):
"""Intitialze download of SLC files for ISCE processing.
Args:
query1: acquisition date of slc reference
query2: acquisition date of slc secondary
azimuth_lks: number of azimuth looks (default 3)
range_lks: number of range looks (default 7)
lonMin: longitude minimum extent
lonMax: longitude maximum extent
latMin: latitude minimum extent
latMax: latitude maximum extent
flag: set to False. Set to true after formation of amplitude to geocode
Return:
None.
"""
self.query1 = query1
self.query2 = query2
self.date1 = None
self.date2 = None
self.azimuth_lks = azimuth_looks
self.range_lks = range_looks
self.bounds = [latMin, latMax, lonMin, lonMax]
self.flag = flag
# Variables that will not be controlled by user inputs
self.user = None
self.password = None
self.title1 = None
self.title2 = None
self.id1 = None
self.id2 = None
self.mode1 = None
self.mode2 = None
self.frame1 = None
self.frame2 = None
self.relative_orbit1 = None
self.relative_orbit2 = None
self.Satellite_number1 = None
self.Satellite_number2 = None
self.polygon1 = None
self.polygon2 = None
self.center1 = None
self.center2 = None
configure_inputs(self)
¶
Writes XML files for TOPSAR processing
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
outdir |
Directory in which XML files will be written |
required |
Returns:
| Type | Description |
|---|---|
None. XML files x3 are written |
Source code in sarlo/isce.py
def configure_inputs(self):
"""Writes XML files for TOPSAR processing
Args:
outdir: Directory in which XML files will be written
Return:
None. XML files x3 are written
"""
cmd_reference_config = """<?xml version="1.0" encoding="UTF-8"?>
<component name="reference">
<property name="safe">data/SLC/{0}.zip</property>
<property name="output directory">referencedir</property>
<property name="orbit directory">data/orbits</property>
</component>
""".format(
self.title1
)
print("writing Reference.xml")
with open(os.path.join(self.process_dir, "Reference.xml"), "w") as fid:
fid.write(cmd_reference_config)
cmd_secondary_config = """<?xml version="1.0" encoding="UTF-8"?>
<component name="reference">
<property name="safe">data/SLC/{0}.zip</property>
<property name="output directory">secondarydir</property>
<property name="orbit directory">data/orbits</property>
</component>
""".format(
self.title2
)
print("writing secondary.xml")
with open(os.path.join(self.process_dir, "secondary.xml"), "w") as fid:
fid.write(cmd_secondary_config)
if None in self.bounds is False:
if self.flag is False:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="region of interest">{0}</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{1}</property>
<property name="azimuth looks">{2}</property>
<property name="geocode bounding box">{0}</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.bounds, self.range_lks, self.azimuth_lks
)
elif self.flag is True:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="region of interest">{0}</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{1}</property>
<property name="azimuth looks">{2}</property>
<property name="geocode bounding box">{0}</property>
<property name="geocode">[./merged/resampOnlyImage.amp]</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.bounds, self.range_lks, self.azimuth_lks
)
elif None in self.bounds is True:
if self.flag is False:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{0}</property>
<property name="azimuth looks">{1}</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.range_lks, self.azimuth_lks
)
elif self.flag is True:
cmd_topsApp_config = """<?xml version="1.0" encoding="UTF-8"?>
<topsApp>
<component name="topsinsar">
<property name="Sensor name">SENTINEL1</property>
<component name="reference">
<catalog>reference.xml</catalog>
</component>
<component name="secondary">
<catalog>secondary.xml</catalog>
</component>
<property name="do interferogram">True</property>
<property name="do denseoffsets">True</property>
<property name="range looks">{0}</property>
<property name="azimuth looks">{1}</property>
<property name="geocode">[./merged/resampOnlyImage.amp]</property>
<property name="do unwrap">True</property>
<property name="unwrapper name">
<value>snaphu_mcf</value>
</property>
</component>
</topsApp>""".format(
self.range_lks, self.azimuth_lks
)
print("writing topsApp.xml")
with open(os.path.join(self.process_dir, "topsApp.xml"), "w") as fid:
fid.write(cmd_topsApp_config)
return
download_SLC(self, user=None, password=None)
¶
Download S1 Single Look Complex imagery from ASF
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
self.user |
string, optional username for Copernicus download |
required | |
self.password |
string, optional password for Copernicus download |
required |
Returns:
| Type | Description |
|---|---|
None. Downloads and formats two S1 SLC files for interferometric computing |
Source code in sarlo/isce.py
def download_SLC(self, user: str = None, password: str = None):
"""Download S1 Single Look Complex imagery from ASF
Args:
self.user: string, optional
username for Copernicus download
self.password: string, optional
password for Copernicus download
Return:
None. Downloads and formats two S1 SLC files for interferometric computing
"""
self.user = user
self.password = password
if (len(self.password) is None | len(self.user) is None) & (
not os.path.exists(os.path.join(os.getenv("HOME"), ".netrc"))
):
print(
"WARNING: The Username and Password needs to be included in ~/.netrc file."
)
print(" The ~/.netrc file does not exist or has not been setup properly")
sys.exit("System Exiting until ~/.netrc file setup properly")
# Now lets download the data to the data directory
if len(self.query1) == 0 | len(self.query2) == 0:
sys.exit("Data links do not exist")
else:
bounds = write_polygon_geojson(
lonMin=self.lonMin,
lonMax=self.lonMax,
latMin=self.latMin,
latMax=self.latMax,
)
download_SLC_pair(
bounds,
self.date1,
self.date2,
slcpath=os.path.join(os.getenv("HOME"), "downloads"),
self=self,
username=self.username,
password=self.password,
)
home_dir = os.path.join(os.getenv("HOME"), "work")
self.process_dir = os.path.join(
home_dir,
"S1_{0}_{1}".format(round(self.center1[1]), round(self.center1[0])),
)
data_dir = os.path.join(self.process_dir, "data")
if os.path.exists(self.process_dir) is False:
print("create ", self.process_dir)
os.makedirs(self.process_dir)
else:
print(self.process_dir, " exists")
if os.path.exists(data_dir) is False:
print("create ", data_dir)
os.makedirs(data_dir)
else:
print(data_dir, " exists")
os.chdir(data_dir)
if not os.path.exists(os.path.join(data_dir, "orbits")):
os.mkdir("orbits")
shutil.move(
os.path.join(os.getenv("HOME"), "downloads", "{0}.zip".format(self.title1)),
os.path.join(data_dir, "SLC"),
)
shutil.move(
os.path.join(os.getenv("HOME"), "downloads", "{0}.zip".format(self.title2)),
os.path.join(data_dir, "SLC"),
)
download_eofs(
sentinel_file=os.path.join(data_dir),
save_dir=os.path.join(data_dir, "orbits"),
)
download_S1_orbits(date1, date2, Satellite_number1, Satellite_number2)
¶
Download S1 orbit files for ISCE processing
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
date1 |
string, date of first S1 SLC file |
required | |
date2 |
string, date of second S1 SLC file |
required | |
sensor1 |
string, S1 sensor A or B |
required | |
sensor2 |
string, S1 sensor A or B |
required |
Returns:
| Type | Description |
|---|---|
None. Downloads orbit files for both SLC files. |
Source code in sarlo/isce.py
def download_S1_orbits(date1, date2, Satellite_number1, Satellite_number2):
"""Download S1 orbit files for ISCE processing
Args:
date1: string, date of first S1 SLC file
date2: string, date of second S1 SLC file
sensor1: string, S1 sensor A or B
sensor2: string, S1 sensor A or B
Return:
None. Downloads orbit files for both SLC files.
"""
if Satellite_number1 == "A":
sensor1 = "S1A"
else:
sensor1 = "S1B"
if Satellite_number2 == "A":
sensor2 = "S1A"
else:
sensor2 = "S1B"
download_eofs([str(date1), str(date2)], [sensor1, sensor2], save_dir="./orbits")
download_SLC_pair(geojson, date1, date2, slcpath='.', self=None, username=None, password=None)
¶
Download S1 SLC files
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
username |
str |
username for scihub.copernicus.eu |
None |
password |
str |
password for scihub.copernicus.eu |
None |
geojson |
geojson with boundary for search area |
required | |
date1 |
str |
slc window for reference slc |
required |
date2 |
str |
slc window for secondary slc |
required |
slcpath |
str |
path to download SLC files |
'.' |
Returns:
| Type | Description |
|---|---|
None. Downloads SLC files. |
Source code in sarlo/isce.py
def download_SLC_pair(
geojson,
date1: str,
date2: str,
slcpath: str = ".",
self=None,
username: str = None,
password: str = None,
):
"""Download S1 SLC files
Args:
username: username for scihub.copernicus.eu
password: password for scihub.copernicus.eu
geojson: geojson with boundary for search area
date1: slc window for reference slc
date2: slc window for secondary slc
slcpath: path to download SLC files
Return:
None. Downloads SLC files.
"""
api = SentinelAPI(username, password, "https://scihub.copernicus.eu/dhus")
footprint = geojson_to_wkt(read_geojson(geojson))
date1 = datetime.datetime.strptime(str(date1, "%Y%m%d"))
enddate1 = datetime.timedelta(days=10) + date1
products = api.query(
area=footprint,
date=(date1, enddate1),
platformname="Sentinel-1",
producttype="SLC",
)
products_df = api.to_geodataframe(products)
products_df_sorted = products_df.sort_values(["ingestiondate"], ascending=[True])
products_df_sorted = products_df_sorted.head(1)
if self is not None:
(
self.title1,
self.date1,
self.id1,
self.mode1,
self.frame1,
self.relative_orbit1,
self.Satellite_number1,
self.polygon1,
self.center1,
) = read_sentinelsat_meta(api, products_df_sorted["uuid"][0])
api.download_all(products=products_df_sorted["uuid"], directory_path=slcpath)
date2 = datetime.datetime.strptime(str(date2, "%Y%m%d"))
enddate2 = datetime.timedelta(days=10) + date2
products = api.query(
area=footprint,
date=(date2, enddate2),
platformname="Sentinel-1",
producttype="SLC",
)
products_df = api.to_geodataframe(products)
products_df_sorted = products_df.sort_values(["ingestiondate"], ascending=[True])
products_df_sorted = products_df_sorted.head(1)
if self is not None:
(
self.title2,
self.date2,
self.id2,
self.mode2,
self.frame2,
self.relative_orbit2,
self.Satellite_number2,
self.polygon2,
self.center2,
) = read_sentinelsat_meta(api, products_df_sorted["uuid"][0])
api.download_all(products=products_df_sorted["uuid"], directory_path=slcpath)
read_sentinelsat_meta(api, id)
¶
Reads the meta
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
api |
object |
class, sentinelsatAPI object |
required |
id |
str |
string, product to be downloaded id |
required |
Returns:
| Type | Description |
|---|---|
title |
S1 file name date: acquisition date id: full product id mode: frame: relative_orbit: Satellite_number: polygon: center: |
Source code in sarlo/isce.py
def read_sentinelsat_meta(api: object, id: str):
"""Reads the meta
Args:
api: class, sentinelsatAPI object
id: string, product to be downloaded id
Returns:
title: S1 file name
date: acquisition date
id: full product id
mode:
frame:
relative_orbit:
Satellite_number:
polygon:
center:
"""
meta = api.get_product_odata(id=id, full=True)
title = meta["title"]
date = meta["date"]
id = meta["id"]
mode = meta["mode"]
frame = meta["Cycle number"]
relative_orbit = meta["Relative orbit (start)"]
Satellite_number = meta["Satellite number"]
polygon = meta["JTS footprint"]
center = center_polygon(polygon_text=polygon)
return (
title,
date,
id,
mode,
frame,
relative_orbit,
Satellite_number,
polygon,
center,
)