#!/usr/bin/python import logging import logging.handlers import astropy.io.fits as pyfits h = logging.handlers.TimedRotatingFileHandler('../data/Obs.log', when='midnight', interval=1, backupCount=0, encoding=None, delay=False, utc=False) h.setFormatter(logging.Formatter(fmt='%(asctime)s %(levelname)s %(message)s')) logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO) logging.root.addHandler(h) try: import gtk gtk.set_interactive(False) except ImportError as AttributeError: pass import numpy as np import fitsbuilder import struct from stardice import configfiles import scheduler reload(scheduler) import threading import stardice.qla import ephem import datetime import stardice.visu from glob import glob import sys import time def lsdir(text): return glob(text+'*') class Functionnality(object): def __init__(self, f): self.f = f self.help = f.__doc__.splitlines()[0].strip() def __call__(self): print(self.help) print(self.f()) def complist(self, text): return [] class CriticalFunctionnality(Functionnality): def __call__(self): print(self.help) confirm = input('Confirm (y/n)?') if confirm == 'y': print(self.f()) class ParametricFunctionnality(Functionnality): def __init__(self, f, type): self.f = f self.help = f.__doc__.splitlines()[0] self.type = type self._values = [] def __call__(self): print(self.help) val = self.type(input('Value ?')) print(self.f(val)) def complist(self, text): return [key for key in self._values if key.startswith(text)] class KeyboardControls(object): def __init__(self): self.functions = {} self.groups_list = ['Camera', 'Focus', 'Mount', 'Led', 'Analysis', 'Interface'] self.groups = dict([(k, []) for k in self.groups_list]) #self.keyboard = Getch() self.register_function(self.print_help, '?') self.register_function(self.quit, 'q') self.status = 'IDLE' self.exptime = 0.1 self.register_function(self.set_exptime, 'e', type=float, group='Camera') self.filter = 'Bi' self.register_function(self.set_filter, 'b', type=str, group='Camera') self.last_image = '' self.progfile = '' self.target = '' def print_help(self): ''' Print this help message ''' s = [] for g in self.groups_list: s.append(g+':') s.append('-------------------') for key in self.groups[g]: s.append('%s: %s' % (key, self.functions[key].help)) s.append('') return '\n'.join(s) def complete(self, text, state): possible = self.complist(text) if state >= len(possible): return None else: return possible[state] def _complist(self, text): return [key for g in self.groups_list for key in self.groups[g] if key.startswith(text)] def set_exptime(self, exptime): ''' Set exptime in seconds [>0.1] ''' self.exptime = exptime return 'exposure time set to %g s' % self.exptime def adjust(self, offset): ''' Adjust focal plane focus by relative offset in steps (positive toward infinity) ''' return s.focus.adjust(self.filter, self.target, offset) def plus(self): ''' Adjust focal plane offset toward infinity by 10 steps ''' return s.focus.plus(self.filter, self.target) def minus(self): ''' Adjust focal plane offset toward nearby by 10 steps ''' return s.focus.minus(self.filter, self.target) def goto_target(self, target): ''' go to a target given by name: ''' self.target = target return s.mount.goto_target(self.target) #- SZF begin: Utilities to change and manually set the configuration def set_target(self, target): ''' set target name ''' self.target = target return target def set_mount_delay(self, delay): ''' set mount delay ''' s.mount.delay = delay return delay #- SZF end def set_home(self, hadec): ''' set the mount home position to ha,dec in degrees set_home_altaz actually takes hadec ''' ha, dec = [float(v) for v in hadec.split(',')] return s.mount.set_home_altaz(ha, dec) def set_altaz(self, altaz): ''' Set zeros of the mount so that current pointing match given alt and az in degrees ''' alt, az = [float(v) for v in altaz.split(',')] return s.mount.set_altaz(alt, az) def set_mount_delta_ha_dec(self, delta_ha_delta_dec): ''' Offset the zero of mount ha axis by the amount given in degree comma separated: delta_ha,delta_dec''' delta_ha, delta_dec = [float(v) for v in delta_ha_delta_dec.split(',')] s.mount.delta_ha = delta_ha s.mount.delta_dec = delta_dec return 'offset adjusted' def goto_radec(self, radec): ''' Go to a target given by coordinate 'ra_deg,dec_deg' ''' ra, dec = [float(v) for v in radec.split(',')] return s.mount.goto_radec(ra, dec) def goto_hadec(self, hadec): ''' Go to a target given by coordinate 'ha_deg,dec_deg' ''' ha, dec = [float(v) for v in hadec.split(',')] s.mount.sync_obs() sideral_time = np.degrees(s.mount.observer.sidereal_time()) ra = sideral_time - ha return s.mount.goto_radec(ra, dec) def goto_altaz(self, altaz): ''' Go to a target given by coordinate 'alt_deg,az_deg' in decimal degrees ''' alt, az = [v for v in altaz.split(',')] return s.mount.goto_altaz(alt, az) def set_filter(self, filter): ''' Set filter in the filter wheel [EMPTY, Bi, Vi, R, I] ''' #- GRISM was removed 2018-07-11 # ''' Set filter in the filter wheel [GRISM, Bi, Vi, R, I] # ''' if filter in s.camera.band_pos: self.filter = filter return 'Filter will be set to %s before next exposure' % filter else: return 'Unknown filter %s. Should be one of %s' % (filter, str(s.camera.band_pos)) def single_image(self): '''Take a single image at the current position and with current settings ''' self.exposure = {'TARGET': self.target, 'RADEG': 'IDLE', 'DECDEG': 'IDLE', 'DELTARA': '0', 'DELTADEC': '0', 'TRACKING': 'False' if (self.target == 'LEDTEST') or (self.target == 'LEDHEAD') or (self.target == 'ZENITH') or (self.target == 'DARK') else 'True', 'EXPTIME': self.exptime, 'SHUTTER': 'Open', 'FOCOFFSET': 'ADJUST', 'FILTER': self.filter, 'LED': 0, 'ILED':0, 'HEADXDEG':0, 'HEADYDEG':0, 'DOME': 'IDLE', } self.status = 'IMG' return 'Will take a single image:' + str(self.exposure) def dark(self): '''Take a single dark with current settings ''' self.exposure = {'TARGET': 'DARK', 'RADEG': 'IDLE', 'DECDEG': 'IDLE', 'DELTARA': '0', 'DELTADEC': '0', 'TRACKING': 'True', 'EXPTIME': self.exptime, 'SHUTTER': 'Closed', 'FOCOFFSET': 'IDLE', 'FILTER': self.filter, 'LED': 0, 'ILED':0, 'HEADXDEG':0, 'HEADYDEG':0, 'DOME': 'IDLE', } self.status = 'IMG' return 'Will take a single dark:' + str(self.exposure) def video(self): '''Swich on the video mode ''' self.status = 'VIDEO' return 'Switching to video mode' def idle(self): '''Swich off the video mode ''' self.status = 'IDLE' return 'Switching off video mode' def analyse_last_image(self): '''Run the analysis code on last registered frame''' if not self.last_image: print('No image registered (try "i" beforehand)') else: s.analyse.analyse_image(self.last_image) def show_match(self): ''' Draw a circle around calspec matches''' import stardice.qla for i in (s.analyse.index != -1).nonzero()[0]: x, y = s.analyse.C2.secat[['x', 'y']][i] imager.d.set('regions', 'image;global color=red;text %f %f {%s};' % (x, y, stardice.qla.calspecs['CALSPEC_NAME'][s.analyse.index[i]])) def time(self): '''Print current UTC and sidereal time''' s.mount.sync_obs() st = s.mount.observer.sidereal_time() ret = 'date: %s (UTC)\n' % (str(s.mount.observer.date)) ret += 'sidereal time: %s (%f deg)' % (str(st), np.degrees(st)) return ret def show_star_list(self): ''' Show regions corresponding to know stars in the area''' cat, flux = s.analyse.star_list() field_stars = np.isfinite(cat['VT']) calspec_stars = ~field_stars sig, b = 0.5, 90 rad = sig * np.sqrt(2 * np.log(flux*self.exptime/(b * sig * np.sqrt(2*np.pi)))) field_stars &= np.isfinite(rad) print(rad[field_stars].min()) imager.d.set('regions', "fk5;global color=green;" + ';'.join(['circle %f %f %f"' % (star['ra'], star['dec'], _rad) for star, _rad in zip(cat[field_stars], rad[field_stars])])) if calspec_stars.any(): print('%d calpsec star in field' % calspec_stars.sum()) imager.d.set('regions', "fk5;global color=red;" + ';'.join(['circle %f %f %f"' % (star['ra'], star['dec'], _rad) for star, _rad in zip(cat[calspec_stars], rad[calspec_stars])])) return '' def switch_webcam_leds_on(self): ''' Switch the webcams IR LEDs ON''' s.webcam.webcams_IR_leds(on=True) def switch_webcam_leds_off(self): ''' Switch the webcams IR LEDs ON''' s.webcam.webcams_IR_leds(on=False) def execute_prog(self, progfile, line=0): ''' Launch the execution of an observation program''' self.progfile = progfile self.startline = line self.status='PROG' imager.d.set('preserve regions no') print("Starting program %s at line %d" % (self.progfile, self.startline)) for im, exposure in s.execute(self.progfile, startline=self.startline): with pyfits.open(im) as fits: imager.d.set_pyfits(fits) logging.info('Image saved in file %s' % im) s.accepted = s.analyse.post_proc(exposure, im, imager) imager.d.set('preserve regions yes') #print '\a' print("Program completed.") self.status='IDLE' def restart_prog(self, line): ''' Restart the observation program at the given line''' if not self.progfile: return 'No program file defined. Use l instead' self.execute_prog(self.progfile, line) def quit(self): ''' Quit ''' if not s.mount.is_parked() == '1': confirm = input('Mount not parked, exit anyway ? (y/n)') if confirm != 'y': return 'Not quitting' self.status='STOP' return 'Quit' def register_function(self, function, key, type=None, critical=False, group='Interface'): assert key not in self.functions, 'A function is already registered under key %s' % key if critical: self.functions[key] = CriticalFunctionnality(function) elif type is not None: self.functions[key] = ParametricFunctionnality(function, type) else: self.functions[key] = Functionnality(function) self.groups[group].append(key) def keyboard_loop(self): while self.status != 'STOP': try: #c = self.keyboard() self.complist = self._complist c = input("> ") if c: self.complist = self.functions[c].complist self.functions[c]() except KeyboardInterrupt as e: print("stopping everything, going back to idle mode (q to quit)") try: s.mount.emergency_stop() except Exception as e: print(repr(e)) s.abort() self.status = 'IDLE' except Exception as e: print(repr(e)) #self.stop = True class Imager(object): def __init__(self, scheduler, keyboard): self.restart_ds9() self.keyboard = keyboard self.s = scheduler def restart_ds9(self): """ Restarts ds9 if closed while rvideo runs """ import pyds9 self.d = pyds9.DS9() self.d.set('preserve regions yes') return "ds9 restarted" def clear(self): ''' Suppress all DS9 regions ''' self.d.set('regions delete all') return '' def select_center(self): self.d.set('crosshair %d %d physical' % self.s.camera.field_center) input('Put the crosshair on target position. Hit return when ready.') ra, dec = [float(v) for v in self.d.get('crosshair wcs').split()] # to get offset, subtract self.s.mount._last_ra et dec return ra, dec def recenter(self): ''' Recenter to a position on the image ''' self.keyboard.status = 'IDLE' ra, dec = self.select_center() rac, decc = stardice.qla.field_center(self.w) delta_ra = (ra - rac) * 3600 delta_dec = (dec - decc) * 3600 self.s.mount.relative_move_radec(delta_ra, delta_dec) self.d.set('mode none') #self.keyboard.status = 'VIDEO' def image_loop(self): frame = [0, 0, 0, 0] mode = 0 while self.keyboard.status != 'STOP': try: if self.keyboard.status == 'VIDEO': if self.s.camera.get_filter() != self.keyboard.filter: self.s.camera.set_filter(self.keyboard.filter) print('filter set to ' + self.keyboard.filter) image = self.s.camera.take_patch(self.keyboard.exptime, frame, mode) self.d.set_np2arr(image['pixels'].astype(float)) self.w = self.s.mount.get_wcs() self.d.set('wcs append', self.w.to_header_string()) elif self.keyboard.status == 'IMG': if self.s.camera.get_filter() != self.keyboard.filter: self.s.camera.set_filter(self.keyboard.filter) print('filter set to ' + self.keyboard.filter) self.s.aborting = False im = self.s.shoot(self.keyboard.exposure) print("Exposure stored in %s" % im) self.keyboard.last_image = im with pyfits.open(im) as fits: self.d.set_pyfits(fits) self.w = self.s.mount.get_wcs() self.d.set('wcs append', self.w.to_header_string()) self.keyboard.status = 'IDLE' elif self.keyboard.status == 'PROG': while self.keyboard.status == 'PROG': time.sleep(1) else: time.sleep(0.1) except Exception as e: #import traceback #print '\n'.join(str(sys.exc_traceback)) print(repr(e)+'\n', end=' ') print('Going back to idle mode') self.keyboard.status = 'IDLE' #s = traceback.format_exc() #sys.stdout.write(s) #return #print 'Exiting image loop' if __name__ == '__main__': import argparse parser = argparse.ArgumentParser( description='Take a video through xmlrpc') parser.add_argument( '-t', '--exposure-time', default=1.0, type=float, help='Exposure time in second (default to 0.1s, shorter possible)') parser.add_argument( '--config', default='scheduler.cfg', help='Use a different config file') parser.add_argument( '-f', '--filter', default='Bi', help='Filter name !') parser.add_argument( '-b', '--binning', default=False, action='store_true', help='Activate 9x9 binning') parser.add_argument( '-F', '--frame', default=[0, 0, 0, 0], nargs=4, type=int, help='Activate framing') parser.add_argument( '-n', '--repeat', default=1, type=int, help='Take several exposure') args = parser.parse_args() cfg = configfiles.Config(args.config) s = scheduler.Scheduler(cfg) keyboard = KeyboardControls() imager = Imager(s, keyboard) # Interface mapping keyboard.register_function(keyboard.video, 'video', group='Interface') keyboard.register_function(keyboard.idle, 'idle', group='Interface') keyboard.register_function(imager.restart_ds9, 'restart_ds9', group='Interface') # Focus keyboard mapping keyboard.register_function(keyboard.plus, '+', group='Focus') keyboard.register_function(keyboard.minus, '-', group='Focus') keyboard.register_function(keyboard.adjust, 'f', int, group='Focus') # Mount keyboard mapping for k, f in s.mount._shortcut: keyboard.register_function(f, k, group='Mount') keyboard.register_function(keyboard.goto_target, 'g', str, group='Mount') #- Begin SZF: utilities to change the config manually keyboard.register_function(keyboard.set_target, 'set_target', str, group='Mount') keyboard.register_function(keyboard.set_mount_delay, 'set_mount_delay', float, group='Mount') #- End SZF keyboard.functions['g']._values = list(s.mount.targets.CALSPEC.keys()) keyboard.register_function(keyboard.goto_radec, 'goto_radec', str, group='Mount') keyboard.register_function(keyboard.goto_altaz, 'goto_altaz', str, group='Mount') keyboard.register_function(keyboard.set_altaz, 'set_altaz', str, group='Mount') keyboard.register_function(keyboard.goto_hadec, 'goto_hadec', str, group='Mount') keyboard.register_function(keyboard.set_mount_delta_ha_dec, 'set_mount_delta_ha_dec', str, group='Mount') keyboard.register_function(s.mount.check_sync, 'check_sync', group='Mount') keyboard.register_function(s.mount.relative_move_east, 'go_east', float, group='Mount') keyboard.register_function(s.mount.relative_move_west, 'go_west', float, group='Mount') keyboard.register_function(s.mount.relative_move_north, 'go_north', float, group='Mount') keyboard.register_function(s.mount.relative_move_south, 'go_south', float, group='Mount') keyboard.register_function(s.mount.relative_move_ra, 'delta_ra', float, group='Mount') keyboard.register_function(s.mount.relative_move_dec, 'delta_dec', float, group='Mount') keyboard.register_function(imager.recenter, 'recenter', group='Mount') keyboard.register_function(s.mount_init, 'init_mount', group='Mount') keyboard.register_function(s.mount.park, 'park', group='Mount', critical=True) keyboard.register_function(s.mount.get_home_altaz, 'get_home_altaz', group='Mount') keyboard.register_function(keyboard.set_home, 'set_home', str, group='Mount') # LED keyboard mapping keyboard.register_function(s.led.sample, 'led_sample', int, group='Led') keyboard.register_function(s.led.led_off, 'led_off', group='Led') keyboard.register_function(s.led.led_on, 'led_on', int, group='Led') keyboard.register_function(s.led.led_on_current, 'led_on_current', str, group='Led') # Webcams keyboard mapping keyboard.register_function(keyboard.switch_webcam_leds_on, 'webcam_leds_on', group='Interface') keyboard.register_function(keyboard.switch_webcam_leds_off, 'webcam_leds_off', group='Interface') # High level stuff keyboard.register_function(keyboard.execute_prog, 'l', str, group='Analysis') keyboard.register_function(keyboard.time, 'time', group='Analysis') keyboard.functions['l'].complist = lsdir keyboard.register_function(keyboard.restart_prog, 'restart', int, group='Analysis') keyboard.register_function(keyboard.analyse_last_image, 'a', group='Analysis') keyboard.register_function(s.analyse.target_plots, 'available_targets', group='Analysis') keyboard.register_function(keyboard.show_match, 'm', group='Analysis') keyboard.register_function(keyboard.show_star_list, 'star_list', group='Analysis') keyboard.register_function(imager.clear, 'clear', group='Analysis') keyboard.register_function(keyboard.single_image, 'i', group='Camera') keyboard.register_function(keyboard.dark, 'dark', group='Camera') keyboard.register_function(s.state, 's') print(keyboard.set_filter(args.filter)) print(keyboard.set_exptime(args.exposure_time)) #mode = 9 if args.binning else 0 thr = threading.Thread(target=imager.image_loop) thr.daemon = True # Set up completion and history import os import readline histfile = os.path.join(os.path.expanduser("~"), ".rvideo") try: readline.read_history_file(histfile) # default history len is -1 (infinite), which looks like just fine to me # readline.set_history_length(1000) readline.set_completer(keyboard.complete) readline.parse_and_bind("tab: complete") except IOError: pass import atexit atexit.register(readline.write_history_file, histfile) if s.mount.is_parked == '1': print('Mount is parked, hit "unpark" to unpark.') try: thr.start() keyboard.keyboard_loop() finally: pass #keyboard.quit() #thr.join()