# coding=utf-8
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
from .DAORunner import DAORunner
from .OutputProviders import *
from astwro.config import find_opt_file
import astwro.starlist as sl
[docs]class Daophot(DAORunner):
""" **daophot** runner
Object of this class maintains single process of **daophot** and it's working directory.
Methods of this class corresponds to **daophot**'s commands, each of those methods
returns result object providing access to daophot screen output as well as easy access
to files generated by **daophot** command.
Instance attributes:
:var str daophotopt: daophotopt.opt file to be copied
:var DPOP_OPtion OPtion_result: results of command OPtion or initial options reported by `daophot`
:var DPOP_ATtach ATtach_result: results of command ATtach
:var DpOp_SKy SKy_result: results of command SKy
:var DpOp_FInd FInd_result: results of command FInd
:var DpOp_PHotometry PHotometry_result: results of command PHotometry
:var DpOp_PIck PIck_result: results of command PIck
:var DpOp_PSf PSf_result: results of command PSf
:var DpOp_SOrt SOrt_result: results of command SOrt (not implemented)
:var DpOp_SUbstar SUbstar_result: results of command SUbstar
:var str image: image which will be automatically ATTACHed before every run
:var options: options which will be automatically added as OPTION command before every run,
can be either:
dictionary:
>>> dp = Daophot()
>>> dp.options = {'GAIN': 9, 'FI': '6.0'}
iterable of tuples:
>>> dp.options = [('GA', 9.0), ('FITTING RADIUS', '6.0')]
filename string of daophot.opt-formatted file:
>>> dp.options = 'config/pydaophot.opt'
"""
PSF_GAUSSIAN = 1 # type: int
PSF_MOFFAT15 = 2 # type: int
PSF_MOFFAT25 = 3 # type: int
PSF_MOFFAT35 = 4 # type: int
PSF_LORENTZ = 5 # type: int
PSF_PENNY1 = 6 # type: int
PSF_PENNY2 = 7 # type: int
def __init__(self, dir=None, image=None, daophotopt=None, options=None, batch=False):
# type: ([str,object], [str], [str], [list,dict], bool) -> Daophot
"""
:param str dir: pathname or TmpDir object - working directory for daophot,
if None temp dir will be used and deleted on `Daophot.close()`
:param str image: if provided this file will be automatically attached (AT) as first daophot command
setting image property has same effect
:param str daophotopt: daophot.opt file, if None build in default file will be used, can be added later
by `Runner.copy_to_runner_dir(file, 'daophot.opt')`
:type options: list or str
:param options: if provided OPTION command will be automatically attached
setting options property has same effect; list of tuples or dict.
Do not set WATCH PROGRESS to sth else than -2
:param bool batch: whether Daophot have to work in batch mode.
.. attribute:: dir
Runner's directory, object of :class:`astwro.utils.TmpDir`, call ``Daophot.dir.path`` for string path to directory
"""
if daophotopt:
self.daophotopt = daophotopt
else:
self.daophotopt = find_opt_file('daophot.opt', package='pydaophot')
self.image = self.expand_path(image)
self.options = {'WA': -2}
if options:
self.options.update(dict(options))
super(Daophot, self).__init__(dir=dir, batch=batch)
# base implementation of __init__ calls `_reset` also
self._update_executable('daophot')
def _reset(self):
super(Daophot, self)._reset()
self.OPtion_result = None #: Results of command OPtion or initial options reported by `daophot`
self.ATtach_result = None
self.SKy_result = None
self.FInd_result = None
self.PHotometry_result = None
self.PIck_result = None
self.PSf_result = None
self.SOrt_result = None
self.SUbstar_result = None
self.GRoup_result = None
self.NEda_result = None
def __deepcopy__(self, memo):
from copy import deepcopy
new = super(Daophot, self).__deepcopy__(memo)
new.daophotopt = deepcopy(self.daophotopt, memo)
new.image = deepcopy(self.image, memo)
new.options = deepcopy(self.options, memo)
return new
def _pre_run(self, wait):
super(Daophot, self)._pre_run(wait)
if self.options:
self._enqueueOPtions(self.options, on_beginning=True)
if self.image:
self._equeueATtach(self.expand_path(self.image), on_beginning=True)
# just for consume options daophot presents on the beginning
opt_processor = DPOP_OPtion()
if self.OPtion_result is None:
self.OPtion_result = opt_processor
self._insert_processing_step('', output_processor=opt_processor, on_beginning=True)
# Empty lines to unhang daophot after error (otherwise it waits for corrected input)
self._insert_processing_step('\n\n\n')
def _on_exit(self):
pass
def _init_workdir_files(self, dir):
super(Daophot, self)._init_workdir_files(dir)
self.link_to_runner_dir(self.daophotopt, 'daophot.opt')
def _equeueATtach(self, image_file, on_beginning=False):
# type: (str, bool) -> DPOP_ATtach
image_file, _ = self._prepare_input_file(image_file)
processor = DPOP_ATtach()
self._insert_processing_step('ATTACH {}\n'.format(image_file),
output_processor=processor,
on_beginning = on_beginning)
if self.ATtach_result is None or not on_beginning:
self.ATtach_result = processor
return processor
[docs] def set_options(self, options, value=None):
# type: ([str,dict,list], [str,float]) -> None
"""
Set option(s) before run.
Options can be either:
dictionary:
``dp.set_options({'GAIN': 9, 'FI': '6.0'})``
iterable of tuples:
``dp.set_options([('GA', 9.0), ('FITTING RADIUS', '6.0')])``
option key, followed by value in 'value' parameter:
``dp.set_options('GA', 9.0)``
filename string of allstar.opt-formatted file (file will be symlinked as `allstar.opt`):
``dp.set_options('opts/newallstar.opt')``
.. warning:: Do not set `WATCH PROGRESS` to something else than -2
:param options: described above
:param value: value if :attr:`options` is just single key
:return: results object also accessible as :attr:`Daophot.OPtion_result` property
:rtype: DPOP_OPtion
"""
if isinstance(options, str) and value is None: # filename
# allstar operates in his tmp dir and has limited buffer for file path
# so symlink file to its working dir
self.link_to_runner_dir(options, 'daophot.opt')
else:
if self.options is None:
self.options = {}
if value is not None: # single value
options = {options:value}
elif isinstance(options, list):
options = dict(options)
self.options.update(options)
# daophot commands
[docs] def ATtach(self, image_file):
# type: (str) -> DPOP_ATtach
"""
Add daophot ATTACH command to execution queue. Available only in "batch" mode.
If image_file parameter is provided in constructor or by set_image method,
ATtach is enqueued automatically (preferred method
until multiple ATTACH commands needed in "batch" mode).
:param str image_file: image to attach file will be symlinked to work dir as ``"i.fits"``,
if ``None``, `i.fits` (file or symlink) is expected in working dir
:return: results object also accessible as :attr:`ATtach_result` property
:rtype: DPOP_ATtach
"""
if not self.batch_mode:
raise Daophot.RunnerException('ATtach is intented for "batch" mode only. Use image property.', self)
self._get_ready_for_commands() # wait for completion before changes in working dir
return self._equeueATtach(image_file)
def EXit(self):
self._insert_processing_step('EXIT\n', output_processor=DaophotCommandOutputProcessor())
def _enqueueOPtions(self, options, value=None, on_beginning=False):
# type: ([str, dict, list], [str], bool) -> DPOP_OPtion
commands = 'OPT\n'
if isinstance(options, str) and value is None: # filename
# daophot operates in his tmp dir and has limited buffer for file path
# so symlink file to its working dir
self._get_ready_for_commands() # wait for completion before changes in working dir
l_opt, a_opt = self._prepare_input_file(options)
commands += l_opt+'\n\n'
else:
commands += '\n' # answer for filename
if value is not None:
options = [(options,value)] # options is str with option name and value is it's value
elif isinstance(options, dict):
options = options.items() # options is dict
# else options is list of pairs
commands += ''.join('%s=%.2f\n' % (k,float(v)) for k,v in options if v is not None)
commands += '\n'
processor = DPOP_OPtion()
self._insert_processing_step(commands, output_processor=processor, on_beginning=on_beginning)
if self.OPtion_result is None or not on_beginning:
self.OPtion_result = processor
return processor
[docs] def OPtions(self, options, value=None):
# type: ([str, dict, list], [str]) -> DPOP_OPtion
"""
Adds daophot OPTION command to execution queue. Available only in "batch" mode.
Use :meth:`set_options()` for options which are set after daophot process start
Parameter `options` can be either:
dictionary:
>>> dp = Daophot(mode = "batch")
>>> dp.OPtions({'GAIN': 9, 'FI': '6.0'})
iterable of tuples:
>>> dp.OPtions([('GA', 9.0), ('FITTING RADIUS', '6.0')])
option key, followed by value in 'value' parameter:
>>> dp.OPtions('GA', 9.0)
filename string of daophot.opt-formatted file:
>>> dp.OPtions('config/pydaophot.opt')
:param options: described above
:param value: value if `options` is just single key
:return: results object also accessible as :attr:`Daophot.OPtion_result` property
:rtype: DPOP_OPtion
"""
if not self.batch_mode:
raise Daophot.RunnerException('OPtions is intented for "batch" mode only. Use set_options().', self)
self._get_ready_for_commands() # wait for completion before changes in working dir
return self._enqueueOPtions(options, value)
[docs] def SKy(self):
"""
Runs (or adds to execution queue in batch mode) daophot SKY command.
:return: results object also accessible as :attr:`Daophot.SKy_result` property
:rtype: DpOp_SKy
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
commands = 'SKY\n'
processor = DpOp_SKy()
self._insert_processing_step(commands, output_processor=processor)
self.SKy_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def FInd(self, frames_av = 1, frames_sum = 1, starlist_file='i.coo'):
"""
Runs (or adds to execution queue in batch mode) daophot FIND command.
:param int frames_av: averaged frames in image, default: 1
:param int frames_sum: summed frames in image, default: 1
:param str starlist_file: output coo file, default: ``"i.coo"``
:return: results object also accessible as :attr:`Daophot.FInd_result` property
:rtype: DpOp_FInd
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
local_starlist_file, abs_starlist_file = self._prepare_output_file(starlist_file)
commands = 'FIND\n{},{}\n{}\nyes\n'.format(frames_av, frames_sum, local_starlist_file)
processor = DpOp_FInd(starlist_file=abs_starlist_file)
self._insert_processing_step(commands, output_processor=processor)
self.FInd_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def PHotometry(self, photoopt=None, IS=0, OS=0, apertures=None, stars='i.coo', photometry_file='i.ap'):
# type: ([str], float, float, [list], [str,sl.StarList], [str]) -> DpOp_PHotometry
"""
Runs (or adds to execution queue in batch mode) daophot PHOTOMETRY command.
Either :py:attr:`photoopt` or :attr:`IS`, :attr:`OS` and :attr:`apertures` have to be set.
eg.: ``IS=35, OS=50, apertures=[8]``
:param str photoopt: photo.opt file to be used, default: None (provide :attr:`IS`, :attr:`OS` and :attr:`apertures`)
:param float IS: inner sky radius, overwrites :attr:`photoopt` file value IS
:param float OS: outer sky radius, overwrites :attr:`photoopt` file value OS
:param list apertures: apertures radius, up to 12, overwrites :attr:`photoopt` file values A1, A2, ...
:param stars: input list of stars, default: ``"i.coo"``
:type stars: str or StarList
:param str photometry_file: output magnitudes file
:return: results object also accessible as `Daophot.PHotometry_result` property
:rtype: DpOp_PHotometry
.. seealso:: :meth:`NEda`
"""
# TODO: When PSF file is found, new daophot uses it and behaves differently,
# TODO: this should avoided, (there is NEda for that),
# TODO: and not using same name for FITS and PSF by default (avoid finding PSF)
# TODO: session:
# Found PSF file mik.psf
# Profile-fitting photometry (default mik.nst): mik.als
# Star ID file (default mik.lst): mik.coo
# Output file (default mik.ap): mik2.ap
apertures = self._check_apertures(IS, OS, apertures, photoopt)
self._get_ready_for_commands() # wait for completion before changes in working dir
l_popt, a_popt = self._prepare_input_file(photoopt)
l_star, a_star = self._prepare_input_file(stars, astwro.starlist.DAO.XY_FILE)
l_phot, a_phot = self._prepare_output_file(photometry_file)
commands = 'PHOT\n{}\n'.format(l_popt)
if IS != 0:
commands += 'IS={}\n'.format(IS)
if OS != 0:
commands += 'OS={}\n'.format(OS)
for i, ap in enumerate(apertures):
commands += 'A{:X}={}\n'.format(i+1, ap)
commands += '\n{}\n{}\n'.format(l_star, l_phot)
processor = DpOp_PHotometry(photometry_file=a_phot)
self._insert_processing_step(commands, output_processor=processor)
self.PHotometry_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def PIck(self, number_of_stars_to_pick=50, faintest_mag=20.0, photometry='i.ap', picked_stars_file='i.lst'):
# type: (float, float, [str,sl.StarList], str) -> DpOp_PIck
"""
Runs (or adds to execution queue in batch mode) daophot PICK command.
:param int: number_of_stars_to_pick
:param float faintest_mag: instrumental magnitude for the faintest star considered to be picked
:param photometry:
input magnitudes file or :class:`~astwro.starlist.StarList`, usually from aperture photometry done by :meth:`PHotometry`.
:type photometry: str or StarList
:param str picked_stars_file: output list of picked stars, default: i.lst
:return: results object also accessible as :var:`Daophot.PIck_result` property
:rtype: DpOp_PIck
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
l_phot, a_phot = self._prepare_input_file(photometry)
l_psfs, a_psfs = self._prepare_output_file(picked_stars_file)
commands = 'PICK\n{}\n{:d},{:f}\n{}\n'.format(
l_phot,
number_of_stars_to_pick,
faintest_mag,
l_psfs
)
processor = DpOp_PIck(picked_stars_file=a_psfs)
self._insert_processing_step(commands, output_processor=processor)
self.PIck_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def PSf(self, photometry='i.ap', psf_stars='i.lst', psf_file='i.psf'):
# type: ([str,sl.StarList], [str,sl.StarList], str) -> DpOp_PSf
"""
Runs (or adds to execution queue in batch mode) daophot PHOTOMETRY command.
:param str or sl.StarList photometry: input magnitudes file or Starlist, e.g. from aperture photometry by :func:`PHotometry`.
:param str or sl.StarList psf_stars: input list of PSF stars, default: i.coo
:param str psf_file: output PSF file, default: i.psf
:return: results object also accessible as `Daophot.PSf_result` property
:rtype: DpOp_PSf
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
l_phot, a_phot = self._prepare_input_file(photometry)
l_psfs, a_psfs = self._prepare_input_file(psf_stars)
l_psf , a_psf = self._prepare_output_file(psf_file)
l_nei , a_nei = self._prepare_output_file(os.path.splitext(l_psf)[0]+'.nei')
l_err , a_err = self._prepare_output_file('i.err')
commands = 'PSF\n{}\n{}\n{}\n'.format(
l_phot,
l_psfs,
l_psf
)
processor = DpOp_PSf(psf_file=a_psf, nei_file=a_nei, err_file=a_err)
self._insert_processing_step(commands, output_processor=processor)
self.PSf_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def SOrt(self, file, by, decreasing=None):
"""
Adds daophot SORT command to execution stack. NOT IMPLEMENTED sorry
Use sorting capabilities of `StarList` and `StarList.renumber()`
:param str file: fname.COO_FILE etc... any fname.*_FILE to sort
:param by: 1-based column number, negative for descending order - daophot standard, or
one of 'id', 'x', 'y', 'mag'
:param bool decreasing: in not None, forces sort order
:return: results object, also accessible as `Daophot.SOrt_result` property
:rtype: DpOp_Sort
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
if isinstance(by, str):
by = by.lower()
if by == 'id': by = 1
elif by == 'x': by = 2
elif by == 'y': by = 3
elif by == 'mag': by = 4
else:
raise Daophot.RunnerValueError('parameter by, if string must be either: "id", "x", "y" or "mag"', self)
if decreasing is not None:
by = -abs(by) if decreasing else abs(by)
raise NotImplementedError("SORT command not implemented")
if not self.batch_mode:
self.run()
[docs] def SUbstar(self, subtract, leave_in=None, subtracted_image='is.fits', psf_file='i.psf'):
# type: (str, str, str, str) -> DpOp_SUbstar
"""
Adds daophot SUBSTAR command to execution stack.
:param subtract: relative to work dir pathname of stars to subtract file
:param leave_in: relative to work dir pathname of stars to be kept file (default: None)
:param psf_file: relative to work dir pathname of file with PSF (default i.psf)
:param subtracted_image: relative to work dir pathname of output fits file (default is.fits)
:return: results object, also accessible as `Daophot.SUbstar_result` property
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
l_out, a_out = self._prepare_output_file(subtracted_image)
l_sub, a_sub = self._prepare_input_file(subtract)
l_psf, a_psf = self._prepare_input_file(psf_file)
if leave_in is not None:
l_lve, a_lve = self._prepare_input_file(leave_in)
commands = 'SUB\n{}\n{}\ny\n{}\n{}\n'.format(
l_psf,
l_sub,
l_lve,
l_out
)
else:
commands = 'SUB\n{}\n{}\nn\n{}\n'.format(
l_psf,
l_sub,
l_out
)
processor = DpOp_SUbstar(subtracted_image_file=a_out)
self._insert_processing_step(commands, output_processor=processor)
self.SUbstar_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def GRoup(self, photometry='i.ap', psf_file='i.psf', critical_overlap=0.1, groups_file='i.grp'):
# type: ([str,sl.StarList], str, float, str) -> DpOp_GRoup
"""
Runs (or adds to execution queue in batch mode) daophot GROUP command to execution stack.
:param str,StarList photometry: stars to be grouped
:param str psf_file: file with PSF
:param float critical_overlap: relative to work dir pathname of file with PSF
:param str groups_file: output gouped stars file
:return: results object, also accessible as `GRoup_result` property
"""
self._get_ready_for_commands() # wait for completion before changes in working dir
l_pho, a_pho = self._prepare_input_file(photometry)
l_psf, a_psf = self._prepare_input_file(psf_file)
l_grp, a_grp = self._prepare_output_file(groups_file)
commands = 'GROUP\n{}\n{}\n{}\n{}\n'.format(
l_pho,
l_psf,
critical_overlap,
l_grp
)
processor = DpOp_GRoup(groups_file=a_grp)
self._insert_processing_step(commands, output_processor=processor)
self.GRoup_result = processor
if not self.batch_mode:
self.run()
return processor
[docs] def NEda(self, photoopt=None, IS=0, OS=0, apertures=None,
psf_file='i.psf', psf_photometry='i.als', stars_id='i.als', neda_photometry_file='i.nap'):
# type: ([str], float, float, [list], [str], [str,sl.StarList], [str,sl.StarList], [str]) -> DpOp_NEda
"""
Runs (or adds to execution queue in batch mode) daophot NEDA command.
Performing aperture photometry with neighbours SPF profiles subtraction.
Either :param photoopt or :param photo_is, :param OS and :param photo_ap have to be set.
:param [str] photoopt: photo.opt file to be used, default: none
(provide :param photo_is, :param OS and :param photo_ap)
:param float IS: inner sky radius, overwrites :param photoopt file value IS
:param float OS: outer sky radius, overwrites :param photoopt file value OS
:param [list] apertures: apertures radius, up to 12, overwrites `photoopt` file values A1, A2, ...
:param str psf_file: file with PSF for profile subtraction
:param [str, sl.StarList] psf_photometry: stars with PSF photometry for profile subtraction, default: i.als
:param [str, sl.StarList] stars_id: input list of stars to be measured, default: i.als
:param str neda_photometry_file: output neda aperture photometry file
:return: results object, also accessible as `NEda_result` property
:rtype: DpOp_NEda
"""
apertures = self._check_apertures(IS, OS, apertures, photoopt)
self._get_ready_for_commands() # wait for completion before changes in working dir
l_popt, a_popt = self._prepare_input_file(photoopt)
l_psf, a_psf = self._prepare_input_file(psf_file)
l_phot, a_phot = self._prepare_input_file(psf_photometry)
l_star, a_star = self._prepare_input_file(stars_id, astwro.starlist.DAO.XY_FILE)
l_neda, a_neda = self._prepare_output_file(neda_photometry_file)
commands = 'NEDA\n{}\n'.format(l_popt)
if IS != 0:
commands += 'IS={}\n'.format(IS)
if OS != 0:
commands += 'OS={}\n'.format(OS)
for i, ap in enumerate(apertures):
commands += 'A{:X}={}\n'.format(i+1, ap)
commands += '\n{}\n{}\n{}\n{}\n'.format(
l_psf,
l_phot,
l_star,
l_neda
)
processor = DpOp_NEda(neda_file=a_neda)
self._insert_processing_step(commands, output_processor=processor)
self.NEda_result = processor
if not self.batch_mode:
self.run()
return processor
def _check_apertures(self, IS, OS, apertures, photoopt):
if photoopt is None and (apertures is None or IS == 0 or OS == 0):
raise Daophot.RunnerValueError('Apertures and IS and OS must be provided, explicitly or as photoopt file',
self)
if apertures is None:
apertures = []
elif len(apertures) > 12:
raise Daophot.RunnerValueError('apertures apertures list can contain maximum 12 elements', self)
return apertures
def _process_starlist(self, s, **kwargs):
if kwargs['add_psf_errors']:
import pandas as pd
err = self.PSf_result.errors
# idx = [i for i,_ in err]
# val = [v for _,v in err]
# col = pd.Series(val, index=idx)
new = pd.concat([s, err.loc[:,['psf_err', 'flags']]], axis=1, join_axes=[s.index])
new.import_metadata(s)
s = new
return s