Sequencer almost complete. Missing real time stream (f3e09937) · Commits · Davide Ricci / noctua

noctua/api/init.py

+17 −16

Original line number	Diff line number	Diff line
		@@ -34,34 +34,35 @@ def dynamic_import(url_path):
		"""
		Import and register resources from api.ini with debug logging of supported methods.
		"""

		try:
		# Avoid overriding special namespaces already registered
		if url_path.startswith(("/blocks", "/sequencer", "/templates")):
		if url_path.startswith(("/blocks", "/sequencer")):
		return

		device_name = ends.get(url_path, "device")
		resource_name = ends.get(url_path, "resource")

		dev = getattr(devices, device_name)
		cls = dev.__class__.__name__
		mod_name = cls.lower()
		dev = getattr(devices, ends.get(url_path,"device")) # devices.light instance
		cls = dev.__class__.__name__ # string "Switch"
		mod_name = cls.lower() # string "switch"
		module = importlib.import_module(f"noctua.api.{mod_name}") # module noctua.api.switch
		resource_class = getattr(module, ends.get(url_path,"resource")) # class noctua.api.switch.State

		module = importlib.import_module(f"noctua.api.{mod_name}")
		resource_class = getattr(module, resource_name)
		# Identify supported HTTP methods by checking for lowercase method names in the class
		methods = [m for m in ["GET", "POST", "PUT", "DELETE"] if hasattr(resource_class, m.lower())]

		# Populate internal registry for dependency checking
		# Populate internal registry
		resource_registry[url_path] = resource_class(dev=dev)

		# Register the MethodView route
		# Register route
		view = resource_class.as_view(url_path, dev=dev)
		api_blueprint.add_url_rule(url_path, view_func=view)

		methods = [m for m in ["GET", "POST", "PUT", "DELETE"] if hasattr(resource_class, m.lower())]
		log.info(f"Route registered: /api{url_path:<40} [{str(methods):<25}]")
		# Improved debug print with methods list
		full_resource_path = f"{module.__name__}.{resource_class.__name__}"
		log.info(f"Route registered: /api{url_path:<40} {full_resource_path:<48} {str(methods):<25}")

		except Exception as e:
		log.warning(f"Error loading route {url_path}: {e}")
		log.warning(f"Error loading route: {url_path<40} {e}")

		# Load all routes from ini
		for section in ends.sections():
		if not section.startswith(("/blocks", "/sequencer")):
		dynamic_import(section)

noctua/api/blocks.py

+42 −31

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

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

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

		# Custom modules
		# Custom modules (Noctua Core)
		from noctua.utils.data_access_object import ObservationBlockObject

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


		class BlocksList(MethodView):
		"""Manage the collection of OB files."""

		@@ -21,6 +22,7 @@ class BlocksList(MethodView):
		res = await dao.list_available()
		return res


		class BlockFile(MethodView):
		"""Manage a specific OB file (Create, Append, Delete, Read All)."""

		@@ -30,24 +32,34 @@ class BlockFile(MethodView):
		content = await dao.read(name)
		return content if content is not None else ({"error": "OB not found"}, 404)

		async def post(self, name):
		"""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)
		async def post(self, name):
		"""
		Create or overwrite an OB.
		If payload is a string, load from defaults.
		If payload is an object/list, save it directly.
		"""

		data = await request.get_json(force=True, silent=True)

		# Logic to distinguish between "Create from Template" and "Save Editor Data"
		if isinstance(data, str):
		# It's a template name (e.g. "observation")
		tpl_content = await dao.read_default(data)
		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
		final_data = [tpl_content] if not isinstance(tpl_content, list) else tpl_content
		else:
		return {"error": f"Template {data} not found"}, 404
		elif isinstance(data, (dict, list)):
		# It's direct JSON data from the editor
		final_data = data
		else:
		# Fallback for empty or malformed payload
		final_data = []

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

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

		async def put(self, name):
		"""Append a template from defaults to the existing OB."""
		@@ -59,16 +71,17 @@ class BlockFile(MethodView):
		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
		except TypeError:
		return {"error": "Wrong data, expecting a template name"}, 500

		if tpl_content:
		content.append(tpl_content)
		success = await dao.write(name, content)
		await dao.write(name, content)
		return content

		return {"error": f"Template '{tpl_name}' not found in defaults"}, 404


		async def delete(self, name):
		"""Delete the whole OB file."""

		@@ -90,6 +103,7 @@ class BlockElement(MethodView):
		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."""

		@@ -102,6 +116,7 @@ class BlockElement(MethodView):
		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."""

		@@ -115,11 +130,7 @@ class BlockElement(MethodView):


		# --- 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'))
		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/templates/basetemplate.py

+51 −16

Original line number	Diff line number	Diff line
		@@ -4,10 +4,14 @@
		# System modules
		import sys
		from time import sleep
		import configparser
		from pathlib import Path

		# Custom templates
		from ..utils.logger import log

		# Other templates
		from ..config.constants import camera_state, filter_state, image_state
		from ..utils.logger import log


		class BaseTemplate:
		@@ -17,7 +21,7 @@ class BaseTemplate:
		def __init__(self):
		'''Contructor'''

		self.name = __name__
		self.name = self.__class__.__name__
		self.description = "Other templates inherit from this"
		self.error = [] # List of errors
		self.paused = False # pause flag
		@@ -25,7 +29,6 @@ class BaseTemplate:
		self.filename = "" # Current filename
		self.filenames = [] # List of saved filenames
		self.output = {} # Other output produced by template (txt?)
		self.used_devices = [] # List to store dev instances in the subclass

		# Exception decorator does NOT go here
		def content(self, params):
		@@ -40,6 +43,7 @@ class BaseTemplate:
		'''

		log.info(f"Running template {self.name}")
		log.info(f"Using following devices: {self.used_devices}")

		if not params:
		log.warning(f"Params are empty: {params}")
		@@ -57,6 +61,40 @@ class BaseTemplate:

		log.info(f"Template {self.name} ended")


		@property
		def used_devices(self):
		"""
		Identify which observatory devices are imported in the subclass.

		Returns
		-------
		list
		A list of device instances extracted from the subclass module.
		"""

		# 1. Get the list of official device names from the configuration
		config = configparser.ConfigParser()
		config_path = Path(__file__).parent.parent / "config" / "devices.ini"
		config.read(config_path)
		official_dev_names = config.sections()

		# 2. Access the module of the actual running subclass
		subclass_module = sys.modules[self.__class__.__module__]
		subclass_vars = vars(subclass_module)

		devices_found = []

		# 3. Intersection between .ini names and module imports
		for name in official_dev_names: # ["light", "lamp", ...]
		if name in subclass_vars: # Does the template have 'light' or 'lamp' imported?
		obj = subclass_vars[name] # Gets the specific instance from the module
		if obj not in devices_found:
		devices_found.append(obj)

		return devices_found


		def check_pause_or_abort(self):
		'''
		Pause/Resume/Stop the execution of the template.
		@@ -82,23 +120,20 @@ class BaseTemplate:

		def abort(self):
		"""
		Iterates through all registered devices and calls their abort
		method if it exists.
		Stop all hardware devices used by the current template.
		"""

		msg = f"Aborting template {self.name}. Cleaning up hardware..."
		msg = f"Aborting template {self.name}. Cleaning up used devices..."
		log.error(msg)
		self.error.append(msg)
		self.aborted = True

		# Automatic cleanup loop
		# Loop through detected devices and call abort() if present
		for dev in self.used_devices:
		# Use duck-typing to check if the device has an abort method
		abort_method = getattr(dev, "abort", None)

		if callable(abort_method):
		abort_call = getattr(dev, "abort", None)
		if callable(abort_call):
		try:
		log.warning(f"Calling abort() on device: {dev.__class__.__name__}")
		abort_method()
		log.warning(f"Aborting device: {dev.__class__.__name__}")
		abort_call()
		except Exception as e:
		log.error(f"Failed to abort device {dev}: {e}")
		log.error(f"Error during abort of {dev}: {e}")

noctua/templates/testlamp.py

+14 −4

Original line number	Diff line number	Diff line
		@@ -6,7 +6,7 @@ from time import sleep

		# Other templates
		from ..config.constants import on_off
		from ..devices import light
		from ..devices import light, lamp
		from ..utils.logger import log
		from .basetemplate import BaseTemplate

		@@ -22,9 +22,19 @@ class Template(BaseTemplate):
		light.state = True
		log.info(f"Switched {light.description} {on_off[light.state]}")

		sleep(3)
		sleep(2)

		# light.state = False
		# log.info(f"Switched {light.description} {on_off[light.state]}")
		lamp.state = True
		log.info(f"Switched {lamp.description} {on_off[lamp.state]}")

		sleep(2)

		light.state = False
		log.info(f"Switched {light.description} {on_off[light.state]}")

		sleep(2)

		lamp.state = False
		log.info(f"Switched {lamp.description} {on_off[lamp.state]}")

		return

noctua/web/init.py

+12 −0

Original line number	Diff line number	Diff line
		@@ -186,6 +186,18 @@ async def webcam():

		return await render_template('webcam.html')

		@web_blueprint.route('/sequencer')
		async def sequencer_page():
		"""
		Render the sequencer page.

		Returns
		-------
		str
		The rendered HTML content.
		"""

		return await render_template('sequencer.html')

		@web_blueprint.route('/subsystem')
		@web_blueprint.route('/subsystem/')