/usr/lib/python2.7/dist-packages/ginga/web/pgw/ipg.py is in python-ginga 2.6.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 | #! /usr/bin/env python
#
# ipg.py -- Module for simple FITS viewer in an HTML5 canvas web browser.
#
# This is open-source software licensed under a BSD license.
# Please see the file LICENSE.txt for details.
#
"""
This example illustrates using a Ginga widget in a web browser, All the
rendering is done on the server side and the browser only acts as a display
front end. Using this you could create an analysis type environment on a
server and view it via a browser.
See example usage with an ipython notebook at:
https://gist.github.com/ejeschke/6067409
You will need a reasonably modern web browser with HTML5 canvas support.
Tested with Chromium 41.0.2272.76, Firefox 37.0.2, Safari 7.1.6
"""
from __future__ import print_function
import sys, os
import math
import logging
import threading
import tornado.web
import tornado.ioloop
from ginga import AstroImage, colors
from ginga.canvas.CanvasObject import get_canvas_types
from ginga.misc import log, Task
from ginga.util import catalog, iohelper
from ginga.Bindings import ImageViewBindings
from ginga.misc.Settings import SettingGroup
from ginga.util.paths import ginga_home
from ginga.web.pgw import templates, js, PgHelp, Widgets, Viewers
class EnhancedCanvasView(Viewers.CanvasView):
def embed(self, width=600, height=650):
"""
Embed a viewer into a Jupyter notebook.
"""
from IPython.display import IFrame
return IFrame(self.url, width, height)
def open(self, new=1):
"""
Open this viewer in a new browser window or tab.
(requires `webbrowser` module)
"""
import webbrowser
webbrowser.open(self.url, new=new)
def show(self):
"""
Capture the window of a viewer.
"""
# force any delayed redraws
# TODO: this really needs to be addressed in get_rgb_image_as_bytes()
# of the various superclasses, as it affects other backends as well
self.redraw_now()
from IPython.display import Image
return Image(data=bytes(self.get_rgb_image_as_bytes(format='png')),
format='png', embed=True)
def load_fits(self, filepath):
"""
Load a FITS file into the viewer.
"""
image = AstroImage.AstroImage(logger=self.logger)
image.load_file(filepath)
self.set_image(image)
load = load_fits
def load_hdu(self, hdu):
"""
Load an HDU into the viewer.
"""
image = AstroImage.AstroImage(logger=self.logger)
image.load_hdu(hdu)
self.set_image(image)
def load_data(self, data_np):
"""
Load raw numpy data into the viewer.
"""
image = AstroImage.AstroImage(logger=self.logger)
image.set_data(data_np)
self.set_image(image)
def add_canvas(self, tag=None):
# add a canvas to the view
my_canvas = self.get_canvas()
DrawingCanvas = my_canvas.get_draw_class('drawingcanvas')
canvas = DrawingCanvas()
# enable drawing on the canvas
canvas.enable_draw(True)
canvas.enable_edit(True)
canvas.set_drawtype(None)
canvas.ui_setActive(True)
canvas.set_surface(self)
canvas.register_for_cursor_drawing(self)
# add the canvas to the view.
my_canvas.add(canvas, tag=tag)
return canvas
class ImageViewer(object):
def __init__(self, logger, window, viewer_class=None, width=512, height=512):
if viewer_class is None:
viewer_class = EnhancedCanvasView
self.logger = logger
self.url = window.url
self.dc = get_canvas_types()
self.pixel_base = 1.0
self.top = window
self.top.add_callback('close', self.closed)
vbox = Widgets.VBox()
vbox.set_border_width(2)
vbox.set_spacing(1)
# load binding preferences if available
cfgfile = os.path.join(ginga_home, "ipg_bindings.cfg")
bindprefs = SettingGroup(name='bindings', logger=logger,
preffile=cfgfile)
bindprefs.load(onError='silent')
bd = ImageViewBindings(logger, settings=bindprefs)
fi = viewer_class(logger, bindings=bd)
fi.url = self.url
fi.enable_autocuts('on')
fi.set_autocut_params('zscale')
fi.set_zoom_algorithm('rate')
fi.set_zoomrate(1.1)
fi.enable_autozoom('on')
#fi.set_callback('drag-drop', self.drop_file)
fi.set_callback('none-move', self.motion)
fi.set_bg(0.2, 0.2, 0.2)
fi.ui_setActive(True)
self.fitsimage = fi
fi.ipg_parent = self
bd = fi.get_bindings()
bd.enable_all(True)
# canvas that we will draw on
canvas = self.dc.DrawingCanvas()
canvas.set_surface(fi)
self.canvas = canvas
# add canvas to view
private_canvas = fi.get_canvas()
private_canvas.add(canvas)
canvas.ui_setActive(True)
fi.set_canvas(canvas)
fi.set_desired_size(width, height)
# force allocation of a surface--may be resized later
fi.configure_surface(width, height)
# add little mode indicator that shows modal states in
# the corner
fi.show_mode_indicator(True)
w = Viewers.GingaViewerWidget(viewer=fi)
vbox.add_widget(w, stretch=1)
self.readout = Widgets.Label("")
vbox.add_widget(self.readout, stretch=0)
hbox = Widgets.HBox()
hbox.add_widget(Widgets.Label('Zoom sensitivity: '))
slider = Widgets.Slider(orientation='horizontal', dtype=float)
slider.add_callback('value-changed',
lambda w, val: self.adjust_scrolling_accel_cb(val))
slider.set_limits(1.0, 9.5, 0.1)
val = 4.0
slider.set_value(val)
self.adjust_scrolling_accel_cb(val)
hbox.add_widget(slider, stretch=1)
hbox.add_widget(Widgets.Label(''), stretch=1)
vbox.add_widget(hbox, stretch=0)
# need to put this in an hbox with an expanding label or the
# browser wants to resize the canvas, distorting it
hbox = Widgets.HBox()
hbox.add_widget(vbox, stretch=0)
hbox.add_widget(Widgets.Label(''), stretch=1)
self.top.set_widget(hbox)
def load_file(self, filepath):
image = AstroImage.AstroImage(logger=self.logger)
image.load_file(filepath)
self.fitsimage.set_image(image)
self.top.set_title(filepath)
def drop_file(self, fitsimage, paths):
fileName = paths[0]
self.load_file(fileName)
def motion(self, viewer, button, data_x, data_y):
# Get the value under the data coordinates
try:
#value = viewer.get_data(data_x, data_y)
# We report the value across the pixel, even though the coords
# change halfway across the pixel
value = viewer.get_data(int(data_x+0.5), int(data_y+0.5))
except Exception:
value = None
pb = self.pixel_base
fits_x, fits_y = data_x + pb, data_y + pb
# Calculate WCS RA
try:
# NOTE: image function operates on DATA space coords
image = viewer.get_image()
if image is None:
# No image loaded
return
ra_txt, dec_txt = image.pixtoradec(fits_x, fits_y,
format='str', coords='fits')
except Exception as e:
self.logger.warning("Bad coordinate conversion: %s" % (
str(e)))
ra_txt = 'BAD WCS'
dec_txt = 'BAD WCS'
text = "RA: %s DEC: %s X: %.2f Y: %.2f Value: %s" % (
ra_txt, dec_txt, fits_x, fits_y, value)
self.readout.set_text(text)
def set_readout_text(self, text):
self.readout.set_text(text)
def adjust_scrolling_accel_cb(self, val):
def f(x):
return (math.log10(x) / (10 - x) * x/12) + 1.0001
val2 = f(val)
self.logger.debug("slider value is %f, setting will be %f" % (val, val2))
# save scale
scale_x, scale_y = self.fitsimage.get_scale_xy()
self.fitsimage.set_zoomrate(val2)
# restore scale
self.fitsimage.scale_to(scale_x, scale_y)
return True
def closed(self, w):
self.logger.info("Top window closed.")
self.top = None
sys.exit()
class FileHandler(tornado.web.RequestHandler):
def initialize(self, name, factory):
self.name = name
self.viewer_factory = factory
self.logger = factory.logger
self.logger.info("filehandler initialize")
def get(self):
self.logger.info("filehandler get")
# Collect arguments
wid = self.get_argument('id', None)
# Get window with this id
window = self.app.get_window(wid)
output = window.render()
self.write(output)
class ViewerFactory(object):
def __init__(self, logger, basedir, app, thread_pool):
"""
Constructor parameters:
`logger` : a logging-module compatible logger object
`basedir`: directory to which paths requested on the viewer
are considered relative to.
"""
self.logger = logger
self.basedir = basedir
self.app = app
self.thread_pool = thread_pool
# dict of viewers
self.viewers = {}
def get_basedir(self):
return self.basedir
def get_threadpool(self):
return self.thread_pool
def get_viewer(self, v_id, viewer_class=None, width=512, height=512,
force_new=False):
"""
Get an existing viewer by viewer id. If the viewer does not yet
exist, make a new one.
"""
if not force_new:
try:
return self.viewers[v_id]
except KeyError:
pass
# create top level window
window = self.app.make_window("Viewer %s" % v_id, wid=v_id)
# our own viewer object, customized with methods (see above)
viewer = ImageViewer(self.logger, window,
viewer_class=viewer_class, width=width, height=height)
#viewer.url = window.url
self.viewers[v_id] = viewer
return viewer
def delete_viewer(self, v_id):
del self.viewers[v_id]
def delete_all_viewers(self):
self.viewers = {}
class WebServer(object):
def __init__(self, app, thread_pool, factory,
host='localhost', port=9909, ev_quit=None):
self.host = host
self.port = port
self.app = app
self.logger = app.logger
self.thread_pool = thread_pool
self.factory = factory
if ev_quit is None:
ev_quit = threading.Event()
self.ev_quit = ev_quit
self.server = None
self.http_server = None
def start(self, use_thread=True, no_ioloop=False):
self.thread_pool.startall()
js_path = os.path.dirname(js.__file__)
self.server = tornado.web.Application([
(r"/js/(.*\.js)", tornado.web.StaticFileHandler,
{"path": js_path}),
(r"/viewer", FileHandler,
dict(name='Ginga', factory=self.factory)),
(r"/app", PgHelp.WindowHandler,
dict(name='Application', url='/app', app=self.app)),
(r"/app/socket", PgHelp.ApplicationHandler,
dict(name='Ginga', app=self.app)),
## ("/viewer/socket", ViewerWidget,
## dict(name='Ginga', factory=self.factory)),
],
factory=self.factory, logger=self.logger)
self.http_server = self.server.listen(self.port, self.host)
if no_ioloop:
self.t_ioloop = None
else:
self.t_ioloop = tornado.ioloop.IOLoop.instance()
if use_thread:
task = Task.FuncTask2(self.t_ioloop.start)
self.thread_pool.addTask(task)
else:
self.t_ioloop.start()
def stop(self):
# how to stop tornado server?
if self.t_ioloop is not None:
self.t_ioloop.stop()
self.thread_pool.stopall()
self.ev_quit.set()
# stop and dereference the tornado server
if self.http_server is not None:
self.http_server.stop()
self.http_server = None
self.server = None
def get_viewer(self, v_id, viewer_class=None, width=512, height=512,
force_new=False):
from IPython.display import display, HTML
v = self.factory.get_viewer(v_id, viewer_class=viewer_class,
width=width, height=height,
force_new=force_new)
url = v.top.url
viewer = v.fitsimage
viewer.url = url
#display(HTML('<a href="%s">link to viewer</a>' % url))
return viewer
def make_server(logger=None, basedir='.', numthreads=5,
host='localhost', port=9909, use_opencv=False):
if logger is None:
logger = log.get_logger("ipg", null=True)
ev_quit = threading.Event()
if use_opencv:
from ginga import trcalc
try:
trcalc.use('opencv')
except Exception as e:
logger.warning("Error using opencv: %s" % str(e))
thread_pool = Task.ThreadPool(numthreads, logger,
ev_quit=ev_quit)
base_url = "http://%s:%d/app" % (host, port)
app = Widgets.Application(logger=logger, base_url=base_url,
host=host, port=port)
factory = ViewerFactory(logger, basedir, app, thread_pool)
server = WebServer(app, thread_pool, factory,
host=host, port=port)
return server
def main(options, args):
logger = log.get_logger("ipg", options=options)
server = make_server(logger=logger, basedir=options.basedir,
numthreads=options.numthreads, host=options.host,
port=options.port, use_opencv=options.use_opencv)
try:
server.start(use_thread=False)
except KeyboardInterrupt:
logger.info("Interrupted!")
server.stop()
logger.info("Server terminating...")
if __name__ == "__main__":
# Parse command line options with nifty optparse module
from optparse import OptionParser
usage = "usage: %prog [options] cmd [args]"
optprs = OptionParser(usage=usage, version=('%%prog'))
optprs.add_option("-d", "--basedir", dest="basedir", metavar="DIR",
default=".",
help="Directory which is at the base of file open requests")
optprs.add_option("--debug", dest="debug", default=False, action="store_true",
help="Enter the pdb debugger on main()")
optprs.add_option("--host", dest="host", metavar="HOST",
default="localhost",
help="HOST used to decide which interfaces to listen on")
optprs.add_option("--log", dest="logfile", metavar="FILE",
help="Write logging output to FILE")
optprs.add_option("--loglevel", dest="loglevel", metavar="LEVEL",
type='int', default=logging.INFO,
help="Set logging level to LEVEL")
optprs.add_option("--numthreads", dest="numthreads", type="int",
default=5, metavar="NUM",
help="Start NUM threads in thread pool")
optprs.add_option("--stderr", dest="logstderr", default=False,
action="store_true",
help="Copy logging also to stderr")
optprs.add_option("--opencv", dest="use_opencv", default=False,
action="store_true",
help="Use OpenCv acceleration")
optprs.add_option("-p", "--port", dest="port",
type='int', default=9909, metavar="PORT",
help="Default PORT to use for the web socket")
optprs.add_option("--profile", dest="profile", action="store_true",
default=False,
help="Run the profiler on main()")
(options, args) = optprs.parse_args(sys.argv[1:])
# Are we debugging this?
if options.debug:
import pdb
pdb.run('main(options, args)')
# Are we profiling this?
elif options.profile:
import profile
print(("%s profile:" % sys.argv[0]))
profile.run('main(options, args)')
else:
main(options, args)
# END
|