Source code for hscmap.window

from .channel import Channel
from .callback import CallbackSet
from .catalog import CatalogManager
from .fitsimage import FitsImageManager
from .snapshot import Snapshot
from .utils import as_json, uid
from .typecheck import TypeCheck
from .hook import Hook
from .config import config


class Window:
    '''
    Represents a hscMap window.
    This can be also accessed with :class:`hscmap.Window<hscmap.window.Window>` instead of :class:`hscmap.window.Window`.

    Example ::

        import hscmap

        w = hscmap.Window(title="My Window")
    '''

    _instances = set()

    def __init__(self, *, title=None, mode=None, url=None):
        '''
        Args:
            title (str): Title of the new window
            mode (str): Insert mode. one of split-{top, left, right, bottom}, tab-{before, after}
        '''
        self.hook = Hook()
        self._cb = {}
        self._title = title or f'hscMap'
        self._sync_state = {}
        self._channel = Channel('hscmap/new-window',
                                'hscmap/reconnect', self._on_comm_msg)
        self._callback = CallbackSet(on_error=self._on_callback_error)

        self.catalogs = CatalogManager(self)
        '''
        Catalog manager (:class:`~hscmap.catalog.CatalogManager`) for this window.
        You can create a catalog through this object.
        '''

        self.fits_images = FitsImageManager(self)
        '''
        Fits image manager (:class:`~hscmap.fitsimage.FitsImageManager`) for this window.
        You can add a fits image through this object.
        '''

        self.snapshot = Snapshot(self)
        '''
        See Also:
            :class:`hscmap.snapshot.Snapshot`
        '''

        self._open(mode, url)
        self._instances.add(self)

    def close(self):
        '''
        Closes this window
        '''
        self._channel.send({
            'type': 'close',
        })

    def _on_close(self):
        self._channel.close()
        self._instances.discard(self)
        self.hook.call('close')

    def _open(self, mode, url):
        mode = mode or config.default_insert_mode
        url = url or config.default_url
        assert mode in ['split-top', 'split-left', 'split-right',
                        'split-bottom', 'tab-before', 'tab-after']
        self._channel.send({
            'url': url,
            'title': self._title, 'mode': mode,
        })

    def _on_comm_msg(self, msg):
        msg_type = msg['content']['data']['type']
        args = msg['content']['data'].get('args', {})
        if msg_type == 'close':
            self._on_close()
        elif msg_type == 'callback':
            self._callback.call(args['cbid'], args['args'])
        elif msg_type == 'sync_from_frontend':
            self._sync_from_frontend(args)
        else:
            raise RuntimeError(f'unknwon message type: {msg_type}')

    def _on_callback_error(self, error):
        self._channel.send({
            'type': 'alert',
            'args': {'message': str(error)},
        })

    def jump_to(self, ra, dec, fov):
        '''
        Sets camera to the position.

        Args:
            ra (float): RA in dgrees
            dec (float): Ddec in decrees
            fov (float): Field of view in degrees

        Example: ::

            w.jump_to(10, 41, 1)
        '''
        args = {
            'ra': ra,
            'dec': dec,
            'fov': fov,
        }
        self._channel.send({'type': 'jump_to', 'args': args})

    @property
    def title(self):
        '''
        Window title. Can be set or read.
        '''
        return self._title

    @title.setter
    def title(self, title):
        TypeCheck(str)(title)
        self._title = title
        self._channel.send({'type': 'set_title', 'args': {'title': title}})

    def _sync_from_frontend(self, args):
        for name, value in args.items():
            self._sync_state[name] = value

    def _restore(self):
        self._channel._reconnect()
        self._channel.send({
            'type': 'sync_from_kernel',
            'args': self._sync_state,
        })
        self.hook.call('restore')

    @classmethod
    def _restore_instances(cls):
        for w in cls._instances:
            w._restore()