camera.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

'''REST API for Camera related operations'''

# Third-party modules
from astropy.time import Time
from flask_restx import Namespace, Resource, fields

# Custom modules
import devices
from config import constants

# Other templates
from templates import observation

api = Namespace('camera', description='Camera related operations')


@api.route("/frame/binning")
class FrameBinning(Resource):
    """Binning of the camera."""

    def get(self):
        """Retrieve the binning of the camera."""
        res = {
            "response": devices.cam.binning,
            "error": devices.cam.error,
        }
        return res

    def put(self):
        """Set a new binning for the camera."""
        binning = api.payload
        devices.cam.binning = binning
        res = {
            "response": devices.cam.binning,
            "error": devices.cam.error,
        }
        return res


@api.route("/cooler")
class Cooler(Resource):
    """Manage the CCD cooler status"""

    @api.response(200, 'Success', fields.Boolean())
    def get(self):
        """Check wether the CCD cooler is on or off."""
        res = {
            "response": devices.cam.cooler,
            "error": devices.cam.error,
        }
        return res

    @api.response(200, 'Success', fields.Boolean())
    def put(self):
        """Set on or off the CCD cooler."""
        state = api.payload
        devices.cam.cooler = state
        res = {
            "response": devices.cam.cooler,
            "error": devices.cam.error,
        }
        return res


@api.route("/cooler/fan")
class CoolerFan(Resource):
    """Info on the cooler power % of stress"""

    @api.response(200, 'Success', fields.Float())
    def get(self):
        """Retrieve the CCD cooler power percentage."""
        res = {
            "response": devices.cam.fan,
            "error": devices.cam.error,
        }
        return res


@api.route("/cooler/temperature")
class CoolerTemperature(Resource):
    """Info on the cooler temperature"""

    @api.response(200, 'Success', fields.Float())
    def get(self):
        """Retrieve the CCD cooler temperature."""
        res = {
            "response": devices.cam.temperature,
            "error": devices.cam.error,
        }
        return res


@api.route("/cooler/temperature/setpoint")
class CoolerTemperatureSetpoint(Resource):
    """Manage the CCD temperature"""

    @api.response(200, 'Success', fields.Float())
    def get(self):
        """Retrieve the temperature of the CCD."""
        res = {
            "response": devices.cam.setpoint,
            "error": devices.cam.error,
        }
        return res

    @api.response(200, 'Success', fields.Float())
    def put(self):
        """Set a new temperature of the CCD."""
        state = api.payload
        devices.cam.temperature = state
        res = {
            "response": devices.cam.temperature,
            "error": devices.cam.error,
        }
        return res


@api.route("/filters")
class Filters(Resource):
    """Camera filters names."""

    def get(self):
        """Retrieve the filter names."""
        res = {
            "response": constants.filter_number,
            "error": devices.cam.error,
        }
        return res


@api.route("/filter")
class Filter(Resource):
    """Camera filter information."""

    def get(self):
        """Retrieve the current filter."""
        res = {
            "response": devices.cam.filter,
            "error": devices.cam.error,
        }
        return res


@api.route("/filter/movement")
class FilterMovement(Resource):
    """Manage the camera filter wheel."""

    @api.response(200, 'Success', fields.Boolean())
    def get(self):
        """Check if the filter wheel is moving."""
        res = {
            "response": devices.cam.is_moving,
            "error": devices.cam.error,
        }
        return res

    @api.response(200, 'Success', fields.Float())
    def post(self):
        """Set a new filter."""
        target = api.payload
        devices.cam.filter = target
        res = {
            "response": devices.cam.filter,
            "error": devices.cam.error,
        }
        return res


@api.route("/frame")
class Frame(Resource):
    """Camera frame information."""

    def get(self):
        """Retrieve the current frame."""
        frame = {
            "xrange": devices.cam.xrange,
            "yrange": devices.cam.yrange,
            "binning": devices.cam.binning,
            "maxrange": devices.cam.max_range,
        }
        res = {
            "response": frame,
            "error": devices.cam.error,
        }
        return res


@api.route("/frame/custom")
class FrameCustom(Resource):
    """Camera custom frame."""

    def put(self):
        """Set a custom windowing."""
        new_frame = api.payload
        devices.cam.binning = new_frame["binning"]
        frame = {
            "xrange": new_frame["xrange"],
            "yrange": new_frame["yrange"],
            "binning": new_frame["binning"]
        }
        res = {
            "response": frame,
            "error": devices.cam.error,
        }
        return res


@api.route("/frame/full")
class FrameFull(Resource):
    """Camera full frame."""

    def put(self):
        """Set the ccd to full frame in current binning."""
        res = {
            "response": devices.cam.full_frame(),
            "error": devices.cam.error,
        }
        return res


@api.route("/frame/half")
class FrameHalf(Resource):
    """Camera frame of half size the full frame."""

    def put(self):
        """Set a center ccd window spanning half the
        size of the full frame in the current binning."""
        res = {
            "response": devices.cam.half_frame(),
            "error": devices.cam.error,
        }
        return res


@api.route("/frame/small")
class FrameSmall(Resource):
    """Camera frame of 2 arcmin."""

    def put(self):
        """Set a center ccd window spanning 2 arcminutes
        on the sky."""
        res = {
            "response": devices.cam.small_frame(),
            "error": devices.cam.error,
        }
        return res


@api.route("/frame/temperature")
class FrameTemperature(Resource):
    """Ambient temperature."""

    def get(self):
        """Retrieve the ambient temperature"""
        res = {
            "response": devices.cam.ambient,
            "error": devices.cam.error,
        }
        return res


@api.route("/snapshot/")
class Snapshot(Resource):
    """The acquired image."""

    def get(self):
        """Get the acquired image."""
        res = {
            "response": devices.cam.download(),
            "error": devices.cam.error,
        }
        return res


@api.route("/snapshot/state")
class SnapshotState(Resource):
    """Manage the acquisition of an image."""

    def get(self):
        """Retrieve the status of the acquisition process."""
        res = {
            "response": devices.cam.state,
            "error": devices.cam.error,
        }
        return res

    # def post(self):
    #     """Start a new acquisition."""
    #     new = api.payload  # exptime, type
    #     gps = Time(devices.ast.clock, format="unix")
    #     res = {
    #         "response": devices.cam.start(new["exptime"],
    #                                       new["type"],
    #                                       gps),
    #         "error": devices.cam.error,
    #     }
    #     return res

    # def delete(self):
    #     """Stop the acquisition process."""
    #     res = {
    #         "response": devices.cam.abort(),
    #         "error": devices.cam.error,
    #     }
    #     return res


##### DAVIDE TEST TO CONTROL APPLY_OFFSET LIVE ###
tpl = observation.Template()

@api.route("/snapshot/acquisition")
class SnapshotAcquisition(Resource):
    """Manage the acquisition of an image."""

    # # Every template module has a Template() class.
    # tpl = observation.Template()

    def get(self, tpl=tpl):
        """Retrieve the status of the acquisition process."""

        res = {
            "response": {
                "paused": tpl.paused,
                "quitting": tpl.aborted,
            },
            "error": tpl.error,
        }

        return res

    def post(self, tpl=tpl):
        """Start a new acquisition."""
        data = api.payload  # exptime, type, repeat, recenter

        # Run the template given its parameters in the json.
        tpl.run(data)

        res = {
            "response": tpl.filenames,
            "error": devices.cam.error,
        }

        return res

    def delete(self, tpl=tpl):
        """Stop the acquisition process."""
        res = {
            "response": tpl.abort(),
            "error": devices.cam.error,
        }
        return res

@api.route("/snapshot/recenter")
class SnapshotAcquisition(Resource):
    """Manage the recentering via the the apply_offset function"""

    def get(self, tpl=tpl):
        """Retrieve the recenter status and parameters of the box."""

        res = {
            "response": {
                "recenter": tpl.recenter,
                "box": tpl.box,
            },
            "error": tpl.error,
        }

        return res

    def put(self, tpl=tpl):
        """Set a given box to the recentering function"""
        data = api.payload  # exptime, type, repeat, recenter

        tpl.box = data

        res = {
            "response": tpl.box,
            "error": tpl.error,
        }

    def post(self, tpl=tpl):
        """Enable the recentering function."""

        tpl.recenter = True

        res = {
            "response": tpl.recenter,
            "error": tpl.error,
        }

        return res

    def delete(self, tpl=tpl):
        """Disable the recentering function."""

        tpl.recenter = False

        res = {
            "response": tpl.recenter,
            "error": tpl.error,
        }
        return res


@api.route("/cooler/warmup")
class CoolerWarmup(Resource):
    """Manage the warmup of the CCD."""

    def post(self):
        """Start the warm up the CCD."""

    def delete(self):
        """Stop the warm up of the CCD."""


@api.route("/power")
class Power(Resource):
    """Manage the CCD camera power using the external switch"""

    @api.response(200, 'Success', fields.Boolean())
    def get(self):
        """Check if the camera is on or off."""
        res = {
            "response": devices.sof.state,
            "error": devices.sof.error,
        }
        return res

    @api.response(200, 'Success', fields.Boolean())
    def put(self):
        """Switch on or off the camera"""
        target = api.payload
        devices.sof.state = target
        res = {
            "response": devices.sof.state,
            "error": devices.sof.error,
        }
        return res


@api.route("/power/cycle")
class PowerCycle(Resource):
    """Manage the CCD camera reboot."""

    def post(self):
        """Reboot the camera."""
        res = {
            "response": devices.sof.reboot(),
            "error": devices.sof.error,
        }
        return res


@api.route("/settings")
class Settings(Resource):
    '''General telescope settings.'''

    def get(self):
        '''Retrieve all-in-one the settings of the camera.'''

        res = {
            "response": devices.cam.all,
            "error": devices.cam.error,
        }
        return res


@api.route("/status")
class Status(Resource):
    '''General telescope status.'''

    def get(self):
        '''Retrieve the status of each compoent.'''

        power = Power().get() # sof
        if power["response"]:
            settings = Settings().get() #cam.all
            res = {
                "power":  power, # sof
                "settings": settings,
                "filter":  Filter().get(),
                "movement":  FilterMovement().get(),
                "frame":  Frame().get(),
                "status":  SnapshotAcquisition().get(),
                "state":  SnapshotState().get(),
            }
        else:
            res = {
                "power":  power, #sof
            }

        return res