623 lines
22 KiB
Python
623 lines
22 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
|
||
|
|
|
||
|
|
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||
|
|
# Copyright 2010 California Institute of Technology. ALL RIGHTS RESERVED.
|
||
|
|
#
|
||
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
||
|
|
# you may not use this file except in compliance with the License.
|
||
|
|
# You may obtain a copy of the License at
|
||
|
|
#
|
||
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
||
|
|
#
|
||
|
|
# Unless required by applicable law or agreed to in writing, software
|
||
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
||
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
|
|
# See the License for the specific language governing permissions and
|
||
|
|
# limitations under the License.
|
||
|
|
#
|
||
|
|
# United States Government Sponsorship acknowledged. This software is subject to
|
||
|
|
# U.S. export control laws and regulations and has been classified as 'EAR99 NLR'
|
||
|
|
# (No [Export] License Required except when exporting to an embargoed country,
|
||
|
|
# end user, or in support of a prohibited end use). By downloading this software,
|
||
|
|
# the user agrees to comply with all applicable U.S. export laws and regulations.
|
||
|
|
# The user has the responsibility to obtain export licenses, or other export
|
||
|
|
# authority as may be required before exporting this software to any 'EAR99'
|
||
|
|
# embargoed foreign country or citizen of those countries.
|
||
|
|
#
|
||
|
|
# Author: Walter Szeliga
|
||
|
|
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
import re
|
||
|
|
import os
|
||
|
|
import copy
|
||
|
|
import numpy as np
|
||
|
|
import struct
|
||
|
|
import datetime
|
||
|
|
import logging
|
||
|
|
import isceobj
|
||
|
|
from isceobj import *
|
||
|
|
import ctypes
|
||
|
|
from isceobj.Planet.Planet import Planet
|
||
|
|
from isceobj.Orbit.Orbit import Orbit
|
||
|
|
from isceobj.Orbit.Orbit import StateVector
|
||
|
|
from isceobj.Planet.AstronomicalHandbook import Const
|
||
|
|
from isceobj.Scene.Frame import Frame
|
||
|
|
from isceobj.Scene.Track import Track
|
||
|
|
from iscesys.DateTimeUtil.DateTimeUtil import DateTimeUtil as DTUtil
|
||
|
|
from iscesys.Component.Component import Component
|
||
|
|
|
||
|
|
ORBIT_TYPE = Component.Parameter('_orbitType',
|
||
|
|
public_name='ORBIT_TYPE',
|
||
|
|
default='',
|
||
|
|
type=str,
|
||
|
|
mandatory=True,
|
||
|
|
doc="Options: ODR, PRC, PDS"
|
||
|
|
)
|
||
|
|
|
||
|
|
ORBIT_DIRECTORY = Component.Parameter('_orbitDir',
|
||
|
|
public_name='ORBIT_DIRECTORY',
|
||
|
|
default='',
|
||
|
|
type=str,
|
||
|
|
mandatory=False,
|
||
|
|
doc="Path to the directory containing the orbit files."
|
||
|
|
)
|
||
|
|
|
||
|
|
ORBITFILE = Component.Parameter('_orbitFile',
|
||
|
|
public_name='ORBITFILE',
|
||
|
|
default='',
|
||
|
|
type=str,
|
||
|
|
mandatory=False,
|
||
|
|
doc='Only used with PDS ORBIT_TYPE'
|
||
|
|
)
|
||
|
|
|
||
|
|
IMAGEFILE = Component.Parameter(
|
||
|
|
'_imageFileName',
|
||
|
|
public_name='IMAGEFILE',
|
||
|
|
default='',
|
||
|
|
type=str,
|
||
|
|
mandatory=True,
|
||
|
|
intent='input',
|
||
|
|
doc='Input image file.'
|
||
|
|
)
|
||
|
|
|
||
|
|
from .Sensor import Sensor
|
||
|
|
class ERS_EnviSAT_SLC(Sensor):
|
||
|
|
|
||
|
|
parameter_list = (ORBIT_TYPE,
|
||
|
|
ORBIT_DIRECTORY,
|
||
|
|
ORBITFILE,
|
||
|
|
IMAGEFILE) + Sensor.parameter_list
|
||
|
|
|
||
|
|
"""
|
||
|
|
A Class for parsing ERS instrument and imagery files (Envisat format)
|
||
|
|
"""
|
||
|
|
|
||
|
|
family = 'ers'
|
||
|
|
logging_name = 'isce.sensor.ers_envisat_slc'
|
||
|
|
|
||
|
|
def __init__(self,family='',name=''):
|
||
|
|
super(ERS_EnviSAT_SLC, self).__init__(family if family else self.__class__.family, name=name)
|
||
|
|
self._imageFile = None
|
||
|
|
#self._instrumentFileData = None #none for ERS
|
||
|
|
self._imageryFileData = None
|
||
|
|
self.dopplerRangeTime = None
|
||
|
|
self.rangeRefTime = None
|
||
|
|
self.logger = logging.getLogger("isce.sensor.ERS_EnviSAT_SLC")
|
||
|
|
self.frame = None
|
||
|
|
self.frameList = []
|
||
|
|
|
||
|
|
#NOTE: copied from ERS_SLC.py... only antennaLength used? -SH
|
||
|
|
# Constants are from
|
||
|
|
# J. J. Mohr and S. N. Madsen. Geometric calibration of ERS satellite
|
||
|
|
# SAR images. IEEE T. Geosci. Remote, 39(4):842-850, Apr. 2001.
|
||
|
|
self.constants = {'polarization': 'VV',
|
||
|
|
'antennaLength': 10,
|
||
|
|
'lookDirection': 'RIGHT',
|
||
|
|
'chirpPulseBandwidth': 15.50829e6,
|
||
|
|
'rangeSamplingRate': 18.962468e6,
|
||
|
|
'delayTime':6.622e-6,
|
||
|
|
'iBias': 15.5,
|
||
|
|
'qBias': 15.5}
|
||
|
|
|
||
|
|
|
||
|
|
def getFrame(self):
|
||
|
|
return self.frame
|
||
|
|
|
||
|
|
def parse(self):
|
||
|
|
"""
|
||
|
|
Parse both imagery and create
|
||
|
|
objects representing the platform, instrument and scene
|
||
|
|
"""
|
||
|
|
|
||
|
|
self.frame = Frame()
|
||
|
|
self.frame.configure()
|
||
|
|
|
||
|
|
self._imageFile = ImageryFile(fileName=self._imageFileName)
|
||
|
|
self._imageryFileData = self._imageFile.parse()
|
||
|
|
|
||
|
|
self.populateMetadata()
|
||
|
|
|
||
|
|
def populateMetadata(self):
|
||
|
|
|
||
|
|
self._populatePlatform()
|
||
|
|
self._populateInstrument()
|
||
|
|
self._populateFrame()
|
||
|
|
#self._populateOrbit()
|
||
|
|
if (self._orbitType == 'ODR'):
|
||
|
|
self._populateDelftOrbits()
|
||
|
|
elif (self._orbitType == 'PRC'):
|
||
|
|
self._populatePRCOrbits()
|
||
|
|
elif (self._orbitType == 'PDS'):
|
||
|
|
self._populatePDSOrbits()
|
||
|
|
#else:
|
||
|
|
# self._populateHeaderOrbit() #NOTE: No leader file
|
||
|
|
#NOTE: remove?
|
||
|
|
self.dopplerRangeTime = self._imageryFileData['doppler']
|
||
|
|
self.rangeRefTime = self._imageryFileData['dopplerOrigin'][0] * 1.0e-9
|
||
|
|
# print('Doppler confidence: ', 100.0 * self._imageryFileData['dopplerConfidence'][0])
|
||
|
|
|
||
|
|
def _populatePlatform(self):
|
||
|
|
"""Populate the platform object with metadata"""
|
||
|
|
platform = self.frame.getInstrument().getPlatform()
|
||
|
|
|
||
|
|
# Populate the Platform and Scene objects
|
||
|
|
platform.setMission("ERS")
|
||
|
|
platform.setPointingDirection(-1)
|
||
|
|
platform.setAntennaLength(self.constants['antennaLength'])
|
||
|
|
platform.setPlanet(Planet(pname="Earth"))
|
||
|
|
|
||
|
|
def _populateInstrument(self):
|
||
|
|
"""Populate the instrument object with metadata"""
|
||
|
|
instrument = self.frame.getInstrument()
|
||
|
|
|
||
|
|
rangeSampleSpacing = Const.c/(2*self._imageryFileData['rangeSamplingRate'])
|
||
|
|
pri = self._imageryFileData['pri']
|
||
|
|
|
||
|
|
|
||
|
|
####These shouldnt matter for SLC data since data is already focused.
|
||
|
|
txPulseLength = 512 / 19207680.000000
|
||
|
|
chirpPulseBandwidth = 16.0e6
|
||
|
|
chirpSlope = chirpPulseBandwidth/txPulseLength
|
||
|
|
|
||
|
|
instrument.setRangePixelSize(rangeSampleSpacing)
|
||
|
|
instrument.setPulseLength(txPulseLength)
|
||
|
|
#instrument.setSwath(imageryFileData['SWATH'])
|
||
|
|
instrument.setRadarFrequency(self._imageryFileData['radarFrequency'])
|
||
|
|
instrument.setChirpSlope(chirpSlope)
|
||
|
|
instrument.setRangeSamplingRate(self._imageryFileData['rangeSamplingRate'])
|
||
|
|
instrument.setPulseRepetitionFrequency(1.0/pri)
|
||
|
|
#instrument.setRangeBias(rangeBias)
|
||
|
|
instrument.setInPhaseValue(self.constants['iBias'])
|
||
|
|
instrument.setQuadratureValue(self.constants['qBias'])
|
||
|
|
|
||
|
|
def _populateFrame(self):
|
||
|
|
"""Populate the scene object with metadata"""
|
||
|
|
numberOfLines = self._imageryFileData['numLines']
|
||
|
|
numberOfSamples = self._imageryFileData['numSamples']
|
||
|
|
pri = self._imageryFileData['pri']
|
||
|
|
startingRange = Const.c * float(self._imageryFileData['timeToFirstSample']) * 1.0e-9 / 2.0
|
||
|
|
rangeSampleSpacing = Const.c/(2*self._imageryFileData['rangeSamplingRate'])
|
||
|
|
farRange = startingRange + numberOfSamples*rangeSampleSpacing
|
||
|
|
first_line_utc = datetime.datetime.strptime(self._imageryFileData['FIRST_LINE_TIME'], '%d-%b-%Y %H:%M:%S.%f')
|
||
|
|
center_line_utc = datetime.datetime.strptime(self._imageryFileData['FIRST_LINE_TIME'], '%d-%b-%Y %H:%M:%S.%f')
|
||
|
|
last_line_utc = datetime.datetime.strptime(self._imageryFileData['LAST_LINE_TIME'], '%d-%b-%Y %H:%M:%S.%f')
|
||
|
|
centerTime = DTUtil.timeDeltaToSeconds(last_line_utc-first_line_utc)/2.0
|
||
|
|
center_line_utc = center_line_utc + datetime.timedelta(microseconds=int(centerTime*1e6))
|
||
|
|
|
||
|
|
self.frame.setStartingRange(startingRange)
|
||
|
|
self.frame.setFarRange(farRange)
|
||
|
|
self.frame.setProcessingFacility(self._imageryFileData['PROC_CENTER'])
|
||
|
|
self.frame.setProcessingSystem(self._imageryFileData['SOFTWARE_VER'])
|
||
|
|
self.frame.setTrackNumber(int(self._imageryFileData['REL_ORBIT']))
|
||
|
|
self.frame.setOrbitNumber(int(self._imageryFileData['ABS_ORBIT']))
|
||
|
|
self.frame.setPolarization(self._imageryFileData['MDS1_TX_RX_POLAR'])
|
||
|
|
self.frame.setNumberOfSamples(numberOfSamples)
|
||
|
|
self.frame.setNumberOfLines(numberOfLines)
|
||
|
|
self.frame.setSensingStart(first_line_utc)
|
||
|
|
self.frame.setSensingMid(center_line_utc)
|
||
|
|
self.frame.setSensingStop(last_line_utc)
|
||
|
|
|
||
|
|
def _populateDelftOrbits(self):
|
||
|
|
"""Populate an orbit object with the Delft orbits"""
|
||
|
|
from isceobj.Orbit.ODR import ODR, Arclist
|
||
|
|
self.logger.info("Using Delft Orbits")
|
||
|
|
arclist = Arclist(os.path.join(self._orbitDir,'arclist'))
|
||
|
|
arclist.parse()
|
||
|
|
print(self.frame.getSensingStart())
|
||
|
|
print(arclist)
|
||
|
|
orbitFile = arclist.getOrbitFile(self.frame.getSensingStart())
|
||
|
|
#print(orbitFile)
|
||
|
|
odr = ODR(file=os.path.join(self._orbitDir,orbitFile))
|
||
|
|
|
||
|
|
|
||
|
|
startTimePreInterp = self.frame.getSensingStart() - datetime.timedelta(minutes=60)
|
||
|
|
stopTimePreInterp = self.frame.getSensingStop() + datetime.timedelta(minutes=60)
|
||
|
|
odr.parseHeader(startTimePreInterp,stopTimePreInterp)
|
||
|
|
startTime = self.frame.getSensingStart() - datetime.timedelta(minutes=5)
|
||
|
|
stopTime = self.frame.getSensingStop() + datetime.timedelta(minutes=5)
|
||
|
|
self.logger.debug("Extracting orbits between %s and %s" % (startTime,stopTime))
|
||
|
|
orbit = odr.trimOrbit(startTime,stopTime)
|
||
|
|
self.frame.setOrbit(orbit)
|
||
|
|
|
||
|
|
def _populatePRCOrbits(self):
|
||
|
|
"""Populate an orbit object the D-PAF PRC orbits"""
|
||
|
|
from isceobj.Orbit.PRC import PRC, Arclist
|
||
|
|
self.logger.info("Using PRC Orbits")
|
||
|
|
arclist = Arclist(os.path.join(self._orbitDir,'arclist'))
|
||
|
|
arclist.parse()
|
||
|
|
orbitFile = arclist.getOrbitFile(self.frame.getSensingStart())
|
||
|
|
self.logger.debug("Using file %s" % (orbitFile))
|
||
|
|
prc = PRC(file=os.path.join(self._orbitDir,orbitFile))
|
||
|
|
prc.parse()
|
||
|
|
startTime = self.frame.getSensingStart() - datetime.timedelta(minutes=5)
|
||
|
|
stopTime = self.frame.getSensingStop() + datetime.timedelta(minutes=5)
|
||
|
|
self.logger.debug("Extracting orbits between %s and %s" % (startTime,stopTime))
|
||
|
|
fullOrbit = prc.getOrbit()
|
||
|
|
orbit = fullOrbit.trimOrbit(startTime,stopTime)
|
||
|
|
self.frame.setOrbit(orbit)
|
||
|
|
|
||
|
|
def _populatePDSOrbits(self):
|
||
|
|
"""
|
||
|
|
Populate an orbit object using the ERS-2 PDS format
|
||
|
|
"""
|
||
|
|
from isceobj.Orbit.PDS import PDS
|
||
|
|
self.logger.info("Using PDS Orbits")
|
||
|
|
pds = PDS(file=self._orbitFile)
|
||
|
|
pds.parse()
|
||
|
|
startTime = self.frame.getSensingStart() - datetime.timedelta(minutes=5)
|
||
|
|
stopTime = self.frame.getSensingStop() + datetime.timedelta(minutes=5)
|
||
|
|
self.logger.debug("Extracting orbits between %s and %s" % (startTime,stopTime))
|
||
|
|
fullOrbit = pds.getOrbit()
|
||
|
|
orbit = fullOrbit.trimOrbit(startTime,stopTime)
|
||
|
|
self.frame.setOrbit(orbit)
|
||
|
|
|
||
|
|
|
||
|
|
def _populateImage(self,outname,width,length):
|
||
|
|
|
||
|
|
#farRange = self.frame.getStartingRange() + width*self.frame.getInstrument().getRangeSamplingRate()
|
||
|
|
# Update the NumberOfSamples and NumberOfLines in the Frame object
|
||
|
|
self.frame.setNumberOfSamples(width)
|
||
|
|
self.frame.setNumberOfLines(length)
|
||
|
|
#self.frame.setFarRange(farRange)
|
||
|
|
# Create a RawImage object
|
||
|
|
rawImage = createSlcImage()
|
||
|
|
rawImage.setFilename(outname)
|
||
|
|
rawImage.setAccessMode('read')
|
||
|
|
rawImage.setByteOrder('l')
|
||
|
|
rawImage.setXmin(0)
|
||
|
|
rawImage.setXmax(width)
|
||
|
|
rawImage.setWidth(width)
|
||
|
|
self.frame.setImage(rawImage)
|
||
|
|
|
||
|
|
def extractImage(self):
|
||
|
|
from datetime import datetime as dt
|
||
|
|
import tempfile as tf
|
||
|
|
|
||
|
|
self.parse()
|
||
|
|
width = self._imageryFileData['numSamples']
|
||
|
|
length = self._imageryFileData['numLines']
|
||
|
|
self._imageFile.extractImage(self.output, width, length)
|
||
|
|
self._populateImage(self.output, width, length)
|
||
|
|
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def extractDoppler(self):
|
||
|
|
"""
|
||
|
|
Return the doppler centroid as defined in the ASAR file.
|
||
|
|
"""
|
||
|
|
quadratic = {}
|
||
|
|
|
||
|
|
r0 = self.frame.getStartingRange()
|
||
|
|
dr = self.frame.instrument.getRangePixelSize()
|
||
|
|
width = self.frame.getNumberOfSamples()
|
||
|
|
|
||
|
|
midr = r0 + (width/2.0) * dr
|
||
|
|
midtime = 2 * midr/ Const.c - self.rangeRefTime
|
||
|
|
|
||
|
|
fd_mid = 0.0
|
||
|
|
tpow = midtime
|
||
|
|
for kk in self.dopplerRangeTime:
|
||
|
|
fd_mid += kk * tpow
|
||
|
|
tpow *= midtime
|
||
|
|
|
||
|
|
|
||
|
|
####For insarApp
|
||
|
|
quadratic['a'] = fd_mid/self.frame.getInstrument().getPulseRepetitionFrequency()
|
||
|
|
quadratic['b'] = 0.
|
||
|
|
quadratic['c'] = 0.
|
||
|
|
|
||
|
|
|
||
|
|
####For roiApp
|
||
|
|
####More accurate
|
||
|
|
from isceobj.Util import Poly1D
|
||
|
|
|
||
|
|
coeffs = self.dopplerRangeTime
|
||
|
|
dr = self.frame.getInstrument().getRangePixelSize()
|
||
|
|
rref = 0.5 * Const.c * self.rangeRefTime
|
||
|
|
r0 = self.frame.getStartingRange()
|
||
|
|
norm = 0.5*Const.c/dr
|
||
|
|
|
||
|
|
dcoeffs = []
|
||
|
|
for ind, val in enumerate(coeffs):
|
||
|
|
dcoeffs.append( val / (norm**ind))
|
||
|
|
|
||
|
|
|
||
|
|
poly = Poly1D.Poly1D()
|
||
|
|
poly.initPoly(order=len(coeffs)-1)
|
||
|
|
poly.setMean( (rref - r0)/dr - 1.0)
|
||
|
|
poly.setCoeffs(dcoeffs)
|
||
|
|
|
||
|
|
|
||
|
|
pix = np.linspace(0, self.frame.getNumberOfSamples(), num=len(coeffs)+1)
|
||
|
|
evals = poly(pix)
|
||
|
|
fit = np.polyfit(pix,evals, len(coeffs)-1)
|
||
|
|
self.frame._dopplerVsPixel = list(fit[::-1])
|
||
|
|
print('Doppler Fit: ', fit[::-1])
|
||
|
|
|
||
|
|
|
||
|
|
return quadratic
|
||
|
|
|
||
|
|
|
||
|
|
class BaseEnvisatFile(object):
|
||
|
|
"""Class for parsing common Envisat-format metadata"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.fp = None
|
||
|
|
self.mphLength = 1247
|
||
|
|
self.sphLength = None
|
||
|
|
self.mph = {}
|
||
|
|
self.sph = {}
|
||
|
|
|
||
|
|
def readMPH(self):
|
||
|
|
"""Unpack the Main Product Header (MPH)"""
|
||
|
|
mphString = self.fp.read(self.mphLength)
|
||
|
|
header = mphString.splitlines()
|
||
|
|
for line in header:
|
||
|
|
(key, sep, value) = line.decode('utf8').partition('=')
|
||
|
|
if (key.isspace() == False):
|
||
|
|
value = str.replace(value,'"','')
|
||
|
|
value = str.strip(value)
|
||
|
|
self.mph[key] = value
|
||
|
|
|
||
|
|
# Grab the length of the SPH section
|
||
|
|
self.sphLength = self._extractValue(value=self.mph['SPH_SIZE'], type=int)
|
||
|
|
|
||
|
|
def readSPH(self):
|
||
|
|
"""Unpack the Specific Product Header (SPH)"""
|
||
|
|
self.fp.seek(self.mphLength)
|
||
|
|
sphString = self.fp.read(self.sphLength)
|
||
|
|
header = sphString.splitlines()
|
||
|
|
|
||
|
|
dsSeen = False
|
||
|
|
dataSet = {}
|
||
|
|
dataSets = []
|
||
|
|
# the Specific Product Header is made of up key-value pairs.
|
||
|
|
# At the end of the header, there are a number of data blocks that
|
||
|
|
# represent the data sets that follow. Since their key names are
|
||
|
|
# not unique, we need to capture them in an array and then tack
|
||
|
|
# this array on the dictionary later. These data sets begin with
|
||
|
|
# a key named "DS_NAME"
|
||
|
|
for line in header:
|
||
|
|
(key, sep, value) = line.decode('utf8').partition('=')
|
||
|
|
if (key.isspace() == False):
|
||
|
|
value = str.replace(value,'"','')
|
||
|
|
value = str.strip(value)
|
||
|
|
# Check to see if we are reading a Data Set record
|
||
|
|
if ((key == 'DS_NAME') and (dsSeen == False)):
|
||
|
|
dsSeen = True
|
||
|
|
|
||
|
|
if (dsSeen == False):
|
||
|
|
self.sph[key] = value
|
||
|
|
else:
|
||
|
|
dataSet[key] = value
|
||
|
|
|
||
|
|
if (key == 'DSR_SIZE'):
|
||
|
|
dataSets.append(copy.copy(dataSet))
|
||
|
|
|
||
|
|
self.sph['dataSets'] = dataSets
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def _readAndUnpackData(self, length=None, format=None, type=None, numberOfFields=1):
|
||
|
|
"""
|
||
|
|
Convenience method for reading and unpacking data.
|
||
|
|
|
||
|
|
length is the length of the field in bytes [required]
|
||
|
|
format is the format code to use in struct.unpack() [required]
|
||
|
|
numberOfFields is the number of fields expected from the call to struct.unpack() [default = 1]
|
||
|
|
type is the function through which the output of struct.unpack will be passed [default = None]
|
||
|
|
"""
|
||
|
|
line = self.fp.read(length)
|
||
|
|
data = struct.unpack(format, line)
|
||
|
|
if (numberOfFields == 1):
|
||
|
|
data = data[0]
|
||
|
|
if (type):
|
||
|
|
try:
|
||
|
|
data = type(data)
|
||
|
|
except ValueError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
return data
|
||
|
|
|
||
|
|
def _extractValue(self,value=None,type=None):
|
||
|
|
"""
|
||
|
|
Some MPH and SPH fields have units appended to the value in the form of:
|
||
|
|
124<bytes>. This method strips off the units and returns a value of the
|
||
|
|
correct type.
|
||
|
|
"""
|
||
|
|
matches = re.search("([+-]?[\w\.]+)<[\w/]+>",value)
|
||
|
|
answer = matches.group(1)
|
||
|
|
if (answer == None):
|
||
|
|
print("No Matches Found")
|
||
|
|
return
|
||
|
|
|
||
|
|
if (type != None):
|
||
|
|
answer = type(answer)
|
||
|
|
|
||
|
|
return answer
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class ImageryFile(BaseEnvisatFile):
|
||
|
|
"""Parse an Envisat Imagery File"""
|
||
|
|
|
||
|
|
def __init__(self, fileName=None):
|
||
|
|
BaseEnvisatFile.__init__(self)
|
||
|
|
self.fileName = fileName
|
||
|
|
self.sqLength = 170
|
||
|
|
self.procParamLength = None
|
||
|
|
self.doppParamLength = 55
|
||
|
|
self.chirpParamLength = 1483
|
||
|
|
self.geoParamLength = None
|
||
|
|
|
||
|
|
def parse(self):
|
||
|
|
|
||
|
|
import pprint
|
||
|
|
imageryDict = {}
|
||
|
|
try:
|
||
|
|
self.fp = open(self.fileName, 'rb')
|
||
|
|
except IOError as errs:
|
||
|
|
errno,strerr = errs
|
||
|
|
print("IOError: %s %s" % (strerr,self.fileName))
|
||
|
|
return
|
||
|
|
|
||
|
|
self.readMPH()
|
||
|
|
self.readSPH()
|
||
|
|
|
||
|
|
####Handling software version change in 6.02
|
||
|
|
ver = float(self.mph['SOFTWARE_VER'].strip()[-4:])
|
||
|
|
|
||
|
|
'''
|
||
|
|
if ver < 6.02:
|
||
|
|
print('Old ESA Software version: ', ver)
|
||
|
|
self.procParamLength = 2009
|
||
|
|
self.geoParamLength = 521*12
|
||
|
|
else:
|
||
|
|
print('New ESA Software version: ', ver)
|
||
|
|
self.procParamLength = 10069
|
||
|
|
self.geoParamLength = 521*13
|
||
|
|
'''
|
||
|
|
print('ESA Software Version: ', ver)
|
||
|
|
self.procParamLength = 10069 #new ERS envisat format, despite software version 6.0 has new metadata format?
|
||
|
|
self.geoParamLength = 521*13
|
||
|
|
|
||
|
|
procDict = self.readProcParams()
|
||
|
|
doppDict = self.readDopplerParams()
|
||
|
|
geoDict = self.readGeoParams()
|
||
|
|
self.fp.close()
|
||
|
|
|
||
|
|
imageryDict.update(self.mph)
|
||
|
|
imageryDict.update(self.sph)
|
||
|
|
imageryDict.update(procDict)
|
||
|
|
imageryDict.update(doppDict)
|
||
|
|
imageryDict.update(geoDict)
|
||
|
|
return imageryDict
|
||
|
|
|
||
|
|
|
||
|
|
def getTotalHeaderLength(self):
|
||
|
|
headerLength = self.mphLength + self.sphLength + self.sqLength +\
|
||
|
|
self.procParamLength + self.doppParamLength + self.chirpParamLength +\
|
||
|
|
self.geoParamLength
|
||
|
|
|
||
|
|
return headerLength
|
||
|
|
|
||
|
|
|
||
|
|
def extractImage(self, outname, width, length):
|
||
|
|
try:
|
||
|
|
self.fp = open(self.fileName, 'rb')
|
||
|
|
except IOError as errs:
|
||
|
|
errno,strerr = errs
|
||
|
|
print("IOError: %s %s" % (strerr,self.fileName))
|
||
|
|
return
|
||
|
|
|
||
|
|
self.fp.seek(self.getTotalHeaderLength())
|
||
|
|
|
||
|
|
fout = open(outname, 'wb')
|
||
|
|
for kk in range(length):
|
||
|
|
|
||
|
|
if ((kk+1) %1000 == 0):
|
||
|
|
print('Extracted line: %d'%(kk+1))
|
||
|
|
|
||
|
|
rec = self.fp.read(17)
|
||
|
|
# num = struct.unpack(">L", rec[13:17])[0]
|
||
|
|
line = np.fromfile(self.fp, dtype='>h', count=2*width)
|
||
|
|
line.astype(np.float32).tofile(fout)
|
||
|
|
|
||
|
|
fout.close()
|
||
|
|
self.fp.close()
|
||
|
|
|
||
|
|
return
|
||
|
|
|
||
|
|
|
||
|
|
def readProcParams(self):
|
||
|
|
"""Unpack information from the processing parameters dataset"""
|
||
|
|
headerLength = self.mphLength + self.sphLength + self.sqLength
|
||
|
|
self.fp.seek(headerLength)
|
||
|
|
record = self.fp.read(self.procParamLength)
|
||
|
|
|
||
|
|
procDict = {}
|
||
|
|
procDict['mdsFirstTime'] = struct.unpack(">3L",record[:12])
|
||
|
|
procDict['mdsLastTime'] = struct.unpack(">3L", record[13:25])
|
||
|
|
procDict['timeDiffSensing'] = struct.unpack(">f",record[37:41])[0]
|
||
|
|
procDict['rangeSpacing'] = struct.unpack(">f",record[44:48])[0]
|
||
|
|
procDict['azimuthSpacing'] = struct.unpack(">f",record[48:52])[0]
|
||
|
|
procDict['pri'] = struct.unpack(">f", record[52:56])[0]
|
||
|
|
procDict['numLines'] = int(struct.unpack(">L", record[56:60])[0])
|
||
|
|
procDict['numSamples'] = int(struct.unpack(">L", record[60:64])[0])
|
||
|
|
procDict['timeDiffZeroDoppler'] = struct.unpack(">f", record[73:77])[0]
|
||
|
|
procDict['firstProcSample'] = int(struct.unpack(">L", record[975:979])[0])
|
||
|
|
procDict['referenceRange'] = struct.unpack(">f", record[979:983])[0]
|
||
|
|
procDict['rangeSamplingRate'] = struct.unpack(">f", record[983:987])[0]
|
||
|
|
procDict['radarFrequency'] = struct.unpack(">f", record[987:991])[0]
|
||
|
|
|
||
|
|
procDict['azimuthFMRate'] = struct.unpack(">3f",record[1289:1301])
|
||
|
|
procDict['azimuthFMOrigin'] = struct.unpack(">f", record[1301:1305])[0]
|
||
|
|
|
||
|
|
procDict['averageEllipiseHeight'] = struct.unpack(">f", record[1537:1541])[0]
|
||
|
|
|
||
|
|
####State vectors starting from 1761
|
||
|
|
return procDict
|
||
|
|
|
||
|
|
def readDopplerParams(self):
|
||
|
|
"""Unpack information from the doppler coefficients dataset"""
|
||
|
|
headerLength = self.mphLength + self.sphLength + self.sqLength + self.procParamLength
|
||
|
|
self.fp.seek(headerLength)
|
||
|
|
record = self.fp.read(self.doppParamLength)
|
||
|
|
|
||
|
|
doppDict = {}
|
||
|
|
doppDict['dopTime'] = struct.unpack(">3L", record[:12])
|
||
|
|
doppDict['dopplerOrigin'] = struct.unpack(">f", record[13:17])
|
||
|
|
doppDict['doppler'] = struct.unpack(">5f",record[17:37])
|
||
|
|
doppDict['dopplerConfidence'] = struct.unpack(">f", record[37:41])
|
||
|
|
doppDict['dopplerDeltas'] = struct.unpack(">5h",record[42:52])
|
||
|
|
return doppDict
|
||
|
|
|
||
|
|
|
||
|
|
def readGeoParams(self):
|
||
|
|
'''Unpack information regarding starting range.'''
|
||
|
|
|
||
|
|
headerLength = self.mphLength + self.sphLength + self.sqLength +\
|
||
|
|
self.procParamLength + self.doppParamLength + self.chirpParamLength
|
||
|
|
|
||
|
|
self.fp.seek(headerLength + 25+44)
|
||
|
|
record = self.fp.read(4)
|
||
|
|
|
||
|
|
geoDict = {}
|
||
|
|
geoDict['timeToFirstSample'] = struct.unpack(">f",record)[0]
|
||
|
|
|
||
|
|
return geoDict
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class ImageOutput():
|
||
|
|
"""An object to represent the output struct from asa_im_decode"""
|
||
|
|
|
||
|
|
def __init__(self, samples, lines):
|
||
|
|
self.samples = samples
|
||
|
|
self.lines = lines
|