Started sequencer (9c850202) · Commits · Davide Ricci / software-di-controllo

noctua/api/init.py

+16 −19

Original line number	Diff line number	Diff line
		@@ -30,41 +30,38 @@ api_blueprint.register_blueprint(sequencer_api, url_prefix='/sequencer')
		# # Uncomment to test dependecy errors
		# resource_registry = {}

		api_blueprint = Blueprint('api', __name__)

		def dynamic_import(url_path):
		"""
		Import and register resources from api.ini with debug logging of supported methods.
		"""

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

		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
		device_name = ends.get(url_path, "device")
		resource_name = ends.get(url_path, "resource")

		# 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())]
		dev = getattr(devices, device_name)
		cls = dev.__class__.__name__
		mod_name = cls.lower()

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

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

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

		# 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}")
		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}]")

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

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

noctua/config/devices.ini

+0 −6

Original line number	Diff line number	Diff line
		@@ -139,12 +139,6 @@ node = IPCAM
		# class = Meteo
		# node = METEO

		# [sof]
		# module = domotics
		# class = Switch
		# node = DOMOTICS
		# outlet = 3

		# [fork]
		# module = domotics
		# class = Sensor

noctua/sequencer.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -181,7 +181,7 @@ class Sequencer():
		self.error.append(msg)

		except Exception as e: # Catch other unexpected errors during template execution
		msg = f"SEQUENCER: Unexpected error during execution of template '{template_name}'"
		msg = f"SEQUENCER: Unexpected error during execution of template '{template_name}': {e}"
		log.error(msg)
		self.error.append(msg)
		self.error.append(str(e))

noctua/templates/basetemplate.py

+21 −42

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

		# Other templates
		from ..config.constants import camera_state, filter_state, image_state
		from ..devices import cam, sof, tel
		from ..utils.logger import log


		@@ -26,6 +25,7 @@ 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):
		@@ -79,47 +79,26 @@ class BaseTemplate:

		return False


		def abort(self):
		'''
		Abort this template.
		'''
		"""
		Iterates through all registered devices and calls their abort
		method if it exists.
		"""

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

		try:
		if sof.state:
		# To be checked:
		# if the camera is off, here cam.state=[],
		# while in ipython cam.state=None
		if (cam.state != 0) or (cam.is_moving != 0) or (cam.ready != 1):
		log.warning(f"Camera is {camera_state[cam.state]}")
		log.warning(f"Filter is {filter_state[cam.is_moving]}")
		log.warning(f"Buffer is {image_state[cam.ready]}")
		log.error(f"Aborting camera.")
		cam.abort()

		while (cam.state != 0) or (cam.is_moving != 0):
		log.warning(
		f"Waiting, camera still {camera_state[cam.state]}")
		log.warning(
		f"Waiting, filter still {filter_state[cam.is_moving]}")
		log.warning(
		f"Waiting, buffer still {image_state[cam.ready]}")
		sleep(0.5)

		except Exception as e:
		msg = f"Interrupt triggered in {__name__}"
		log.error(msg)
		log.error(e)
		self.error.append(msg)
		self.error.append(e)
		# Automatic cleanup loop
		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):
		try:
		log.warning(f"Exiting {__name__}")
		self.aborted = True # This maybe work if called outside the sequencer
		# sys.exit(1) # This only works for sequencer
		log.warning(f"Calling abort() on device: {dev.__class__.__name__}")
		abort_method()
		except Exception as e:
		log.error(f"sys.exit() does not work here: {e}")
		raise e
		log.error(f"Failed to abort device {dev}: {e}")

noctua/templates/testsonoff.py

+52 −52

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

		# System modules
		import sys
		from time import sleep
		# # System modules
		# import sys
		# from time import sleep

		# Third-party modules
		from astropy.io import fits
		# # Third-party modules
		# from astropy.io import fits

		# Other templates
		from ..devices import sof
		from ..utils.logger import log
		from .basetemplate import BaseTemplate
		# # Other templates
		# from ..devices import sof
		# from ..utils.logger import log
		# from .basetemplate import BaseTemplate


		class Template(BaseTemplate):
		def __init__(self):
		super().__init__()
		self.name = "testsonoff"
		self.description = "Test sonoff"
		# class Template(BaseTemplate):
		# def __init__(self):
		# super().__init__()
		# self.name = "testsonoff"
		# self.description = "Test sonoff"

		def content(self, params):
		# def content(self, params):

		########################
		##### Params check #####
		########################
		try:
		getstatus = params.get("getstatus")
		setstatusto = params.get("setstatusto")
		except KeyError as e:
		log.error(f"Parameter {e} not found")
		# ########################
		# ##### Params check #####
		# ########################
		# try:
		# getstatus = params.get("getstatus")
		# setstatusto = params.get("setstatusto")
		# except KeyError as e:
		# log.error(f"Parameter {e} not found")

		########################
		##### GET #####
		########################
		# ########################
		# ##### GET #####
		# ########################

		if getstatus == 1:
		log.info(f"Sonoff is {sof.state}")
		# if getstatus == 1:
		# log.info(f"Sonoff is {sof.state}")

		########################
		##### SET #####
		########################
		# ########################
		# ##### SET #####
		# ########################

		if setstatusto is None:
		return
		# if setstatusto is None:
		# return

		if setstatusto == 0:
		sof.state = "Off"
		sleep(0.2)
		if (sof.state != "Off"):
		log.error(f"Failed to switch off the Sonoff!")
		else:
		log.info(f"Sonoff is now off")
		return
		# if setstatusto == 0:
		# sof.state = "Off"
		# sleep(0.2)
		# if (sof.state != "Off"):
		# log.error(f"Failed to switch off the Sonoff!")
		# else:
		# log.info(f"Sonoff is now off")
		# return

		if setstatusto == 1:
		sof.state = "On"
		sleep(2)
		if (sof.state != "On"):
		log.error(f"Failed to switch on the Sonoff!")
		else:
		log.info(f"Sonoff is now on")
		return
		# if setstatusto == 1:
		# sof.state = "On"
		# sleep(2)
		# if (sof.state != "On"):
		# log.error(f"Failed to switch on the Sonoff!")
		# else:
		# log.info(f"Sonoff is now on")
		# return

		return
		# return