from urllib.error import HTTPError
import urllib.request
import json
import numpy as np
import rasterio
import math
import datetime
from PIL import Image, ImageOps
import time
from .__prelude__ import ComputedValue
from .observable import ObservableStruct, ObservableList, dynamic, SELF, EACH
[docs]
class InterferogramNetwork(ObservableStruct):
[docs]
class Interferogram(ObservableStruct):
start_date = dynamic.readonly()
end_date = dynamic.readonly()
selected = dynamic.variable(False)
def __init__(self, start_date, end_date):
super().__init__()
self._start_date = start_date
self._end_date = end_date
[docs]
def strides_date(self, date):
return date >= self.start_date and date <= self.end_date
@property
def start_date_str(self):
return datetime.datetime.strftime(datetime.datetime.fromtimestamp(self.start_date), "%Y%m%d")
@property
def end_date_str(self):
return datetime.datetime.strftime(datetime.datetime.fromtimestamp(self.end_date), "%Y%m%d")
@property
def preview(self):
raise NotImplementedError
interferograms = dynamic.readonly()
date_baselines = dynamic.readonly()
observed_timestamp = dynamic.variable()
[docs]
def preview(self, interfero):
raise NotImplementedError
@dynamic.memo(SELF.interferograms / EACH(SELF), SELF.observed_timestamp)
def selected_interferograms(self):
ts = self.observed_timestamp
if ts is None:
return []
else:
return [interf for interf in self.interferograms if interf.strides_date(ts)]
[docs]
class GDMSarNetwork(InterferogramNetwork):
[docs]
class Interferogram(InterferogramNetwork.Interferogram):
def __init__(self, start_date, end_date, meta_url, unwrapped_preview_url, wrapped_filtered_preview_url, aps_preview_url):
super().__init__(start_date, end_date)
self.meta_url = meta_url
self.unwrapped_preview_url = unwrapped_preview_url
self.wrapped_filtered_preview_url = wrapped_filtered_preview_url
self.aps_preview_url = aps_preview_url
def __repr__(self):
return f"Interferogram({self.start_date}, {self.end_date}, {self.meta_url}, {self.unwrapped_preview_url})"
def __get_preview(self, url):
yield 0.0
with urllib.request.urlopen(url) as preview_file:
image = Image.open(preview_file)
yield 0.5
ret = np.array(ImageOps.contain(image, (256,256)))
yield 1.0
return ret
@dynamic.memo()
def unwrapped_preview(self):
ret = ComputedValue(None)
ret.recompute(lambda: self.__get_preview(self.unwrapped_preview_url))
return ret
@dynamic.memo()
def wrapped_filtered_preview(self):
ret = ComputedValue(None)
ret.recompute(lambda: self.__get_preview(self.wrapped_filtered_preview_url))
return ret
@dynamic.memo()
def aps_preview(self):
ret = ComputedValue(None)
if self.aps_preview_url is not None:
ret.recompute(lambda: self.__get_preview(self.aps_preview_url))
return ret
@dynamic.memo()
def baseline(self):
meta_dict = { }
try:
with urllib.request.urlopen(self.meta_url) as meta_file:
for line in meta_file:
key, val = line.decode('utf-8').split(':', maxsplit=1)
meta_dict[key] = val.strip()
except:
raise Exception(f"Failed to find meta file at {meta_url}")
return float(meta_dict['Perpendicular_baseline'])
def __init__(self, base_url):
super().__init__()
self.base_url = base_url
if self.base_url.startswith('/'):
self.base_url = f"file://{self.base_url}"
self._interferograms = ObservableList()
with urllib.request.urlopen(f"{self.base_url}/backup_product.json") as f:
summary = json.load(f)
for k in summary.keys():
if k.startswith('iw'):
swath = k
break
else:
raise Exception("Could not find a valid swath in backup_product.json")
interferograms = []
for start_end, product in summary[swath].items():
if len(start_end) <= 8 or start_end[8] != '-':
continue
start, end = start_end.split("-")
start_time = datetime.datetime.strptime(start, "%Y%m%d").timestamp()
end_time = datetime.datetime.strptime(end, "%Y%m%d").timestamp()
_, _, meta_file, _, preview_file = product['InU_geo']
meta_url = f"{self.base_url}/{swath}/{meta_file}"
unwrapped_preview_url = f"{self.base_url}/{swath}/{preview_file}"
_, _, _, _, preview_file = product['InWF_geo']
wrapped_filtered_preview_url = f"{self.base_url}/{swath}/{preview_file}"
aps_preview_url = None
if 'APS_geo' in product:
_, _, _, _, preview_file = product['APS_geo']
aps_preview_url = f"{self.base_url}/{swath}/{preview_file}"
interferograms.append(self.Interferogram(
start_time, end_time,
meta_url = meta_url,
unwrapped_preview_url = unwrapped_preview_url,
wrapped_filtered_preview_url = wrapped_filtered_preview_url,
aps_preview_url = aps_preview_url,
))
self._interferograms[:] = interferograms
try:
self._date_baselines = self.__fetch_date_baselines()
except HTTPError:
self._date_baselines = self.__long_compute_baselines()
self.observed_timestamp = list(self._date_baselines.keys())[0]
def __fetch_date_baselines(self):
date_baselines = { }
with urllib.request.urlopen(f"{self.base_url}/baseline.rsc") as f:
for line in f:
date, baseline, *_ = line.decode('utf-8').split()
date_time = datetime.datetime.strptime(date, "%Y%m%d").timestamp()
date_baselines[date_time] = float(baseline)
return date_baselines
def __long_compute_baselines(self):
total_baselines = { }
for interf in self.interferograms:
interf_baseline_delta = interf.baseline
start = interf.start_date
end = interf.end_date
start_master, start_baseline = total_baselines.setdefault(start, (start, 0.0))
end_master, end_baseline = total_baselines.setdefault(end, (start_master, start_baseline + interf_baseline_delta))
if start_master != end_master:
# merge baselines, inefficiently
for date in list(total_baselines.keys()):
date_master, date_baseline = total_baselines[date]
if date_master == end_master:
total_baselines[date] = (start_master, date_baseline + interf_baseline_delta)
return { date: baseline for date, (_, baseline) in total_baselines.items() }