Newer
Older
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API for Camera related operations'''
Davide Ricci
committed
# System modules
# Third-party modules
from .baseresource import ResourceDev
from .sequencer_instance import seq
"""Binning of the camera."""
# def get(self):
# """Retrieve the binning of the camera."""
# res = {
# "response": self.dev.binning,
# "error": self.dev.error,
def put(self):
"""Set a new binning for the camera."""
"response": self.dev.binning,
"error": self.dev.error,
}
return res
"""Manage the CCD cooler status"""
def get(self):
"""Check wether the CCD cooler is on or off."""
res = {
"raw": self.dev.cooler,
"response": constants.on_off[self.dev.cooler],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
def put(self):
"""Set on or off the CCD cooler."""
"response": self.dev.cooler,
"error": self.dev.error,
}
return res
# # @api.route("/cooler/fan")
# class CoolerFan(ResourceDev):
# """Info on the cooler power % of stress"""
Davide Ricci
committed
# def get(self):
# """Retrieve the CCD cooler power percentage."""
# res = {
# "response": self.dev.fan,
# "error": self.dev.error,
# # @api.route("/cooler/temperature")
# class CoolerTemperature(ResourceDev):
# """Info on the cooler temperature"""
# def get(self):
# """Retrieve the CCD cooler temperature."""
# res = {
# "response": self.dev.temperature,
# "error": self.dev.error,
# @api.route("/cooler/temperature/setpoint")
"""Manage the CCD temperature"""
# def get(self):
# """Retrieve the temperature of the CCD."""
# res = {
# "response": self.dev.setpoint,
# "error": self.dev.error,
def put(self):
"""Set a new temperature of the CCD."""
res = {
"response": self.dev.temperature,
"error": self.dev.error,
}
return res
"""Camera filters names."""
def get(self):
"""Retrieve the filter names."""
res = {
"response": constants.filter_number,
}
return res
"""Camera filter information."""
def get(self):
"""Retrieve the current filter."""
res = {
"raw": self.dev.filter,
"response": constants.filter_name[self.dev.filter],
}
return res
# @api.route("/filter/movement")
"""Manage the camera filter wheel."""
def get(self):
"""Check if the filter wheel is moving."""
res = {
"raw": self.dev.is_moving,
"response": constants.filter_state[self.dev.is_moving],
}
return res
def post(self):
"""Set a new filter."""
res = {
"response": self.dev.filter,
"error": self.dev.error,
}
return res
# # @api.route("/frame")
# class Frame(ResourceDev):
# """Camera frame information."""
# def get(self):
# """Retrieve the current frame."""
# frame = {
# "xrange": self.dev.xrange,
# "yrange": self.dev.yrange,
# "binning": self.dev.binning,
# "maxrange": self.dev.max_range,
# }
# res = {
# "response": frame,
# "error": self.dev.error,
"""Camera custom frame."""
def put(self):
"""Set a custom windowing."""
self.dev.binning = new_frame["binning"]
frame = {
"xrange": new_frame["xrange"],
"yrange": new_frame["yrange"],
"binning": new_frame["binning"]
}
res = {
"response": frame,
}
return res
"""Camera full frame."""
def put(self):
"""Set the ccd to full frame in current binning."""
res = {
"response": self.dev.full_frame(),
"error": self.dev.error,
}
return res
"""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": self.dev.half_frame(),
"error": self.dev.error,
}
return res
"""Set a center ccd window spanning 2 arcminutes
res = {
"response": self.dev.small_frame(),
"error": self.dev.error,
}
return res
# # @api.route("/frame/temperature")
# class FrameTemperature(ResourceDev):
# """Ambient temperature."""
# def get(self):
# """Retrieve the ambient temperature"""
# res = {
# "response": self.dev.ambient,
# "error": self.dev.error,
"""The acquired image."""
# def get(self):
# """Get the acquired image."""
# res = {
# "response": self.dev.download(),
# "error": self.dev.error,
def post(self):
new = request.json # exptime, type
res = {
"response": self.dev.start(new["exptime"],
new["type"],
}
return res
def delete(self):
"""Stop the acquisition process."""
res = {
"response": self.dev.abort(),
"error": self.dev.error,
}
return res
# @api.route("/snapshot/state")
"""Manage the acquisition of an image."""
def get(self):
"""Retrieve the status of the acquisition process."""
res = {
"raw": self.dev.state,
"response": constants.camera_state[self.dev.state],
# @api.route("/snapshot/acquisition")
"""Manage the acquisition of an image using th sequencer."""
def get(self, seq=seq):
"""Retrieve the status of the acquisition process."""
res = {
Davide Ricci
committed
"response": {
"paused": seq.tpl.paused if seq.tpl else None,
"quitting": seq.tpl.aborted if seq.tpl else None,
Davide Ricci
committed
},
"error": "ciccio",
"timestamp": self.timestamp,
return res
def post(self, seq=seq):
"""Start a new acquisition."""
data = request.json # exptime, type, repeat, recenter
# Run the template given its parameters in the json.
seq.tpl.run(data)
res = {
"response": seq.tpl.filenames,
return res
def delete(self, seq=seq):
Davide Ricci
committed
"""Stop the acquisition process."""
res = {
"response": seq.tpl.abort(),
Davide Ricci
committed
}
return res
# @api.route("/snapshot/recenter")
"""Manage the recentering via the the apply_offset function"""
def get(self, seq=seq):
"""Retrieve the recenter status and parameters of the box."""
res = {
"response": {
"recenter": getattr(seq.tpl, 'recenter', None),
"box": getattr(seq.tpl, 'box', None),
"error": "ciccio",
"timestamp": self.timestamp,
}
return res
def put(self, seq=seq):
"""Set a given box to the recentering function"""
data = request.json # exptime, type, repeat, recenter
seq.tpl.box = data
"response": getattr(seq.tpl, 'box', None),
"error": "ciccio",
"timestamp": self.timestamp,
def post(self, seq=seq):
"""Enable the recentering function."""
seq.tpl.recenter = True
"response": getattr(seq.tpl, 'recenter', None),
"error": "ciccio",
"timestamp": self.timestamp,
}
return res
def delete(self, seq=seq):
"""Disable the recentering function."""
seq.tpl.recenter = False
"response": getattr(seq.tpl, 'recenter', None),
"error": "ciccio",
"timestamp": self.timestamp,
# @api.route("/snapshot/domeslewing")
"""Get dome slewing status"""
Andrea Damonte
committed
def get(self, seq=seq):
Andrea Damonte
committed
"""Retrieve the domeslewing status"""
res = {
"response": {
"domeslewing": getattr(seq.tpl, 'domeslewing', None),
Andrea Damonte
committed
},
"error": "ciccio",
"timestamp": self.timestamp,
Andrea Damonte
committed
}
return res
def post(self, seq=seq):
Andrea Damonte
committed
"""Enable the domeslewing function."""
seq.tpl.domeslewing = True
Andrea Damonte
committed
res = {
"response": getattr(seq.tpl, 'domeslewing', None),
"error": "ciccio",
"timestamp": self.timestamp,
Andrea Damonte
committed
}
return res
def delete(self, seq=seq):
Andrea Damonte
committed
"""Disable the domeslewing function."""
seq.tpl.domeslewing = False
Andrea Damonte
committed
res = {
"response": getattr(seq.tpl, 'domeslewing', None),
"error": "ciccio",
"timestamp": self.timestamp,
Andrea Damonte
committed
}
return res
"""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."""
'''Manage the connection to ASCOM.'''
def get(self):
'''Check if the telescope is connected to ASCOM.'''
res = {
"response": self.dev.connection,
"error": self.dev.error,
'''General telescope settings.'''
def get(self):
'''Retrieve all-in-one the settings of the camera.'''
res = {
"response": self.dev.all,
"error": self.dev.error,
# # @api.route("/status")
# class Status(ResourceDev):
# '''General camera status.'''
Davide Ricci
committed
# def get(self):
# '''Retrieve the status of each compoent.'''
# connection = Connection(dev=self.dev).get()
# if connection["response"]:
# settings = Settings(dev=self.dev).get() etc
# else:
# res = {
# "connection": connection,
# }