# DESCRIPTION
# Agilepy software
#
# NOTICE
# Any information contained in this software
# is property of the AGILE TEAM and is strictly
# private and confidential.
# Copyright (C) 2005-2020 AGILE Team.
# Baroncelli Leonardo <leonardo.baroncelli@inaf.it>
# Addis Antonio <antonio.addis@inaf.it>
# Bulgarelli Andrea <andrea.bulgarelli@inaf.it>
# Parmiggiani Nicolò <nicolo.parmiggiani@inaf.it>
# All rights reserved.
#This program is free software: you can redistribute it and/or modify
#it under the terms of the GNU General Public License as published by
#the Free Software Foundation, either version 3 of the License, or
#(at your option) any later version.
#This program is distributed in the hope that it will be useful,
#but WITHOUT ANY WARRANTY; without even the implied warranty of
#MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
#GNU General Public License for more details.
#You should have received a copy of the GNU General Public License
#along with this program. If not, see <https://www.gnu.org/licenses/>.
from os.path import join
from astropy.wcs import WCS
from astropy.io import fits
import scipy.ndimage as ndimage
import matplotlib.pyplot as plt
from agilepy.utils.Utils import Utils
from agilepy.core.ScienceTools import Cwt2, Met, Ccl
from agilepy.core.AGBaseAnalysis import AGBaseAnalysis
[docs]class AGAnalysisWavelet(AGBaseAnalysis):
"""This class contains the high-level API methods you can use to run wavelet analysis.
This class requires you to setup a ``yaml configuration file`` to specify the software's behaviour.
WARNING: This class is deprecated due to the new Python wavelet tool in development
"""
[docs] def __init__(self, configurationFilePath, sourcesFilePath = None):
"""AGAnalysis constructor.
Args:
configurationFilePath (str): a relative or absolute path to the yaml configuration file.
sourcesFilePath (str, optional): a relative or absolute path to a file containing the description of the sources. Defaults to None. \
Three different types of format are supported: AGILE format (.txt), XML format (.xml) and AGILE catalog files (.multi).
"""
super().__init__(configurationFilePath)
self.config.loadConfigurationsForClass("AGAnalysisWavelet")
self.logger = self.agilepyLogger.getLogger(__name__, "AGAnalysisWavelet")
[docs] @staticmethod
def getConfiguration(confFilePath, userName, outputDir, verboselvl, ctsmap, scaletype, scalemin, scalemax, scalenum, methistsize, cclsizemin, cclsizemax, cclradmin, cclradmax, cclscalemin, cclscalemax):
"""Utility method to create a configuration file.
Args:
confFilePath (str): the path and filename of the configuration file that is going to be created.
userName (str): the username of who is running the software.
outputDir (str): the path to the output directory. The output directory will be created using the following format: 'userName_sourceName_todaydate'
verboselvl (int): the verbosity level of the console output. Message types: level 0 => critical, warning, level 1 => critical, warning, info, level 2 => critical, warning, info, debug
ctsmap (str): the path of cts map generated by AGAnalys class
scaletype (str):
scalemin (float):
scalemax (float):
scalenum (int):
methistsize (float):
cclsizemin (float)
cclsizemax (float)
cclradmin (float)
cclradmax (float)
cclscalemin (float)
cclscalemax (float)
Returns:
None
"""
configuration = """
output:
outdir: %s
filenameprefix: wavelet_product
username: %s
sourcename: wavelet
verboselvl: %d
map:
ctsmap: %s
wavelet:
scaletype: %s
scalemin: %f
scalemax: %f
scalenum: %d
methistsize: %d
cclsizemin: %f
cclsizemax: %f
cclradmin: %f
cclradmax: %f
cclscalemin: %f
cclscalemax: %f
"""%(outputDir, userName, verboselvl, ctsmap, scaletype, scalemin, scalemax, scalenum, methistsize, cclsizemin, cclsizemax, cclradmin, cclradmax, cclscalemin, cclscalemax)
with open(Utils._expandEnvVar(confFilePath),"w") as cf:
cf.write(configuration)
[docs] def waveletAnalysis(self):
"""It runs the three science tools required to perform wavelet analysis, and it creates three directories respectively for any tool
"""
####-------CWT2------------------
self.logger.info("Running CWT2..")
cwt2 = Cwt2("python3 $AGILE/scripts/PYWTOOLS/cwt2d.py", self.logger)
cwt2.configureTool(self.config)
f1 = cwt2.call()
####-------MET--------------
self.logger.info("Running MET..")
met = Met("python3 $AGILE/scripts/PYWTOOLS/met2d.py", self.logger)
extraParams = {"Cwt2OutfilePath":f1}
met.configureTool(self.config, extraParams=extraParams)
f2 = met.call()
####------CCL----------
self.logger.info("Running CCL..")
ccl = Ccl("python3 $AGILE/scripts/PYWTOOLS/ccl2d.py", self.logger)
extraParams = {"MetOutfilePath":f2}
ccl.configureTool(self.config, extraParams=extraParams)
f3 = ccl.call()
return f1[0], f2[0], f3[0]
[docs] def waveletDisplay(self, fitsfile, saveImage=False, multiImage=False):
"""It displays the wevelets generated by waveletAnalysis.
Args:
fitsfile (str): fits filepath
saveImage (bool, optional): if set to true, saves the image into the output directory. It defaults to False.
multiImage (bool): if set to true plots an image for each wavelet
Returns:
It returns the paths to the image files written on disk.
"""
hdu = fits.open(fitsfile)[0]
wcs = WCS(hdu.header)
namefile = fitsfile.split("/")
namefile = namefile[-1]
data = hdu.data
if not multiImage:
fig, ax = plt.subplots(nrows=8, ncols=6, subplot_kw={'projection': wcs, "slices":("x", "y",21)}, figsize=(20, 35))
ax = ax.flatten()
for i in range(hdu.header["NAXIS3"]):
ax[i].imshow(data[i,:,:], origin='lower', norm=None)
if len(hdu.header) == 96: #file wtf
ax[i].set_title("scale: "+ "{:.5f}".format(hdu.header[48+i]))
if saveImage:
filepath = join(self.outdir, "wavelet_analysis/", namefile+".plot.png")
self.logger.info( "plot at: %s", filepath)
plt.savefig(filepath, bbox_inches='tight')
else:
plt.show()
else:
fig, ax = plt.subplots(subplot_kw={'projection': wcs, "slices":("x", "y",21)}, figsize=(10, 10))
for i in range(hdu.header["NAXIS3"]):
ax.imshow(data[i,:,:], origin='lower', norm=None)
if len(hdu.header) == 96: # file wtf
ax.set_title("scale: " + "{:.5f}".format(hdu.header[48+i]))
if saveImage:
filepath = join(self.outdir, "wavelet_analysis/", namefile+str(i)+".plot.png")
self.logger.info( "plot at: %s", filepath)
plt.savefig(filepath)
else:
plt.show()