/usr/lib/python3/dist-packages/ginga/rv/plugins/RC.py is in python3-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 | #
# RC.py -- Remote Control plugin for Ginga fits viewer
#
# This is open-source software licensed under a BSD license.
# Please see the file LICENSE.txt for details.
#
"""
The RC plugin implements a remote control interface for the Ginga FITS
viewer.
Show example usage:
$ grc help
Show help for a specific ginga method:
$ grc help ginga <method>
Show help for a specific channel method:
$ grc help channel <chname> <method>
Ginga methods can be called like this:
$ grc ginga <method> <arg1> <arg2> ...
Channel methods can be called like this:
$ grc channel <chname> <method> <arg1> <arg2> ...
Calls can be made from a remote host by adding the options
--host=<hostname> --port=9000
(in the plugin GUI be sure to remove the 'localhost' prefix
from the addr, but leave the colon and port)
Examples:
Create a new channel:
$ grc ginga add_channel FOO
Load a file into current channel:
$ grc ginga load_file /home/eric/testdata/SPCAM/SUPA01118797.fits
Load a file into a specific channel:
$ grc ginga load_file /home/eric/testdata/SPCAM/SUPA01118797.fits FOO
Cut levels:
$ grc channel FOO cut_levels 163 1300
Auto cut levels:
$ grc channel FOO auto_levels
Zoom to a specific level:
$ grc -- channel FOO zoom_to -7
Zoom to fit:
$ grc channel FOO zoom_fit
Transform (args are boolean triplet: (flipx flipy swapxy)):
$ grc channel FOO transform 1 0 1
Rotate:
$ grc channel FOO rotate 37.5
Change color map:
$ grc channel FOO set_color_map rainbow3
Change color distribution algorithm:
$ grc channel FOO set_color_algorithm log
Change intensity map:
$ grc channel FOO set_intensity_map neg
"""
import sys
import numpy
import binascii
import bz2
from io import BytesIO
from ginga import GingaPlugin
from ginga import AstroImage
from ginga.gw import Widgets
from ginga.util import grc
from ginga.util.six.moves import map, zip
help_msg = sys.modules[__name__].__doc__
class RC(GingaPlugin.GlobalPlugin):
def __init__(self, fv):
# superclass defines some variables for us, like logger
super(RC, self).__init__(fv)
# What port to listen for requests
self.port = 9000
# If blank, listens on all interfaces
self.host = 'localhost'
self.ev_quit = fv.ev_quit
def build_gui(self, container):
vbox = Widgets.VBox()
fr = Widgets.Frame("Remote Control")
captions = [
("Addr:", 'label', "Addr", 'llabel', 'Restart', 'button'),
("Set Addr:", 'label', "Set Addr", 'entry'),
]
w, b = Widgets.build_info(captions)
self.w.update(b)
addr = self.host + ':' + str(self.port)
b.addr.set_text(addr)
b.restart.set_tooltip("Restart the server")
b.restart.add_callback('activated', self.restart_cb)
b.set_addr.set_length(100)
b.set_addr.set_text(addr)
b.set_addr.set_tooltip("Set address to run remote control server")
b.set_addr.add_callback('activated', self.set_addr_cb)
fr.set_widget(w)
vbox.add_widget(fr, stretch=0)
# stretch
vbox.add_widget(Widgets.Label(''), stretch=1)
btns = Widgets.HBox()
btns.set_spacing(4)
btns.set_border_width(4)
btn = Widgets.Button("Close")
btn.add_callback('activated', lambda w: self.close())
btns.add_widget(btn)
btns.add_widget(Widgets.Label(''), stretch=1)
vbox.add_widget(btns)
container.add_widget(vbox, stretch=1)
def start(self):
self.robj = GingaWrapper(self.fv, self.logger)
self.server = grc.RemoteServer(self.robj, host=self.host, port=self.port,
ev_quit=self.fv.ev_quit)
self.server.start(thread_pool=self.fv.get_threadPool())
def stop(self):
self.server.stop()
def restart_cb(self, w):
# restart server
self.server.stop()
self.start()
def set_addr_cb(self, w):
# get and parse address
addr = w.get_text()
host, port = addr.split(':')
self.host = host
self.port = int(port)
self.w.addr.set_text(addr)
def close(self):
self.fv.stop_global_plugin(str(self))
return True
def __str__(self):
return 'rc'
class GingaWrapper(object):
def __init__(self, fv, logger):
self.fv = fv
self.logger = logger
# List of XML-RPC acceptable return types
self.ok_types = list(map(type, [str, int, float, bool, list, tuple]))
def help(self, *args):
"""Get help for a remote interface method.
Examples
--------
help('ginga', `method`)
name of the method for which you want help
help('channel', `chname`, `method`)
name of the method in the channel for which you want help
Returns
-------
help: string
a help message
"""
if len(args) == 0:
return help_msg
which = args[0].lower()
if which == 'ginga':
method = args[1]
_method = getattr(self.fv, method)
return _method.__doc__
elif which == 'channel':
chname = args[1]
method = args[2]
chinfo = self.fv.get_channel(chname)
_method = getattr(chinfo.viewer, method)
return _method.__doc__
else:
return "Please use 'help ginga <method>' or 'help channel <chname> <method>'"
def load_buffer(self, imname, chname, img_buf, dims, dtype,
header, metadata, compressed):
"""Display a FITS image buffer.
Parameters
----------
`imname`: string
a name to use for the image in Ginga
`chname`: string
channel in which to load the image
`img_buf`: string
the image data, encoded as a base64 ascii encoded string
`dims`: tuple
image dimensions in pixels (usually (height, width))
`dtype`: string
numpy data type of encoding (e.g. 'float32')
`header`: dict
fits file header as a dictionary
`metadata`: dict
other metadata about image to attach to image
`compressed`: bool
decompress buffer using "bz2"
Returns
-------
0
Notes
-----
* Get array dims: data.shape
* Get array dtype: str(data.dtype)
* Make a string from a numpy array: buf = data.tostring()
* Compress the buffer: buf = bz2.compress(buf)
* Convert buffer to ascii-encoding: buf = binascii.b2a_base64(buf)
"""
self.logger.info("received image data len=%d" % (len(img_buf)))
# Unpack the data
try:
# Decode binary data
img_buf = binascii.a2b_base64(img_buf)
# Uncompress data if necessary
decompress = metadata.get('decompress', None)
if compressed or (decompress == 'bz2'):
img_buf = bz2.decompress(img_buf)
# dtype string works for most instances
if dtype == '':
dtype = numpy.float32
byteswap = metadata.get('byteswap', False)
# Create image container
image = AstroImage.AstroImage(logger=self.logger)
image.load_buffer(img_buf, dims, dtype, byteswap=byteswap,
metadata=metadata)
image.update_keywords(header)
image.set(name=imname, path=None)
except Exception as e:
# Some kind of error unpacking the data
errmsg = "Error creating image data for '%s': %s" % (
imname, str(e))
self.logger.error(errmsg)
raise GingaPlugin.PluginError(errmsg)
# Display the image
channel = self.fv.gui_call(self.fv.get_channel_on_demand, chname)
# Note: this little hack needed to let window resize in time for
# file to auto-size properly
self.fv.gui_do(self.fv.change_channel, channel.name)
self.fv.gui_do(self.fv.add_image, imname, image,
chname=channel.name)
return 0
def load_fits_buffer(self, imname, chname, file_buf, num_hdu,
metadata):
from astropy.io import fits
# Unpack the data
try:
# Decode binary data
file_buf = binascii.a2b_base64(file_buf)
# Uncompress data if necessary
decompress = metadata.get('decompress', None)
if decompress == 'bz2':
file_buf = bz2.decompress(file_buf)
self.logger.info("received data: len=%d num_hdu=%d" % (
len(file_buf), num_hdu))
in_f = BytesIO(file_buf)
# Create image container
with fits.open(in_f, 'readonly') as fits_f:
image = AstroImage.AstroImage(metadata=metadata,
logger=self.logger)
image.load_hdu(fits_f[num_hdu], fobj=fits_f)
image.set(name=imname, path=None, idx=num_hdu)
except Exception as e:
# Some kind of error unpacking the data
errmsg = "Error creating image data for '%s': %s" % (
imname, str(e))
self.logger.error(errmsg)
raise GingaPlugin.PluginError(errmsg)
# Display the image
channel = self.fv.gui_call(self.fv.get_channel_on_demand, chname)
# Note: this little hack needed to let window resize in time for
# file to auto-size properly
self.fv.gui_do(self.fv.change_channel, channel.name)
self.fv.gui_do(self.fv.add_image, imname, image,
chname=channel.name)
return 0
def channel(self, chname, method_name, *args, **kwdargs):
chinfo = self.fv.get_channel(chname)
_method = getattr(chinfo.viewer, method_name)
return self.fv.gui_call(_method, *args, **kwdargs)
def ginga(self, method_name, *args, **kwdargs):
_method = getattr(self.fv, method_name)
return self.fv.gui_call(_method, *args, **kwdargs)
#END
|