<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> import sys , os <EOL> import logging <EOL> import threading <EOL> import tornado . web <EOL> import tornado . ioloop <EOL> from ginga import AstroImage , colors <EOL> from ginga . canvas . CanvasObject import get_canvas_types <EOL> from ginga . misc import log , Task <EOL> from ginga . util import catalog , iohelper <EOL> from ginga . web . pgw import templates , js , PgHelp , Widgets , Viewers <EOL> class EnhancedCanvasView ( Viewers . CanvasView ) : <EOL> def embed ( self , width = <NUM_LIT> , height = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> from IPython . display import IFrame <EOL> return IFrame ( self . url , width , height ) <EOL> def open ( self , new = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> import webbrowser <EOL> webbrowser . open ( self . url , new = new ) <EOL> def show ( self ) : <EOL> """<STR_LIT>""" <EOL> from IPython . display import Image <EOL> return Image ( data = bytes ( self . get_rgb_image_as_bytes ( format = '<STR_LIT>' ) ) , <EOL> format = '<STR_LIT>' , embed = True ) <EOL> def load_fits ( self , filepath ) : <EOL> """<STR_LIT>""" <EOL> image = AstroImage . AstroImage ( logger = self . logger ) <EOL> image . load_file ( filepath ) <EOL> self . set_image ( image ) <EOL> load = load_fits <EOL> def load_hdu ( self , hdu ) : <EOL> """<STR_LIT>""" <EOL> image = AstroImage . AstroImage ( logger = self . logger ) <EOL> image . load_hdu ( hdu ) <EOL> self . set_image ( image ) <EOL> def load_data ( self , data_np ) : <EOL> """<STR_LIT>""" <EOL> image = AstroImage . AstroImage ( logger = self . logger ) <EOL> image . set_data ( data_np ) <EOL> self . set_image ( image ) <EOL> def add_canvas ( self , tag = None ) : <EOL> my_canvas = self . get_canvas ( ) <EOL> DrawingCanvas = my_canvas . get_draw_class ( '<STR_LIT>' ) <EOL> canvas = DrawingCanvas ( ) <EOL> canvas . enable_draw ( True ) <EOL> canvas . enable_edit ( True ) <EOL> canvas . ui_setActive ( True ) <EOL> canvas . set_surface ( self ) <EOL> canvas . register_for_cursor_drawing ( self ) <EOL> my_canvas . add ( canvas , tag = tag ) <EOL> return canvas <EOL> def resize ( self , width , height ) : <EOL> self . configure_window ( width , height ) <EOL> self . pgcanvas . resize ( width , height ) <EOL> app = self . pgcanvas . get_app ( ) <EOL> app . do_operation ( '<STR_LIT>' , id = self . pgcanvas . id ) <EOL> class ImageViewer ( object ) : <EOL> def __init__ ( self , logger , window , viewer_class = None ) : <EOL> if viewer_class is None : <EOL> viewer_class = EnhancedCanvasView <EOL> self . logger = logger <EOL> self . url = window . url <EOL> self . drawcolors = colors . get_colors ( ) <EOL> self . dc = get_canvas_types ( ) <EOL> self . top = window <EOL> self . top . add_callback ( '<STR_LIT>' , self . closed ) <EOL> vbox = Widgets . VBox ( ) <EOL> vbox . set_border_width ( <NUM_LIT:2> ) <EOL> vbox . set_spacing ( <NUM_LIT:1> ) <EOL> fi = viewer_class ( logger ) <EOL> fi . url = self . url <EOL> fi . enable_autocuts ( '<STR_LIT>' ) <EOL> fi . set_autocut_params ( '<STR_LIT>' ) <EOL> fi . enable_autozoom ( '<STR_LIT>' ) <EOL> fi . set_zoom_algorithm ( '<STR_LIT>' ) <EOL> fi . set_zoomrate ( <NUM_LIT> ) <EOL> fi . show_pan_mark ( True ) <EOL> fi . set_callback ( '<STR_LIT>' , self . drop_file ) <EOL> fi . set_callback ( '<STR_LIT>' , self . motion ) <EOL> fi . set_bg ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> fi . ui_setActive ( True ) <EOL> self . fitsimage = fi <EOL> bd = fi . get_bindings ( ) <EOL> bd . enable_all ( True ) <EOL> settings = bd . get_settings ( ) <EOL> settings . set ( scroll_zoom_direct_scale = True ) <EOL> canvas = self . dc . DrawingCanvas ( ) <EOL> canvas . setSurface ( fi ) <EOL> self . canvas = canvas <EOL> private_canvas = fi . get_canvas ( ) <EOL> private_canvas . add ( canvas ) <EOL> canvas . ui_setActive ( True ) <EOL> fi . set_canvas ( canvas ) <EOL> private_canvas . add ( self . dc . ColorBar ( side = '<STR_LIT>' , offset = <NUM_LIT:10> ) ) <EOL> private_canvas . add ( self . dc . ModeIndicator ( corner = '<STR_LIT>' , fontsize = <NUM_LIT> ) ) <EOL> bm = fi . get_bindmap ( ) <EOL> bm . add_callback ( '<STR_LIT>' , lambda * args : fi . redraw ( whence = <NUM_LIT:3> ) ) <EOL> fi . set_desired_size ( <NUM_LIT> , <NUM_LIT> ) <EOL> w = Viewers . GingaViewerWidget ( viewer = fi ) <EOL> vbox . add_widget ( w , stretch = <NUM_LIT:1> ) <EOL> self . readout = Widgets . Label ( "<STR_LIT>" ) <EOL> vbox . add_widget ( self . readout , stretch = <NUM_LIT:0> ) <EOL> hbox = Widgets . HBox ( ) <EOL> btn3 = Widgets . CheckBox ( "<STR_LIT>" ) <EOL> btn3 . add_callback ( '<STR_LIT>' , lambda w , tf : self . use_trackpad_cb ( tf ) ) <EOL> hbox . add_widget ( btn3 ) <EOL> hbox . add_widget ( Widgets . Label ( '<STR_LIT>' ) , stretch = <NUM_LIT:1> ) <EOL> vbox . add_widget ( hbox , stretch = <NUM_LIT:0> ) <EOL> hbox = Widgets . HBox ( ) <EOL> hbox . add_widget ( vbox , stretch = <NUM_LIT:0> ) <EOL> hbox . add_widget ( Widgets . Label ( '<STR_LIT>' ) , stretch = <NUM_LIT:1> ) <EOL> self . top . set_widget ( hbox ) <EOL> def load_file ( self , filepath ) : <EOL> image = AstroImage . AstroImage ( logger = self . logger ) <EOL> image . load_file ( filepath ) <EOL> self . fitsimage . set_image ( image ) <EOL> self . top . set_title ( filepath ) <EOL> def drop_file ( self , fitsimage , paths ) : <EOL> fileName = paths [ <NUM_LIT:0> ] <EOL> self . load_file ( fileName ) <EOL> def motion ( self , viewer , button , data_x , data_y ) : <EOL> try : <EOL> value = viewer . get_data ( int ( data_x + <NUM_LIT:0.5> ) , int ( data_y + <NUM_LIT:0.5> ) ) <EOL> except Exception : <EOL> value = None <EOL> fits_x , fits_y = data_x + <NUM_LIT:1> , data_y + <NUM_LIT:1> <EOL> try : <EOL> image = viewer . get_image ( ) <EOL> if image is None : <EOL> return <EOL> ra_txt , dec_txt = image . pixtoradec ( fits_x , fits_y , <EOL> format = '<STR_LIT:str>' , coords = '<STR_LIT>' ) <EOL> except Exception as e : <EOL> self . logger . warning ( "<STR_LIT>" % ( <EOL> str ( e ) ) ) <EOL> ra_txt = '<STR_LIT>' <EOL> dec_txt = '<STR_LIT>' <EOL> text = "<STR_LIT>" % ( <EOL> ra_txt , dec_txt , fits_x , fits_y , value ) <EOL> self . readout . set_text ( text ) <EOL> def use_trackpad_cb ( self , state ) : <EOL> settings = self . fitsimage . get_bindings ( ) . get_settings ( ) <EOL> val = <NUM_LIT:1.0> <EOL> if state : <EOL> val = <NUM_LIT:0.1> <EOL> settings . set ( scroll_zoom_acceleration = val ) <EOL> def closed ( self , w ) : <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> self . top = None <EOL> sys . exit ( ) <EOL> class FileHandler ( tornado . web . RequestHandler ) : <EOL> def initialize ( self , name , factory ) : <EOL> self . name = name <EOL> self . viewer_factory = factory <EOL> self . logger = factory . logger <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> def get ( self ) : <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> wid = self . get_argument ( '<STR_LIT:id>' , None ) <EOL> window = self . app . get_window ( wid ) <EOL> output = window . render ( ) <EOL> self . write ( output ) <EOL> class ViewerFactory ( object ) : <EOL> def __init__ ( self , logger , basedir , app , thread_pool ) : <EOL> """<STR_LIT>""" <EOL> self . logger = logger <EOL> self . basedir = basedir <EOL> self . app = app <EOL> self . thread_pool = thread_pool <EOL> self . viewers = { } <EOL> def get_basedir ( self ) : <EOL> return self . basedir <EOL> def get_threadpool ( self ) : <EOL> return self . thread_pool <EOL> def get_viewer ( self , v_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . viewers [ v_id ] <EOL> except KeyError : <EOL> pass <EOL> window = self . app . make_window ( "<STR_LIT>" % v_id , wid = v_id ) <EOL> viewer = ImageViewer ( self . logger , window ) <EOL> self . viewers [ v_id ] = viewer <EOL> return viewer <EOL> class WebServer ( object ) : <EOL> def __init__ ( self , app , thread_pool , factory , <EOL> host = '<STR_LIT:localhost>' , port = <NUM_LIT> , ev_quit = None ) : <EOL> self . host = host <EOL> self . port = port <EOL> self . app = app <EOL> self . logger = app . logger <EOL> self . thread_pool = thread_pool <EOL> self . factory = factory <EOL> if ev_quit is None : <EOL> ev_quit = threading . Event ( ) <EOL> self . ev_quit = ev_quit <EOL> self . server = None <EOL> def start ( self , use_thread = True , no_ioloop = False ) : <EOL> self . thread_pool . startall ( ) <EOL> js_path = os . path . dirname ( js . __file__ ) <EOL> self . server = tornado . web . Application ( [ <EOL> ( r"<STR_LIT>" , tornado . web . StaticFileHandler , <EOL> { "<STR_LIT:path>" : js_path } ) , <EOL> ( r"<STR_LIT>" , FileHandler , <EOL> dict ( name = '<STR_LIT>' , factory = self . factory ) ) , <EOL> ( r"<STR_LIT>" , PgHelp . WindowHandler , <EOL> dict ( name = '<STR_LIT>' , url = '<STR_LIT>' , app = self . app ) ) , <EOL> ( r"<STR_LIT>" , PgHelp . ApplicationHandler , <EOL> dict ( name = '<STR_LIT>' , app = self . app ) ) , <EOL> ] , <EOL> factory = self . factory , logger = self . logger ) <EOL> self . server . listen ( self . port , self . host ) <EOL> if no_ioloop : <EOL> self . t_ioloop = None <EOL> else : <EOL> self . t_ioloop = tornado . ioloop . IOLoop . instance ( ) <EOL> if use_thread : <EOL> task = Task . FuncTask2 ( self . t_ioloop . start ) <EOL> self . thread_pool . addTask ( task ) <EOL> else : <EOL> self . t_ioloop . start ( ) <EOL> def stop ( self ) : <EOL> if not self . t_ioloop is None : <EOL> self . t_ioloop . stop ( ) <EOL> self . thread_pool . stopall ( ) <EOL> self . ev_quit . set ( ) <EOL> def get_viewer ( self , v_id ) : <EOL> from IPython . display import display , HTML <EOL> v = self . factory . get_viewer ( v_id ) <EOL> url = v . top . url <EOL> viewer = v . fitsimage <EOL> viewer . url = url <EOL> return viewer <EOL> def make_server ( logger = None , basedir = '<STR_LIT:.>' , numthreads = <NUM_LIT:5> , <EOL> host = '<STR_LIT:localhost>' , port = <NUM_LIT> , use_opencv = False ) : <EOL> if logger is None : <EOL> logger = log . get_logger ( "<STR_LIT>" , null = True ) <EOL> ev_quit = threading . Event ( ) <EOL> if use_opencv : <EOL> from ginga import trcalc <EOL> try : <EOL> trcalc . use ( '<STR_LIT>' ) <EOL> except Exception as e : <EOL> logger . warning ( "<STR_LIT>" % str ( e ) ) <EOL> thread_pool = Task . ThreadPool ( numthreads , logger , <EOL> ev_quit = ev_quit ) <EOL> base_url = "<STR_LIT>" % ( host , port ) <EOL> app = Widgets . Application ( logger = logger , base_url = base_url , <EOL> host = host , port = port ) <EOL> factory = ViewerFactory ( logger , basedir , app , thread_pool ) <EOL> server = WebServer ( app , thread_pool , factory , <EOL> host = host , port = port ) <EOL> return server <EOL> def main ( options , args ) : <EOL> logger = log . get_logger ( "<STR_LIT>" , options = options ) <EOL> server = make_server ( logger = logger , basedir = options . basedir , <EOL> numthreads = options . numthreads , host = options . host , <EOL> port = options . port , use_opencv = options . use_opencv ) <EOL> try : <EOL> server . start ( use_thread = False ) <EOL> except KeyboardInterrupt : <EOL> logger . info ( "<STR_LIT>" ) <EOL> server . stop ( ) <EOL> logger . info ( "<STR_LIT>" ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> from optparse import OptionParser <EOL> usage = "<STR_LIT>" <EOL> optprs = OptionParser ( usage = usage , version = ( '<STR_LIT>' ) ) <EOL> optprs . add_option ( "<STR_LIT>" , "<STR_LIT>" , dest = "<STR_LIT>" , metavar = "<STR_LIT>" , <EOL> default = "<STR_LIT:.>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , default = False , action = "<STR_LIT:store_true>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT:host>" , metavar = "<STR_LIT>" , <EOL> default = "<STR_LIT:localhost>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , metavar = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , metavar = "<STR_LIT>" , <EOL> type = '<STR_LIT:int>' , default = logging . INFO , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , type = "<STR_LIT:int>" , <EOL> default = <NUM_LIT:5> , metavar = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , default = False , <EOL> action = "<STR_LIT:store_true>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , default = False , <EOL> action = "<STR_LIT:store_true>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , "<STR_LIT>" , dest = "<STR_LIT:port>" , <EOL> type = '<STR_LIT:int>' , default = <NUM_LIT> , metavar = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> optprs . add_option ( "<STR_LIT>" , dest = "<STR_LIT>" , action = "<STR_LIT:store_true>" , <EOL> default = False , <EOL> help = "<STR_LIT>" ) <EOL> ( options , args ) = optprs . parse_args ( sys . argv [ <NUM_LIT:1> : ] ) <EOL> if options . debug : <EOL> import pdb <EOL> pdb . run ( '<STR_LIT>' ) <EOL> elif options . profile : <EOL> import profile <EOL> print ( ( "<STR_LIT>" % sys . argv [ <NUM_LIT:0> ] ) ) <EOL> profile . run ( '<STR_LIT>' ) <EOL> else : <EOL> main ( options , args ) </s>
