Skip to content
Commits on Source (110)
Expand all Hide whitespace changes
Inline Side-by-side
.gitignore
View file @ b8fce05f
......@@ -2,7 +2,6 @@
*#
.#*
__pycache__
web/static/js9/
ob/*json
data/fits/20*
data/log/OARPAF*
\ No newline at end of file
README.md
View file @ b8fce05f
......@@ -2,24 +2,11 @@
OARPAF Instrument Control Software
## git configuration on observatory pc
To push directly on observatory pc (for people without access to INAF repo)
```
# on obseravatory pc
git config receive.denyCurrentBranch updateInstead
# on local machine
git remote add machine user@machine:software-di-controllo
git push machine main
```
## System dependencies:
# System dependencies:
Maybe I'll trop these:
```
sudo apt install gnuplot gnupot-x11
sudo /home/zap/MBB/mbb -w 9000 -i 22000 -l /home/zap/MBB/mbb.log 10.185.119.248
```
## Python dependencies:
......@@ -29,13 +16,38 @@ sudo /home/zap/MBB/mbb -w 9000 -i 22000 -l /home/zap/MBB/mbb.log 10.185.119.248
pip3 install requests loguru astropy
pip3 install gnuplotlib pyvantagepro # maybe I'll drop these.
# For the API and the web interface:
pip3 install flask flask-restx flask-httpauth flask-apscheduler werkzeug
pip3 install requests ping3 # maybe I'll drop these. It's just in home page.
# For the API interface:
pip3 install flask flask-restx flask-cors
# Tested with flask==3.0.2 flask-restx==1.3.0 flask-httpauth==4.8.0 flask-apscheduler==1.13.1 werkzeug-3.0.1
./app.py --noweb
./app.py --noweb port hostname_or_ip
# http://server:port/api
# for the web pages:
pip3 install flask flask-socketio flask-httpauth
./app.py
./app.py port hostname_or_ip
# http://server:port/api
# http://server:port/web
```
# Dome board:
## git configuration on observatory pc
To push directly on observatory pc (for people without access to INAF repo)
```
# on obseravatory pc
git config receive.denyCurrentBranch updateInstead
# on local machine
git remote add machine user@machine:software-di-controllo
git push machine main
```
## Dome board:
```
sudo apt-get install python-dev libusb-dev swig libwxgtk3.0-gtk3-dev
......@@ -57,11 +69,20 @@ sudo make pyinstall guiinstall
git clone https://github.com/lfini/SW-OPC-2.git
```
## Siemens modbus interface
```
sudo apt install gnuplot gnupot-x11
sudo /home/zap/MBB/mbb -w 9000 -i 22000 -l /home/zap/MBB/mbb.log 10.185.119.248
```
# Branches:
- main
- dev
[![Latest Release](https://www.ict.inaf.it/gitlab/davide.ricci/software-di-controllo/-/badges/release.svg)](https://www.ict.inaf.it/gitlab/davide.ricci/software-di-controllo/-/releases)
<!-- [![Latest Release](https://www.ict.inaf.it/gitlab/davide.ricci/software-di-controllo/-/badges/release.svg)](https://www.ict.inaf.it/gitlab/davide.ricci/software-di-controllo/-/releases) -->
api/__init__.py
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
############################
# REST API
############################
'''Automatic REST APIs'''
# System modules
import configparser
import importlib
import sys
# Third-party modules
from flask import Blueprint
from flask_httpauth import HTTPBasicAuth
from flask_restx import Api, reqparse
from flask_restx import Api, Namespace, Resource
# Custom modules
import devices
from utils.url_stuff import build_url
from .blocks import api as blocks_api
from .camera import api as camera_api
from .dome import api as dome_api
from .environment import api as environment_api
from .home import api as home_api
from .other import api as other_api
from .telescope import api as telescope_api
from .templates import api as templates_api
from .webcam import api as webcam_api
from .display import api as display_api
from .testfulvio import api as testfulvio_api
from .testfulvio import api as fulvio_api
api_blueprint = Blueprint('api', __name__)
api = Api(api_blueprint,
title='OARPAF APIs for observatory control',
title='Generic APIs for observatory control',
version='1.0',
description='A description',
description=['A description']
)
api.add_namespace(home_api)
api.add_namespace(blocks_api)
api.add_namespace(templates_api)
api.add_namespace(other_api)
base = "api." # this module
api.add_namespace(dome_api)
api.add_namespace(telescope_api)
api.add_namespace(camera_api)
api.add_namespace(webcam_api)
api.add_namespace(display_api)
api.add_namespace(environment_api)
api.add_namespace(fulvio_api)
ends = configparser.ConfigParser()
ends.read('./config/api.ini')
def dynamic_import(url):
"""Dynamically import into this module api.ini files.
"""
############################
# WEB VIEW
############################
# System modules
import json # for users
parts = url.split('/') # /dome/shutter/movement
namespace = parts[1] # dome
end = '/' + '/'.join(parts[2:]) # /shutter/movement
# Third-party modules
from .blocks import web as blocks_web
from .control import web as control_web
from .environment import web as environment_web
from .home import web as home_web
from .init import web as init_web
from .other import web as other_web
from .templates import web as templates_web
from .webcam import web as webcam_web
from .display import web as display_web
from .testfulvio import web as fulvio_web
auth = HTTPBasicAuth()
dev_api = Namespace(namespace) # adding namespace /dome. Endpoint will be added
users_file = 'users.json'
dev = getattr(devices, ends.get(url,"device")) # devices.light instance
cls = dev.__class__.__name__ # string "Switch"
mod_name = cls.lower() # string "switch"
module = importlib.import_module(base + mod_name ) # module api.switch
module_class = getattr(module, ends.get(url,"resource")) # class api.switch.State
try:
with open(users_file) as json_file:
users = json.load(json_file)
except FileNotFoundError as e:
print(f"{users_file} containing user:pass not found")
exit(0)
# Assigning endpoint to api.ShutterMovement
dev_api.add_resource(module_class, end , resource_class_kwargs={"dev": dev})
api.add_namespace(dev_api) # Adding to /dome: /shutter/movement
for end in ends.sections():
dynamic_import(end)
@auth.verify_password
def verify(username, password):
if not (username and password):
return False
return users.get(username) == password
# Apart, what is not in the api.ini.
api.add_namespace(environment_api)
web_blueprint = Blueprint('web', __name__)
web = Api(web_blueprint,
title='OARPAF Web view for observatory control',
version='1.0',
description='A description',
decorators=[auth.login_required], # on every page.
)
api.add_namespace(blocks_api)
api.add_namespace(templates_api)
web.add_namespace(home_web)
web.add_namespace(blocks_web)
web.add_namespace(templates_web)
web.add_namespace(webcam_web)
web.add_namespace(display_web)
web.add_namespace(environment_web)
web.add_namespace(other_web)
web.add_namespace(control_web)
web.add_namespace(init_web)
web.add_namespace(fulvio_web)
api.add_namespace(testfulvio_api)
api/baseresource.py 0 → 100644
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''Estension of flask_restx Resource to pass the device in the constructor'''
# System modules
from datetime import datetime
# Third-party modules
from flask_restx import Resource
class ResourceDev(Resource):
'''Add the device to the constructor of the Resource class.
So other resources will inherit from this,
without always passing it to the constructor'''
def __init__(self, *args, **kwargs):
'''Constructor.'''
super().__init__(self, *args, **kwargs)
self.dev = kwargs["dev"]
@property
def timestamp(self):
"""Generate a iso timestamp"""
clock = datetime.utcnow().isoformat()
return clock
api/blocks.py
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API to edit Observation Blocks'''
# Third-party modules
from flask import make_response, render_template, request
from flask_restx import Namespace, Resource
from .data_access_object import ObservationBlockObject as DAO
from .data_access_object import guess
# Custom modules
from utils.data_access_object import ObservationBlockObject as DAO
from utils.data_access_object import guess
dao = DAO("ob")
tpl = DAO("defaults")
############################
# REST API
############################
api = Namespace('blocks', description='Observation blocks')
@api.route("/")
class BlockList(Resource):
"""Show and create OBs"""
......@@ -38,7 +35,6 @@ class BlockList(Resource):
dao.delete(name)
return "", 204
@api.route("/<string:name>")
@api.param("name", "The uniqe name of the OB")
class Block(Resource):
......@@ -53,9 +49,9 @@ class Block(Resource):
def put(self, name):
"""Update the OB"""
data = api.payload
for d in data:
for key, val in d["params"].items():
d["params"][key] = guess(val)
for datum in data:
for key, val in datum["params"].items():
datum["params"][key] = guess(val)
dao.update(name, data)
return "", 204
......@@ -74,50 +70,3 @@ class Block(Resource):
data.pop(instance - 1) # jinjia loop starts from 1 in html
dao.update(name, data)
return "", 204
############################
# WEB VIEW
############################
web = Namespace('blocks', description='OB generator')
@web.route("/")
class BlockListWeb(Resource):
"""Show and create OBs"""
def get(self):
"""Show all observation OB files"""
data = dao.todos
return make_response(render_template("blocks.html", data=data, tpl=tpl.todos))
@web.route("/<string:name>")
@web.param("name", "The uniqe name of the OB")
class BlockWeb(Resource):
"""Show a selected OB, update it, or add a new
template in it.
"""
def get(self, name):
"""Show a specific OB"""
included = request.args.get("included") or False
data = dao.show(name)
#data = block.get(name)
return make_response(render_template("block.html",
data=data, tpl=tpl.todos,
included=included))
@web.route("/<string:name>/<int:instance>")
@web.param("name", "The uniqe name of the OB")
@web.param("instance", "The template instance in the selected OB")
class BlockInstanceWeb(Resource):
def get(self, name, instance):
"""Show a specific template instance of the selected OB"""
data = dao.show(name)
data["content"] = [data["content"][instance]]
return make_response(render_template("block.html", data=data))
# return [data]
api/camera.py
View file @ b8fce05f
This diff is collapsed.
api/display.py deleted 100644 → 0
View file @ b04af93b
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API for fits files'''
# System modules
import time
# Third-party modules
from astropy.time import Time
from flask import Response, make_response, render_template, request
from flask_restx import Namespace, Resource, fields
# Custom modules
# import devices
############################
# REST API
############################
api = Namespace('display', description='Fits files related operations')
# @api.route("/position")
# class Pointing(Resource):
# """Position of the webcam."""
# def get(self):
# """Retrieve the alt az coordinates of the webcam."""
# res = {
# "response": devices.ipcam.altaz,
# "error": devices.ipcam.error,
# }
# return res
# def put(self):
# """Set new alt az coordinates of the webcam."""
# target = api.payload
# devices.ipcam.altaz = target
# res = {
# "response": devices.ipcam.altaz,
# "error": devices.ipcam.error,
# }
# return res
# @api.route("/test2")
# class Pointing(Resource):
# """test."""
# def get(self):
# """BSON test"""
# res = {
# "response": (1.234*1+np.zeros((100,100))).tolist(),
# "error": (1.234*1+np.zeros((100,100))).tolist(),
# }
# # Convert JSON to binary
# b = msgpack.packb(res)
# #json.dumps(res).encode('utf-8')
# response = make_response(b)
# response.headers['Content-Type'] = 'application/octet-stream'
# return response
# def put(self):
# """Set new alt az coordinates of the webcam."""
# target = api.payload
# res = {
# "response": devices.ipcam.altaz,
# "error": devices.ipcam.error,
# }
# return res
# @api.route("/snapshot")
# class Snapshot(Resource):
# """Image from the webcam."""
# def __init__(self, *args, **kwargs):
# '''Constructor.'''
# super().__init__(self)
# self.last = None
# def get(self):
# """Retrieve a raw base/64 image from the webcam."""
# img = devices.ipcam.image
# if not img:
# code = 401
# elif devices.ipcam.error:
# code = 501
# else:
# code = 200
# res = {
# "response": img.decode("ISO-8859-1") if img else img,
# "error": devices.ipcam.error,
# }
# self.last = img
# return res, code
############################
# WEB VIEW
############################
web = Namespace('display', description='Display fits file interface')
@web.route("/")
class Init(Resource):
def get(self):
data = {}
return make_response(render_template("display.html", data=data))
api/dome.py
View file @ b8fce05f
......@@ -4,268 +4,197 @@
'''REST API for Dome related operations'''
# Third-party modules
from flask_restx import Namespace, Resource, fields
from flask import request
# Custom modules
import devices
from config import constants
from .baseresource import ResourceDev
api = Namespace('dome', description='Dome related operations')
model_generic = api.model('Generic', {
'error': fields.List(fields.String),
})
model_bool = api.clone('Boolean', model_generic, {
'response': fields.Boolean,
})
model_int = api.clone('Integer', model_generic, {
'response': fields.Integer,
})
model_float = api.clone('Float', model_generic, {
'response': fields.Float,
})
sub_model_position = api.model('SubPosition', {
"azimuth": fields.Float,
"parked": fields.Boolean,
})
model_position = api.clone('Position', model_generic, {
'response': fields.Nested(sub_model_position),
})
@api.route("/light")
class Light(Resource):
'''Manage the dome illumination.'''
@api.marshal_with(model_bool, 200)
def get(self, **kwargs):
'''Check if the dome illumination is on or off.'''
res = {
"response": devices.light.state,
"error": devices.light.error,
}
return res
@api.marshal_with(model_bool, 200)
def put(self):
'''Switch on or off the dome illumination.'''
state = api.payload
devices.light.state = state
res = {
"response": devices.light.state,
"error": devices.light.error,
}
return res
@api.route("/position")
class Position(Resource):
# @api.route("/position")
class Position(ResourceDev):
'''Get the position of the dome.'''
@api.marshal_with(model_position, 200)
def get(self):
'''Check the azimuth of the dome, and if
this corresponds to the parking position.'''
res = {
"raw": self.dev.is_parked,
"response": {
"azimuth": devices.dom.azimuth,
"parked": devices.dom.is_parked,
"azimuth": self.dev.azimuth,
"parked": constants.yes_no[self.dev.is_parked],
},
"error": devices.dom.error,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/shutter")
class Shutter(Resource):
# @api.route("/shutter")
class Shutter(ResourceDev):
'''Dome Shutter'''
@api.marshal_with(model_int, 200)
def get(self):
'''Check the state of the dome shutter.'''
res = {
"response": devices.dom.shutter,
"error": devices.dom.error,
"raw": self.dev.shutter,
"response": constants.open_state[self.dev.shutter],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/shutter/movement")
class ShutterMovement(Resource):
# @api.route("/shutter/movement")
class ShutterMovement(ResourceDev):
'''Manage the movement of the dome shutter.'''
def post(self):
'''Move the shutter to a new position
(open or closed).'''
target = api.payload
devices.dom.open = target
target = request.json
self.dev.open = target
res = {
"response": devices.dom.open,
"error": devices.dom.error,
"raw": self.dev.open,
"response": constants.open_state[self.open],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
def delete(self):
'''Stop the movement of the shutter.'''
res = {
"response": devices.dom.abort(),
"error": devices.dom.error,
"response": self.dev.abort(),
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/position/slaved")
class PositionSlaved(Resource):
# @api.route("/position/slaved")
class PositionSlaved(ResourceDev):
'''Manage the minioning of the shutter
to the telescope.'''
@api.marshal_with(model_bool, 200)
def get(self):
'''Check if the dome is slaved
to the telescope or not.'''
res = {
"response": devices.dom.slave,
"error": devices.dom.error,
"raw": self.dev.slave,
"response": constants.yes_no[self.dev.slave],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.marshal_with(model_bool, 200)
def put(self):
'''Make the dome slaved to the telescope
or release it.'''
slaved = api.payload
devices.dom.slave = slaved
slaved = request.json
self.dev.slave = slaved
res = {
"response": devices.dom.slave,
"error": devices.dom.error,
"raw": self.dev.slave,
"response": constants.yes_no[self.dev.slave],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/position/sync")
class PositionSync(Resource):
# @api.route("/position/sync")
class PositionSync(ResourceDev):
'''Manage the reference azimuth of the dome'''
@api.marshal_with(model_float, 200)
def put(self):
'''Tell the dome that its current position
corresponds to a given azimuth.'''
new_azimuth = api.payload
new_azimuth = request.json
res = {
"response": devices.dom.sync(new_azimuth),
"error": devices.dom.error,
"response": self.dev.sync(new_azimuth),
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/position/movement")
class PositionMovement(Resource):
# @api.route("/position/movement")
class PositionMovement(ResourceDev):
'''Manage the movement of the dome.'''
@api.marshal_with(model_bool, 200)
# @api.marshal_with(model_bool, 200)
def get(self):
'''Return if the dome is moving or not.'''
res = {
"response": devices.dom.is_moving,
"error": devices.dom.error,
"raw": self.dev.is_moving,
"response": constants.yes_no[self.dev.is_moving],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
def delete(self):
'''Stop the movement of the dome.'''
res = {
"response": devices.dom.abort(),
"error": devices.dom.error,
"response": self.dev.abort(),
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/position/movement/park")
class PositionMovementPark(Resource):
# @api.route("/position/movement/park")
class PositionMovementPark(ResourceDev):
'''Manage the dome parking.'''
def post(self):
'''Park the dome.'''
res = {
"response": devices.dom.park(),
"error": devices.dom.error,
"response": self.dev.park(),
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/position/movement/azimuth")
class PositionMovementAzimuth(Resource):
# @api.route("/position/movement/azimuth")
class PositionMovementAzimuth(ResourceDev):
'''Manage the position of the dome.'''
@api.marshal_with(model_float, 200)
def post(self):
'''Set a new dome azimuth.'''
azimuth = api.payload
devices.dom.azimuth = azimuth
azimuth = request.json
self.dev.azimuth = azimuth
res = {
"response": devices.dom.azimuth,
"error": devices.dom.error,
"response": self.dev.azimuth,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/connection")
class Connection(Resource):
# @api.route("/connection")
class Connection(ResourceDev):
'''Manage the connection to ASCOM.'''
def get(self):
'''Check if the dome is connected to ASCOM.'''
'''check if the dome is connected to ASCOM.'''
conn = self.dev.connection
res = {
"response": devices.dom.connection,
"error": devices.dom.error,
"raw": conn,
"response": constants.yes_no[conn] if bool(conn) else False,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.response(200, 'Success', fields.Boolean())
def put(self):
'''Connect or disconnect the telescope to ASCOM.'''
connection = api.payload
devices.dom.connection = connection
connection = request.json
self.dev.connection = connection
res = {
"response": devices.dom.connection,
"error": devices.dom.error,
"raw": self.dev.connection,
"response": constants.yes_no[self.dev.connection],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/status")
class Status(Resource):
'''General dome status.'''
def __init__(self, *args, **kwargs):
'''Constructor.'''
super().__init__(self)
self.last = None
def get(self):
'''Retrieve the status of each compoent.'''
connection = Connection().get()
if connection["response"]:
res = {
"connection": connection,
"light": Light().get(),
"position": Position().get(),
"moving": PositionMovement().get(),
"shutter": Shutter().get(),
"slaved": PositionSlaved().get(),
}
else:
res = {
"connection": connection,
"light": Light().get(),
}
self.last = res
return res
api/environment.py
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API to get Meteo data'''
# Third-party modules
from astropy.time import Time
from flask import Response, make_response, render_template, request
from flask_restx import Model, Namespace, Resource, fields
from flask_restx import Namespace, Resource
# Custom modules
import devices
......@@ -14,7 +13,6 @@ import devices
# REST API
############################
api = Namespace(
'environment', description='Environment data from multiple devices')
......@@ -23,11 +21,6 @@ api = Namespace(
class Status(Resource):
"""Asks all the environment data"""
def __init__(self, *args, **kwargs):
'''Constructor.'''
super().__init__(self)
self.last = None
def get(self):
"""Downloads all the environment values available """
meteo = Meteo().get()
......@@ -35,7 +28,6 @@ class Status(Resource):
# meteo.update(internal)
res = {"external": meteo, "internal": internal}
self.last = res
return res
......@@ -73,19 +65,19 @@ class Internal(Resource):
"""Downloads the current meteo station values"""
try:
telescope_temperatures = devices.tel.temperature
telescope_temperatures = devices.tel_temp.temperature
if not telescope_temperatures:
print("telescope_temperatures device error!")
telescope_temperatures = [None, None, None]
telescope_temperatures = [None, None, None]
except Exception as e:
print("telescope_temperatures exception error:")
print(e)
telescope_temperatures = [None, None, None]
# try:
# if devices.sof.state:
# if devices.sof.state:
# camera_ambient = devices.cam.ambient
# camera_ambient = None
# camera_ambient = None
# except Exception as e:
# log.debug(f"Environment camera_ambient error: {e}")
# camera_ambient = None
......@@ -111,27 +103,3 @@ class Internal(Resource):
}
return res
############################
# WEB VIEW
############################
web = Namespace('environment', description='Meteo Weather page')
@web.route("/")
class EnvironmentWeb(Resource):
""" Weatherpage """
def get(self):
""" Display the meteo """
data = Status().get()
exclude = ["Barometric_trend", "Battery_status", "Battery_volts",
"ET_day", "ET_month", "ET_year",
"Storm_start_date", "Forecast_rule_number", "Forecast_icon"
]
for e in exclude:
data["external"].pop(e, None)
return make_response(render_template("meteo.html", data=data))
api/focuser.py 0 → 100644
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API for Focuser related operations'''
# Third-party modules
from flask import request
from .baseresource import ResourceDev
# @api.route("/focuser")
class Focuser(ResourceDev):
'''Secondary mirror position.'''
def get(self):
'''Retrieve the secondary mirror position.'''
res = {
"response": self.dev.position,
"error": self.dev.error,
}
return res
# @api.route("/focuser/movement")
class FocuserMovement(ResourceDev):
'''Manage the secondary mirror position.'''
def get(self):
'''Check if the secondary mirror is moving.'''
res = {
"response": self.dev.is_moving,
"error": self.dev.error,
}
return res
def put(self):
'''Update the secondary mirror position.'''
target = request.json
self.dev.position = target
res = {
"response": self.dev.position,
"error": self.dev.error,
}
return res
# def delete(self):
# '''Stop the movement.'''
# return
# @api.route("/connection")
class Connection(ResourceDev):
'''Manage the connection to ASCOM.'''
def get(self):
'''Check if the focuser is connected.'''
res = {
"response": self.dev.connection,
"error": self.dev.error,
}
return res
def put(self):
'''Connect or disconnect the focuser.'''
connection = request.json
self.dev.connection = connection
res = {
"response": self.dev.connection,
"error": self.dev.error,
}
return res
api/old/defaults.py deleted 100644 → 0
View file @ b04af93b
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Third-party modules
from flask import make_response, render_template
from flask_restx import Namespace, Resource, fields
from .data_access_object import TemplateObject as DAO
# defaults_folder = "defaults"
# api = Namespace('defaults', description='Defaults template parameters')
web = Namespace('templates', description='Defaults template parameters')
dao = DAO()
@web.route("/")
class TemplateList(Resource):
def get(self):
data = dao.todos
return make_response(render_template("ob_controls.html", data=data))
@web.route("/<string:name>")
class Template(Resource):
def get(self,name):
data = dao.show(name)
data["content"] = [data["content"]]
return make_response(render_template("ob.html", data=data))
api/old/logger.py deleted 100644 → 0
View file @ b04af93b
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# System modules
import json
# adjusted flask_logger
import time
# Third-party modules
from astropy.time import Time
from flask import (Response, make_response, render_template, request,
stream_with_context)
from flask_restx import Namespace, Resource
from loguru import logger
from .data_access_object import ObservationBlockObject as DAO
from .data_access_object import guess
dao = DAO("ob")
tpl = DAO("defaults")
############################
# REST API
############################
api = Namespace('logger', description='Logger')
@api.route("/")
class Logger(Resource):
"""Show and create OBs"""
def get(self):
return make_response(render_template("per-orso.html"))
# return "ppp"
# def generate():
# while True:
# content = Time.now().isot
# yield str(content)
# time.sleep(1)
# return Response(stream_with_context(generate()))
############################
# WEB VIEW
############################
web = Namespace('logger', description='OB generator')
@web.route("/")
class Logger(Resource):
"""Show and create OBs"""
def get(self):
return make_response(render_template("per-orso.html"))
# @web.route("/")
# class BlockListWeb(Resource):
# """Show and create OBs"""
# def get(self):
# """Show all observation OB files"""
# data = dao.todos
# return make_response(render_template("ob_list.html", data=data, tpl=tpl.todos))
# def post(self):
# """Create a new OB file based on name"""
# name = api.payload
# dao.create(name)
# return "", 204
# def delete(self):
# """Delete an OB file based on name"""
# name = api.payload
# dao.delete(name)
# return "", 204
api/old/ob_generator.py deleted 100644 → 0
View file @ b04af93b
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Third-party modules
from flask import make_response, render_template
from flask_restx import Namespace, Resource, fields
from .data_access_object import ObservationBlockObject as DAO
# @web.route("/")
# class (Resource):
# def get(self):
# data = dao.todos
# return make_response(render_template("defaults.html", data=data))
# @web.route("/list")
# class TodoListWeb(Resource):
# def get(self):
# data = dao.todos
# return make_response(render_template("defaults-list.html", data=data))
# @web.route("/<string:id>")
# class TodoWeb(Resource):
# def get(self,id):
# data = [dao.read(id)]
# return make_response(render_template("defaults.html", data=data))
api/rotator.py 0 → 100644
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API for Rotator related operations'''
# Third-party modules
from flask import request
from .baseresource import ResourceDev
# @api.route("/rotator")
class Rotator(ResourceDev):
'''Field derotator position.'''
def get(self):
'''Retrieve the field derotator position.'''
res = {
"response": self.dev.position,
"error": self.dev.error,
}
return res
# @api.route("/rotator/movement")
class RotatorMovement(ResourceDev):
'''Manage the field derotator position.'''
def get(self):
'''Check if the field derotator is moving.'''
res = {
"response": self.dev.is_moving,
"error": self.dev.error,
}
return res
def post(self):
'''Set the field derotator to a new position.'''
target = request.json
self.dev.position = target
res = {
"response": self.dev.position,
"error": self.dev.error,
}
return res
# @api.route("/connection")
class Connection(ResourceDev):
'''Manage the connection to ASCOM.'''
def get(self):
'''Check if the rotator is connected.'''
res = {
"response": self.dev.connection,
"error": self.dev.error,
}
return res
def put(self):
'''Connect or disconnect the rotator.'''
connection = request.json
self.dev.connection = connection
res = {
"response": self.dev.connection,
"error": self.dev.error,
}
return res
api/sequencer_instance.py 0 → 100644
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''Unique sequencer instance for all APIs'''
# Custom modules
from sequencer import Sequencer
seq = Sequencer()
api/switch.py 0 → 100644
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API for Switch related operations'''
# System modules
# Third-party modules
from flask import request
# Custom modules
from config import constants
from .baseresource import ResourceDev
# @api.route("/dome/light")
# @api.route("/telescope/lamp")
# @api.route("/camera/power")
class State(ResourceDev):
'''Manage a switch state.'''
def __init__(self, *args, **kwargs):
'''Constructor.'''
super().__init__(self, *args, **kwargs)
self.dev = kwargs["dev"]
def get(self):
'''Check if the switch on or off.'''
res = {
"raw": self.dev.state,
"response": constants.on_off[self.dev.state],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
def put(self):
'''Switch on or off.'''
state = request.json
self.dev.state = state
res = {
"raw": self.dev.state,
"response": constants.on_off[self.dev.state],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
api/telescope.py
View file @ b8fce05f
......@@ -3,156 +3,49 @@
'''REST API for Telescope related operations'''
# System modules
from datetime import datetime
# Third-party modules
from astropy.time import Time
from flask_restx import Namespace, Resource, fields
from flask import request
# Custom modules
import devices
from config import constants
from utils.coordinates import to_hms, to_hms_dms, to_radec
api = Namespace('telescope', description='Telescope related operations')
@api.route("/lamp")
class Lamp(Resource):
'''Manage the flat field lamp.'''
@api.response(200, 'Success', fields.Boolean())
def get(self):
'''Check if the flat field lamp is on or off.'''
res = {
"response": devices.lamp.state,
"error": devices.lamp.error,
}
return res
@api.response(200, 'Success', fields.Boolean())
# @api.expect(model)
def put(self):
'''Switch on or off the flat field lamp.'''
state = api.payload
devices.lamp.state = state
res = {
"response": devices.lamp.state,
"error": devices.lamp.error,
}
return res
@api.route("/focuser")
class Focuser(Resource):
'''Secondary mirror position.'''
@api.response(200, 'Success', fields.Float())
def get(self):
'''Retrieve the secondary mirror position.'''
res = {
"response": devices.foc.position,
"error": devices.foc.error,
}
return res
@api.route("/focuser/movement")
class FocuserMovement(Resource):
'''Manage the secondary mirror position.'''
@api.response(200, 'Success', fields.Boolean())
def get(self):
'''Check if the secondary mirror is moving.'''
res = {
"response": devices.foc.is_moving,
"error": devices.foc.error,
}
return res
@api.response(200, 'Success', fields.Float())
def put(self):
'''Update the secondary mirror position.'''
target = api.payload
devices.foc.position = target
res = {
"response": devices.foc.position,
"error": devices.foc.error,
}
return res
# def delete(self):
# '''Stop the movement.'''
# return
@api.route("/rotator")
class Rotator(Resource):
'''Field derotator position.'''
@api.response(200, 'Success', fields.Float())
def get(self):
'''Retrieve the field derotator position.'''
res = {
"response": devices.rot.position,
"error": devices.rot.error,
}
return res
@api.route("/rotator/movement")
class RotatorMovement(Resource):
'''Manage the field derotator position.'''
@api.response(200, 'Success', fields.Boolean())
def get(self):
'''Check if the field derotator is moving.'''
res = {
"response": devices.rot.is_moving,
"error": devices.rot.error,
}
return res
@api.response(200, 'Success', fields.Float())
def post(self):
'''Set the field derotator to a new position.'''
target = api.payload
devices.rot.position = target
res = {
"response": devices.rot.position,
"error": devices.rot.error,
}
return res
from .baseresource import ResourceDev
@api.route("/cover")
class Cover(Resource):
# @api.route("/cover")
class Cover(ResourceDev):
'''Petals covering the primary mirror.'''
@api.response(200, 'Success', fields.Float())
def get(self):
'''Retrieve the petals position.'''
res = {
"response": devices.tel.cover,
"error": devices.tel.error,
"response": self.dev.cover,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/cover/movement")
class CoverMovement(Resource):
# @api.route("/cover/movement")
class CoverMovement(ResourceDev):
'''Manage the primary mirror cover petals.'''
# def get(self):
# '''Check if petals are moving.'''
# return
@api.response(200, 'Success', fields.Boolean())
def post(self):
'''Set a new position for cover petals
(open or closed).'''
target = api.payload
devices.tel.open = target
target = request.json # api.payload
self.dev.open = target
res = {
"response": devices.tel.open,
"error": devices.tel.error,
"response": self.dev.open,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
......@@ -161,343 +54,281 @@ class CoverMovement(Resource):
# return
@api.route("/error")
class Error(Resource):
# @api.route("/error")
class Error(ResourceDev):
'''Manage the cabinet errors.'''
@api.response(200, 'Success', fields.String())
def get(self):
'''Check if there are errors.'''
res = {
"response": devices.tel.state, # state is global
"error": devices.tel.error,
"response": self.dev.state, # state is global
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.response(200, 'Success', fields.Integer())
def delete(self):
'''Try to clean the errors.'''
error_number = api.payload
error_number = request.json
res = {
"response": devices.tel.clear(error_number),
"error": devices.tel.error,
"response": self.dev.clear(error_number),
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/error/details")
class ErrorDetails(Resource):
# @api.route("/error/details")
class ErrorDetails(ResourceDev):
'''Get the global status to have more info.'''
@api.response(200, 'Success', fields.String())
def get(self):
'''Check if there are errors.'''
res = {
"response": devices.tel.status, # state is global
"error": devices.tel.error,
"response": self.dev.status, # state is global
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/clock")
class Clock(Resource):
# @api.route("/clock")
class Clock(ResourceDev):
'''Telescope time from GPS.'''
def get(self):
'''Get the telescope time from GPS.'''
res = {
"response": devices.tel.clock,
"error": devices.tel.error,
"response": self.dev.clock,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates")
class Coordinates(Resource):
# @api.route("/coordinates")
class Coordinates(ResourceDev):
'''Telescope coordinates.'''
def get(self):
'''Get the telescope coordinates.'''
coor = self.dev.coordinates
lst = coor["lst"]
ra = coor["radec"][0]
coor["ha"] = to_hms(lst-ra)
coor["radec"] = to_hms_dms(coor["radec"])
coor["lst"] = to_hms(coor["lst"])
utc = coor["utc"]
coor["utc"] = datetime.fromtimestamp(utc).isoformat()
res = {
"response": devices.tel.coordinates,
"error": devices.tel.error,
"response": coor,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/movement")
class CoordinatesMovement(Resource):
# @api.route("/coordinates/movement")
class CoordinatesMovement(ResourceDev):
'''Manage the pointing of the telescope.'''
def get(self):
'''Check the status of the telescope movement.'''
res = {
"response": devices.tel.is_moving,
"error": devices.tel.error,
"response": self.dev.is_moving,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
def delete(self):
'''Stop the telescope movement.'''
res = {
"response": devices.tel.abort(),
"error": devices.tel.error,
"response": self.dev.abort(),
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/movement/radec")
class CoordinatesMovementRadec(Resource):
# @api.route("/coordinates/movement/radec")
class CoordinatesMovementRadec(ResourceDev):
'''Point the telescope in Ra, Dec.'''
def post(self):
'''Set new Ra and Dec coordinates.'''
target = api.payload
target = request.json
radec = to_radec(target)
print(radec)
devices.tel.radec = radec
self.dev.radec = radec
res = {
"response": devices.tel.radec,
"error": devices.tel.error,
"response": self.dev.radec,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/movement/altaz")
class CoordinatesMovementAltaz(Resource):
# @api.route("/coordinates/movement/altaz")
class CoordinatesMovementAltaz(ResourceDev):
'''Point the telescope in Alt, Az.'''
def post(self):
'''Set new Alt and Az coordinates.'''
target = api.payload
devices.tel.altaz = target
target = request.json
self.dev.altaz = target
res = {
"response": devices.tel.altaz,
"error": devices.tel.error
"response": self.dev.altaz,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/movement/atpark")
class CoordinatesMovementAtpark(Resource):
# @api.route("/coordinates/movement/atpark")
class CoordinatesMovementAtpark(ResourceDev):
'''Park the telescope.'''
def get(self):
'''Send the telescope to park position.'''
devices.tel.park
parked = self.dev.park
res = {
"response": devices.tel.park,
"error": devices.tel.error,
"raw": parked,
"response": constants.yes_no[parked],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/movement/park")
class CoordinatesMovementPark(Resource):
# @api.route("/coordinates/movement/park")
class CoordinatesMovementPark(ResourceDev):
'''Park the telescope.'''
def post(self):
'''Send the telescope to park position.'''
devices.tel.park = True
self.dev.park = True
res = {
"response": devices.tel.park,
"error": devices.tel.error,
"response": self.dev.park,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/movement/unpark")
class CoordinatesMovementUnpark(Resource):
# @api.route("/coordinates/movement/unpark")
class CoordinatesMovementUnpark(ResourceDev):
'''Unpark the telescope.'''
def post(self):
'''Remove the telescope from park position.'''
devices.tel.park = False
res = {"response": devices.tel.park,
"error": devices.tel.error,
self.dev.park = False
res = {
"response": self.dev.park,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/offset")
class CoordinatesOffset(Resource):
# @api.route("/coordinates/offset")
class CoordinatesOffset(ResourceDev):
'''Manage the coordinates offset.'''
def get(self):
'''Retrieve the telescope offset in Zdist, Az.'''
res = {
"response": devices.tel.offset,
"error": devices.tel.error,
"response": self.dev.offset,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
def put(self):
'''Apply to the telescope new offsets in degrees.'''
offset = api.payload
devices.tel.offset = offset
offset = request.json
self.dev.offset = offset
res = {
"response": devices.tel.offset,
"error": devices.tel.error,
"response": self.dev.offset,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/coordinates/tracking")
class CoordinatesTracking(Resource):
# @api.route("/coordinates/tracking")
class CoordinatesTracking(ResourceDev):
'''Enable or disable the telescope tracking.'''
def get(self):
'''Check if the telescope is tracking or not.'''
res = {
"response": devices.tel.tracking,
"error": devices.tel.error,
"raw": self.dev.tracking,
"response": constants.yes_no[self.dev.tracking],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.response(200, 'Success', fields.Boolean())
def put(self):
'''Set the telescope in tracking mode or not'''
tracking = api.payload
devices.tel.tracking = tracking
tracking = request.json
self.dev.tracking = tracking
res = {
"response": devices.tel.tracking,
"error": devices.tel.error,
"response": self.dev.tracking,
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/power")
class Power(Resource):
'''Manage the cabinet power.'''
def get(self):
'''Check if the telescope cabinet is on or off.'''
res = {
"response": devices.cab.state,
"error": devices.cab.error,
}
return res
@api.response(200, 'Success', fields.Boolean())
def put(self):
'''Switch on or off the telescope cabinet.'''
power = api.payload
devices.cab.state = power
res = {
"response": devices.cab.state,
"error": devices.cab.error,
}
return res
@api.route("/connection")
class Connection(Resource):
# @api.route("/connection")
class Connection(ResourceDev):
'''Manage the connection to ASCOM.'''
def get(self):
'''Check if the telescope is connected to ASCOM.'''
res = {
"response": devices.tel.connection,
"error": devices.tel.error,
"raw": self.dev.connection,
"response": constants.yes_no[self.dev.connection],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.response(200, 'Success', fields.Boolean())
def put(self):
'''Connect or disconnect the telescope to ASCOM.'''
connection = api.payload
devices.tel.connection = connection
connection = request.json
self.dev.connection = connection
res = {
"response": devices.tel.connection,
"error": devices.tel.error,
"raw": self.dev.connection,
"response": constants.yes_no[self.dev.connection],
"error": self.dev.error,
"timestamp": self.timestamp,
}
return res
@api.route("/status")
class Status(Resource):
'''General telescope status.'''
# clock = Clock(dev=self.dev).get()
# if clock["response"] and type(clock["response"]) != str:
def __init__(self, *args, **kwargs):
'''Constructor.'''
super().__init__(self)
self.last = None
# # 2) astelOS has data from the GPS
def get(self):
'''Retrieve the status of each compoent.'''
power = Power().get() # cab
if power["response"]:
# 1) Powered cabinet
clock = Clock().get()
if clock["response"] and type(clock["response"]) != str:
# 2) astelOS has data from the GPS
coor = Coordinates().get()
lst = coor["response"]["lst"]
ra = coor["response"]["radec"][0]
coor["response"]["ha"] = to_hms(lst-ra)
coor["response"]["radec"] = to_hms_dms(
coor["response"]["radec"])
coor["response"]["lst"] = to_hms(coor["response"]["lst"])
coor["response"]["utc"] = Time(
coor["response"]["utc"], format="unix").isot
connection = Connection().get()
if connection["response"]:
# 3) Telescope manually connected to ASCOM
res = {
"power": power, # cab
"connection": connection,
"parked": CoordinatesMovementAtpark().get(),
"lamp": Lamp().get(),
"focuser": Focuser().get(),
"focuser-movement": FocuserMovement().get(),
"rotator": Rotator().get(),
"rotator-movement": RotatorMovement().get(),
"coordinates-movement": CoordinatesMovement().get(),
"coordinates-tracking": CoordinatesTracking().get(),
"clock": clock, # ast
"cover": Cover().get(), # ast
"offset": CoordinatesOffset().get(), # ast
"error": Error().get(), # ast
"error-details": ErrorDetails().get(), # ast
"coordinates": coor, # ast
}
else:
# 3) No telescope ASCOM connection
res = {
"power": power, # cab
"connection": connection,
#"clock": clock, # ast
"cover": Cover().get(), # ast
"offset": CoordinatesOffset().get(), # ast
"error": Error().get(), # ast
"coordinates": coor, # ast
}
else:
# 2) AstelOS not ready
res = {
"power": power, # cab
#"clock": clock, # ast
}
else:
# 1) Cabinet powered off
res = {
"power": power, # cab
}
self.last = res
return res
# connection = Connection(dev=self.dev).get()
# if connection["response"]:
# # 3) Telescope manually connected to ASCOM
# else:
# # 3) No telescope ASCOM connection
# else:
# # 2) AstelOS not ready
# return res
api/templates.py
View file @ b8fce05f
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''REST API to launch observation blocks'''
# System modules
# Third-party modules
from flask import Response, make_response, render_template, request
from flask_restx import Namespace, Resource, fields
import json
# Custom modules
from sequencer import Sequencer
from .data_access_object import ObservationBlockObject as DAO
from api.sequencer_instance import seq
from utils.data_access_object import ObservationBlockObject as DAO
dao = DAO("ob")
seq = Sequencer()
############################
# REST API
......@@ -68,27 +67,26 @@ class BobRun(Resource):
'''Constructor.'''
super().__init__(self)
self.seq = seq
self.last = None
def get(self):
"""Show the sequencer status"""
if self.seq.tpl:
res = {
"response": {
"name": self.seq.tpl.name,
"paused": self.seq.tpl.paused,
"quitting": self.seq.tpl.aborted,
"output": self.seq.tpl.output,
"error": self.seq.tpl.error,
"filename": self.seq.tpl.filename,
},
"error": self.seq.error,
}
res = {
"response": {
"name": self.seq.tpl.name,
"paused": self.seq.tpl.paused,
"quitting": self.seq.tpl.aborted,
"output": self.seq.tpl.output,
"error": self.seq.tpl.error,
"filename": self.seq.tpl.filename,
},
"error": self.seq.error,
}
else:
res = {
"response": {
"name": None,
......@@ -100,8 +98,7 @@ class BobRun(Resource):
},
"error": "No tpl object",
}
self.last = res
return res
def put(self):
......@@ -110,7 +107,7 @@ class BobRun(Resource):
PUT /sequencer/run true: resume,
PUT /sequencer/run false: pause
"""
in_execution = api.payload
self.seq.resume() if in_execution else self.seq.pause()
res = {
......@@ -163,23 +160,9 @@ class BobRun(Resource):
"""Stop the sequencer"""
self.seq.quit()
#res = "None"
if ob in self.seq and self.seq.ob:
if getattr(self.seq, "ob", False) and self.seq.ob:
res = {
"response": self.seq.ob if self.seq.ob else None,
"error": self.seq.error,
}
return res
############################
# WEB VIEW
############################
web = Namespace('sequencer', description='Web Sequencer interface')
@web.route("/")
class BobWeb(Resource):
def get(self):
data = dao.todos
return make_response(render_template("sequencer.html", data=data))
api/testfulvio.py
View file @ b8fce05f
......@@ -4,10 +4,8 @@
'''REST API test for Fulvio'''
# Third-party modules
from flask import Response, make_response, render_template, request
from flask_restx import Namespace, Resource, fields
from flask_restx import Namespace, Resource
# Custom modules
# import devices
############################
......@@ -16,7 +14,6 @@ from flask_restx import Namespace, Resource, fields
api = Namespace('fulvio', description='Fulvio test')
@api.route("/test")
class TestApi(Resource):
"""First get/post test of Fulvio."""
......@@ -36,27 +33,10 @@ class TestApi(Resource):
if target != "bbb":
err = "Non va bene, scrivi bbb"
else:
err = "nessun errore"
err = "nessun errore"
res = {
"response": target,
"error": err,
}
return res
############################
# WEB VIEW
############################
web = Namespace('fulvio', description='Web fulvio test pages')
@web.route("/test")
class TestWeb(Resource):
"""First get/post test web page of Fulvio."""
def get(self):
"""GET fulvio test web page."""
data = {}
return make_response(render_template("other/fulvio.html", data=data))