Now /api/blocks/ work, with additional features with respect to the original... (b92645d1) · Commits · Davide Ricci / software-di-controllo

noctua/api/init.py

+2 −49

Original line number	Diff line number	Diff line
		@@ -9,12 +9,14 @@ from quart import Blueprint
		# Custom modules
		from noctua import devices
		from .blocks import blocks_api
		from .defaults import defaults_api
		from .templates import templates_api

		api_blueprint = Blueprint('api', __name__)

		# Register Manual Blueprints (Namespaces)
		api_blueprint.register_blueprint(blocks_api, url_prefix='/blocks')
		api_blueprint.register_blueprint(defaults_api, url_prefix='/templates')
		api_blueprint.register_blueprint(templates_api, url_prefix='/sequencer')

		# Load api.ini
		@@ -50,52 +52,3 @@ def dynamic_import(url_path):

		for section in ends.sections():
		dynamic_import(section)

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

		# '''Automatic REST APIs using Quart Blueprints'''
		# import configparser
		# import importlib
		# from pathlib import Path
		# from quart import Blueprint
		# from noctua import devices
		# from .baseresource import BaseResource

		# # 1. Importiamo i blueprint "complessi" definiti nei file
		# from .blocks import blocks_api
		# from .templates import templates_api

		# api_blueprint = Blueprint('api', __name__)

		# # 2. Caricamento dinamico per api.ini (Camera, Telescope, Switch, etc.)
		# ends = configparser.ConfigParser()
		# ends.read(Path(__file__).parent.parent / "config" / "api.ini")

		# def dynamic_import(url_path):
		# try:
		# # Se la rotta è già gestita dai blueprint manuali, saltala
		# if url_path.startswith(('/blocks', '/sequencer')): return

		# res_class_name = ends.get(url_path, "resource")
		# dev_name = ends.get(url_path, "device")
		# dev_instance = getattr(devices, dev_name)
		# mod_name = dev_instance.__class__.__name__.lower()

		# module = importlib.import_module(f"noctua.api.{mod_name}")
		# resource_class = getattr(module, res_class_name)

		# # Registrazione dinamica come MethodView
		# view = resource_class.as_view(url_path, dev=dev_instance)
		# api_blueprint.add_url_rule(url_path, view_func=view)
		# print(f"Route registered: /api{url_path:<40} {module.__name__:>5}.{resource_class.__name__}")

		# except Exception as e:
		# print(f"Route skipped: /api{url_path:<40} -- {str(e):>5}")

		# for section in ends.sections():
		# dynamic_import(section)

		# # 3. Registrazione dei Blueprint manuali
		# api_blueprint.register_blueprint(blocks_api, url_prefix='/blocks')
		# api_blueprint.register_blueprint(templates_api, url_prefix='/sequencer')

noctua/api/blocks.py

+94 −71

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --

		'''REST API to edit Observation Blocks'''
		'''REST API to manage Observation Block JSON files using the DAO'''

		# Third-party modules
		from quart import Blueprint
		from quart import Blueprint, request
		from quart.views import MethodView

		# Custom modules
		from noctua.utils.data_access_object import ObservationBlockObject as DAO
		from noctua.utils.data_access_object import guess
		from .baseresource import BaseResource
		from noctua.utils.data_access_object import ObservationBlockObject

		blocks_api = Blueprint('blocks', __name__)
		dao = ObservationBlockObject()

		dao = DAO("ob")
		tpl = DAO("defaults")

		class BlockList(BaseResource):
		"""Show and create OBs"""
		class BlocksList(MethodView):
		"""Manage the collection of OB files."""

		async def get(self):
		"""Show all observation OB files"""
		res = await self.run_blocking(lambda: dao.todos)
		return self.make_response(res)

		async def post(self):
		"""Create a new OB file based on name"""
		name = await self.get_payload()
		res = await self.run_blocking(dao.create, name)
		return self.make_response(res, status_code=204)

		async def delete(self):
		"""Delete an OB file based on name"""
		name = await self.get_payload()
		res = await self.run_blocking(dao.delete, name)
		return self.make_response(res, status_code=204)
		"""List all OB files available in the OB folder."""
		res = await dao.list_available()
		return res


		class Block(BaseResource):
		"""Show a selected OB, update it, or add a new
		template in it.
		"""
		class BlockFile(MethodView):
		"""Manage a specific OB file (Create, Append, Delete, Read All)."""

		async def get(self, name):
		"""Show a specific OB"""
		res = await self.run_blocking(dao.show, name)
		return self.make_response(res)

		async def put(self, name):
		"""Update the OB"""
		data = await self.get_payload()

		def logic():
		for datum in data:
		for key, val in datum["params"].items():
		datum["params"][key] = guess(val)
		return dao.update(name, data)

		res = await self.run_blocking(logic)
		return self.make_response(res, status_code=204)
		"""Show the whole OB content."""
		content = await dao.read(name)
		return content if content is not None else ({"error": "OB not found"}, 404)

		async def post(self, name):
		"""Add a template to the selected OB"""
		payload = await self.get_payload()

		def logic():
		content = dao.content(name)
		new_data = tpl.content(payload)
		data = content + new_data
		return dao.update(name, data)
		"""Create a new OB. If payload is a valid default name, use its content."""
		data = []
		try:
		tpl_name = await request.get_json(force=True, silent=True)
		if tpl_name:
		# Recycled DAO method
		tpl_content = await dao.read_default(tpl_name)
		if tpl_content:
		# Ensure it's a list for the OB file
		data = [tpl_content] if not isinstance(tpl_content, list) else tpl_content
		except Exception:
		pass

		await dao.write(name, data)
		return data, 201

		res = await self.run_blocking(logic)
		return self.make_response(res, status_code=204)
		async def put(self, name):
		"""Append a template from defaults to the existing OB."""
		content = await dao.read(name)
		if content is None:
		return {"error": "OB not found"}, 404

		async def delete(self, name):
		"""Delete a template instance in the selected OB"""
		instance = await self.get_payload()
		try:
		tpl_name = await request.get_json(force=True)
		tpl_content = await dao.load_default(tpl_name)
		except TypeError as e:
		return {"error": f"Wrong data, expecting a template name"}, 500

		def logic():
		data = dao.content(name)
		data.pop(instance - 1) # jinjia loop starts from 1 in html
		return dao.update(name, data)
		if tpl_content:
		content.append(tpl_content)
		success = await dao.write(name, content)
		return content

		res = await self.run_blocking(logic)
		return self.make_response(res, status_code=204)
		return {"error": f"Template '{tpl_name}' not found in defaults"}, 404

		# Association of classes to endpoints (internal routing)
		blocks_api.add_url_rule('/', view_func=BlockList.as_view('block_list', dev=dao))
		blocks_api.add_url_rule('/<name>', view_func=Block.as_view('block_detail', dev=dao))
		async def delete(self, name):
		"""Delete the whole OB file."""
		success = await dao.delete(name)
		if success:
		return {"message": f"OB {name} deleted"}, 200
		return {"error": "OB not found"}, 404


		class BlockElement(MethodView):
		"""Manage specific templates inside an OB using 1-based indexing."""

		async def get(self, name, index):
		"""Show a specific template inside the OB."""
		content = await dao.read(name)
		try:
		return content[index - 1]
		except (IndexError, TypeError, KeyError):
		return {"error": "Index out of range or OB not found"}, 404

		async def put(self, name, index):
		"""Update a specific template inside the OB."""
		new_data = await request.get_json(force=True)
		content = await dao.read(name)
		try:
		content[index - 1] = new_data
		await dao.write(name, content)
		return content
		except (IndexError, TypeError, KeyError):
		return {"error": "Index out of range or OB not found"}, 404

		async def delete(self, name, index):
		"""Delete a specific template inside the OB."""
		content = await dao.read(name)
		try:
		content.pop(index - 1)
		await dao.write(name, content)
		return content
		except (IndexError, TypeError, KeyError):
		return {"error": "Index out of range or OB not found"}, 404


		# --- ROUTING ---
		blocks_api.add_url_rule('/', view_func=BlocksList.as_view('blocks_list'))
		blocks_api.add_url_rule('/<name>', view_func=BlockFile.as_view('block_file'))
		blocks_api.add_url_rule('/<name>/', view_func=BlockFile.as_view('block_file_read'))
		blocks_api.add_url_rule('/<name>/<int:index>', view_func=BlockElement.as_view('block_element'))

noctua/api/defaults.py

0 → 100644

+37 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --

		'''REST API to manage Template Defaults'''

		# Third-party modules
		from quart import Blueprint
		from quart.views import MethodView

		# Custom modules
		from noctua.utils.data_access_object import ObservationBlockObject

		# We name the blueprint 'defaults', but we will map it to '/templates' in __init__.py
		defaults_api = Blueprint('defaults', __name__)
		dao = ObservationBlockObject()

		class TemplateList(MethodView):
		"""List all available template names."""

		async def get(self):
		"""GET /api/templates/"""
		res = await dao.list_defaults()
		return res

		class TemplateContent(MethodView):
		"""Show the content of a specific default template."""

		async def get(self, name):
		"""GET /api/templates/<name>"""
		content = await dao.read_default(name)
		if content is not None:
		return content
		return {"error": f"Template '{name}' not found"}, 404

		# --- ROUTING ---
		defaults_api.add_url_rule('/', view_func=TemplateList.as_view('template_list'))
		defaults_api.add_url_rule('/<name>', view_func=TemplateContent.as_view('template_content'))

noctua/api/templates.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@ from .baseresource import BaseResource

		templates_api = Blueprint('sequencer', __name__)

		dao = DAO("ob")
		dao = DAO()

		class BobList(BaseResource):
		"""Show and modify a list of OBs to be sequenced"""

noctua/config/constants.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -117,6 +117,8 @@ DATA_FOLDER = "data"
		LOG_FOLDER = "log"
		FITS_FOLDER = "fits"
		FOCUS_FOLDER = "focus"
		OB_FOLDER = "ob"
		DEFAULTS_FOLDER = "defaults"

		dir_type = {
		0: "dark",