Loading noctua/api/__init__.py +37 −21 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- '''Automatic REST APIs with Internal Resource Registry''' # System modules import configparser import importlib from pathlib import Path # Third-party modules from quart import Blueprint from quart import Blueprint, request, abort # Custom modules from noctua import devices from noctua.api.sequencer_instance import seq from noctua.api.basecontext import ends, resource_registry, config_path from .blocks import blocks_api from .defaults import defaults_api from .sequencer import sequencer_api Loading @@ -22,35 +26,47 @@ api_blueprint.register_blueprint(blocks_api, url_prefix='/blocks') api_blueprint.register_blueprint(defaults_api, url_prefix='/templates') api_blueprint.register_blueprint(sequencer_api, url_prefix='/sequencer') # Load api endpoint config file ends = configparser.ConfigParser() config_path = Path(__file__).parent.parent / "config" / "api.ini" ends.read(config_path) # This will store { "/telescope/power": <Instance of State>, ... } resource_registry = {} api_blueprint = Blueprint('api', __name__) def dynamic_import(url_path): """Dynamically import into this module api.ini files.""" """ Import and register resources from api.ini with debug logging of supported methods. """ try: # Skip manual routes if url_path.startswith(("/blocks", "/sequencer", "/templates")): if url_path.startswith(("/blocks", "/sequencer")): return res_class_name = ends.get(url_path, "resource") # CoolerWarmup dev_name = ends.get(url_path, "device") # cam dev_instance = getattr(devices, dev_name) # noctua.devices.stx.Camera object mod_name = dev_instance.__class__.__name__.lower() # camera 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}") # noctua.api.camera resource_class = getattr(module, res_class_name) # noctua.api.camera.CoolerWarmup module = importlib.import_module(f"noctua.api.{mod_name}") resource_class = getattr(module, res_class_name) # Register dynamic route as MethodView # This automatically maps GET/POST/PUT/DELETE to class methods # 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 resource_registry[url_path] = resource_class(dev=dev_instance) # Register route 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__}") # Improved debug print with methods list full_resource_path = f"{module.__name__}.{resource_class.__name__}" print(f"Route registered: /api{url_path:<40} {full_resource_path:<48} {str(methods):<25}") except Exception as e: print(f"Route skipped: /api{url_path:<40} -- {str(e):>5}") print(f"Error loading {url_path}: {e}") # Load all routes from ini for section in ends.sections(): if not section.startswith(("/blocks", "/sequencer")): dynamic_import(section) noctua/api/basecontext.py 0 → 100644 +8 −0 Original line number Diff line number Diff line import configparser from pathlib import Path ends = configparser.ConfigParser() config_path = Path(__file__).parent.parent / "config" / "api.ini" ends.read(config_path) resource_registry = {} noctua/api/baseresource.py +164 −30 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- '''Base class for REST API''' # System modules import asyncio from datetime import datetime # Third-party modules from quart import request, jsonify from quart import request, jsonify, current_app from quart.views import MethodView from .basecontext import ends, resource_registry class BaseResource(MethodView): """ Equivalent to flask_restx Resource. Methods get(), post(), put(), delete() will be called automatically. Base class for all API resources, providing dependency management and standardized responses. """ def __init__(self, dev): """ Initialize the resource with a specific device instance. Parameters ---------- dev : object The hardware device instance (Layer 1) or Sequencer (Layer 2). """ super().__init__() self.dev = dev @property def timestamp(self): """ Generate a ISO-formatted UTC timestamp. Returns ------- str The current timestamp string. """ return datetime.utcnow().isoformat() async def get_payload(self): """ Parse JSON payload. silent=True prevents Quart from returning a 400 error before execution. Implicitly parse JSON payload even if Content-Type header is missing. Returns ------- dict or list or None The parsed JSON data. """ return await request.get_json(force=True, silent=True) async def run_blocking(self, func, *args, **kwargs): """ Execute blocking L1/L2 hardware calls in a separate thread to keep the event loop responsive. Parameters ---------- func : callable The synchronous function to execute. *args : any Positional arguments for the function. **kwargs : any Keyword arguments for the function. Returns ------- any The result of the function call. """ return await asyncio.to_thread(func, *args, **kwargs) def make_response(self, response_data, raw_data=None, errors=None, status_code=200): """ Standardized JSON response format. If errors is provided as a list, it uses that. Otherwise, it pulls from self.dev.error. Parameters ---------- response_data : any The main payload to return. raw_data : any, optional The raw hardware value, if different from response_data. errors : list, optional A list of strings representing errors. status_code : int, optional The HTTP status code. Defaults to 200. Returns ------- tuple A tuple containing (JSON dictionary, status_code). """ # Logic: use manual errors if provided, otherwise check the device for errors dev_errors = getattr(self.dev, 'error', []) final_errors = errors if errors is not None else dev_errors # Force a non-200 status code if there are errors if final_errors and status_code == 200: status_code = 400 status_code = 500 res = { "raw": raw_data if raw_data is not None else response_data, Loading @@ -57,24 +106,109 @@ class BaseResource(MethodView): "timestamp": self.timestamp, } return jsonify(res), status_code return res, status_code async def _check_dependencies(self, endpoint_path): """ Recursively check if the dependency chain is active using the internal resource registry. Parameters ---------- endpoint_path : str The URL path of the current endpoint to check. Returns ------- tuple (bool: is_available, str: error_message) """ if request.args.get('force') == 'true': return True, None # from noctua.api import ends, resource_registry dependency = ends.get(endpoint_path, "depends-on", fallback=None) if dependency: # Look up the parent resource instance directly parent_resource = resource_registry.get(dependency) if not parent_resource: return False, f"Dependency configuration error: {dependency} not found in registry" # Call the parent's get() method directly (internal Python call) # This returns (dict, status_code) data, status_code = await parent_resource.get() if status_code != 200 or not data.get("raw") or data.get("error"): return False, f"Dependency failed: parent {dependency} is OFF or in error state" # Recurse return await self._check_dependencies(dependency) return True, None async def dispatch_request(self, *args, **kwargs): """ Override the default dispatch to enforce dependency checks and validate method existence before execution. Returns ------- Response The Quart response object or a JSON error. """ # Determine the logical path for dependency checking full_path = request.path api_path = full_path.replace("/api", "") # 1. Validate if the class actually implements the requested HTTP method # This prevents the "TypeError: first argument must be callable" handler = getattr(self, request.method.lower(), None) if handler is None: # This will trigger our global 405 JSON error handler from quart import abort abort(405) # 2. Validate dependencies (Topology check) is_available, err_msg = await self._check_dependencies(api_path) if not is_available: return jsonify({ "raw": None, "response": None, "error": [err_msg], "timestamp": self.timestamp, # "hint": "Use ?force=true to bypass this check" }), 424 # 3. Proceed to standard MethodView dispatching return await super().dispatch_request(*args, **kwargs) def register_error_handlers(app): now = datetime.utcnow().isoformat() """ Binds global HTTP errors to a standard JSON format. Parameters ---------- app : Quart The Quart application instance. """ @app.errorhandler(400) async def bad_request(e): return jsonify({"error": ["Bad Request"], "timestamp": now}), 400 return jsonify({"error": ["Bad Request"], "timestamp": datetime.utcnow().isoformat()}), 400 @app.errorhandler(404) async def not_found(e): return jsonify({"error": ["URL not found"], "timestamp": now}), 404 return jsonify({"error": ["URL not found"], "timestamp": datetime.utcnow().isoformat()}), 404 @app.errorhandler(405) async def method_not_allowed(e): return jsonify({"error": ["Method not allowed"], "timestamp": now}), 405 @app.errorhandler(500) async def internal_error(e): return jsonify({"error": [str(e)], "timestamp": now}), 500 async def not_found(e): return jsonify({"error": ["Method not allowed"], "timestamp": datetime.utcnow().isoformat()}), 404 @app.errorhandler(424) async def failed_dependency(e): return jsonify({"error": ["Failed Dependency"], "timestamp": datetime.utcnow().isoformat()}), 424 noctua/api/camera.py +13 −12 Original line number Diff line number Diff line Loading @@ -55,6 +55,19 @@ class CoolerTemperatureSetpoint(BaseResource): res = await self.run_blocking(action) return self.make_response(res) class CoolerWarmup(BaseResource): """Manage the warmup of the CCD.""" async def post(self): """Start the warm up the CCD.""" res = await self.run_blocking(lambda: self.dev.put('cooler', False)) return self.make_response(res) async def delete(self): """Stop the warm up of the CCD.""" return self.make_response("Warmup sequence aborted") class Filters(BaseResource): """Camera filters names.""" Loading Loading @@ -256,18 +269,6 @@ class SnapshotDomeslewing(BaseResource): seq.tpl.domeslewing = False return self.make_response(False) class CoolerWarmup(BaseResource): """Manage the warmup of the CCD.""" async def post(self): """Start the warm up the CCD.""" res = await self.run_blocking(lambda: self.dev.put('cooler', False)) return self.make_response(res) async def delete(self): """Stop the warm up of the CCD.""" return self.make_response("Warmup sequence aborted") class Connection(BaseResource): '''Manage the connection to ASCOM.''' Loading noctua/api/common.py +46 −0 Original line number Diff line number Diff line # In noctua/api/common.py (o direttamente nell'init delle API) from noctua.api import api_blueprint from noctua.api.__init__ import ends # Il configparser caricato from quart import request, jsonify import asyncio @api_blueprint.route('/all/<namespace>') async def get_all(namespace): """Aggregatore intelligente con gestione della cascata""" endpoints = {} # 1. Filtriamo le sezioni dell'ini per il namespace e ordiniamo per get-priority sections = [] for section in ends.sections(): if section.startswith("/" + namespace): priority = ends[section].getint('get-priority', 999) sections.append((priority, section)) sections.sort() # Ordina per priorità (1, 2, 3...) # 2. Scansione sequenziale for priority, section in sections: # Troviamo l'endpoint corrispondente nel server Quart # In Quart, possiamo simulare una chiamata interna per massima coerenza from quart import current_app client = current_app.test_client() response = await client.get(f"/api{section}") data = await response.get_json() name = section.split("/")[-1] # es. "power" o "light" endpoints[name] = data # 3. LOGICA A CASCATA: # Se un elemento con priorità ha un 'raw' False (o errore), # interrompiamo la scansione per questo namespace if priority < 999: # Solo per elementi critici (power, connection) if data.get("raw") is False or data.get("error") is not None: # Marchiamo i successivi come "unavailable" break return jsonify(endpoints) import asyncio from datetime import datetime from dataclasses import dataclass Loading Loading
noctua/api/__init__.py +37 −21 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- '''Automatic REST APIs with Internal Resource Registry''' # System modules import configparser import importlib from pathlib import Path # Third-party modules from quart import Blueprint from quart import Blueprint, request, abort # Custom modules from noctua import devices from noctua.api.sequencer_instance import seq from noctua.api.basecontext import ends, resource_registry, config_path from .blocks import blocks_api from .defaults import defaults_api from .sequencer import sequencer_api Loading @@ -22,35 +26,47 @@ api_blueprint.register_blueprint(blocks_api, url_prefix='/blocks') api_blueprint.register_blueprint(defaults_api, url_prefix='/templates') api_blueprint.register_blueprint(sequencer_api, url_prefix='/sequencer') # Load api endpoint config file ends = configparser.ConfigParser() config_path = Path(__file__).parent.parent / "config" / "api.ini" ends.read(config_path) # This will store { "/telescope/power": <Instance of State>, ... } resource_registry = {} api_blueprint = Blueprint('api', __name__) def dynamic_import(url_path): """Dynamically import into this module api.ini files.""" """ Import and register resources from api.ini with debug logging of supported methods. """ try: # Skip manual routes if url_path.startswith(("/blocks", "/sequencer", "/templates")): if url_path.startswith(("/blocks", "/sequencer")): return res_class_name = ends.get(url_path, "resource") # CoolerWarmup dev_name = ends.get(url_path, "device") # cam dev_instance = getattr(devices, dev_name) # noctua.devices.stx.Camera object mod_name = dev_instance.__class__.__name__.lower() # camera 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}") # noctua.api.camera resource_class = getattr(module, res_class_name) # noctua.api.camera.CoolerWarmup module = importlib.import_module(f"noctua.api.{mod_name}") resource_class = getattr(module, res_class_name) # Register dynamic route as MethodView # This automatically maps GET/POST/PUT/DELETE to class methods # 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 resource_registry[url_path] = resource_class(dev=dev_instance) # Register route 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__}") # Improved debug print with methods list full_resource_path = f"{module.__name__}.{resource_class.__name__}" print(f"Route registered: /api{url_path:<40} {full_resource_path:<48} {str(methods):<25}") except Exception as e: print(f"Route skipped: /api{url_path:<40} -- {str(e):>5}") print(f"Error loading {url_path}: {e}") # Load all routes from ini for section in ends.sections(): if not section.startswith(("/blocks", "/sequencer")): dynamic_import(section)
noctua/api/basecontext.py 0 → 100644 +8 −0 Original line number Diff line number Diff line import configparser from pathlib import Path ends = configparser.ConfigParser() config_path = Path(__file__).parent.parent / "config" / "api.ini" ends.read(config_path) resource_registry = {}
noctua/api/baseresource.py +164 −30 Original line number Diff line number Diff line #!/usr/bin/env python3 # -*- coding: utf-8 -*- '''Base class for REST API''' # System modules import asyncio from datetime import datetime # Third-party modules from quart import request, jsonify from quart import request, jsonify, current_app from quart.views import MethodView from .basecontext import ends, resource_registry class BaseResource(MethodView): """ Equivalent to flask_restx Resource. Methods get(), post(), put(), delete() will be called automatically. Base class for all API resources, providing dependency management and standardized responses. """ def __init__(self, dev): """ Initialize the resource with a specific device instance. Parameters ---------- dev : object The hardware device instance (Layer 1) or Sequencer (Layer 2). """ super().__init__() self.dev = dev @property def timestamp(self): """ Generate a ISO-formatted UTC timestamp. Returns ------- str The current timestamp string. """ return datetime.utcnow().isoformat() async def get_payload(self): """ Parse JSON payload. silent=True prevents Quart from returning a 400 error before execution. Implicitly parse JSON payload even if Content-Type header is missing. Returns ------- dict or list or None The parsed JSON data. """ return await request.get_json(force=True, silent=True) async def run_blocking(self, func, *args, **kwargs): """ Execute blocking L1/L2 hardware calls in a separate thread to keep the event loop responsive. Parameters ---------- func : callable The synchronous function to execute. *args : any Positional arguments for the function. **kwargs : any Keyword arguments for the function. Returns ------- any The result of the function call. """ return await asyncio.to_thread(func, *args, **kwargs) def make_response(self, response_data, raw_data=None, errors=None, status_code=200): """ Standardized JSON response format. If errors is provided as a list, it uses that. Otherwise, it pulls from self.dev.error. Parameters ---------- response_data : any The main payload to return. raw_data : any, optional The raw hardware value, if different from response_data. errors : list, optional A list of strings representing errors. status_code : int, optional The HTTP status code. Defaults to 200. Returns ------- tuple A tuple containing (JSON dictionary, status_code). """ # Logic: use manual errors if provided, otherwise check the device for errors dev_errors = getattr(self.dev, 'error', []) final_errors = errors if errors is not None else dev_errors # Force a non-200 status code if there are errors if final_errors and status_code == 200: status_code = 400 status_code = 500 res = { "raw": raw_data if raw_data is not None else response_data, Loading @@ -57,24 +106,109 @@ class BaseResource(MethodView): "timestamp": self.timestamp, } return jsonify(res), status_code return res, status_code async def _check_dependencies(self, endpoint_path): """ Recursively check if the dependency chain is active using the internal resource registry. Parameters ---------- endpoint_path : str The URL path of the current endpoint to check. Returns ------- tuple (bool: is_available, str: error_message) """ if request.args.get('force') == 'true': return True, None # from noctua.api import ends, resource_registry dependency = ends.get(endpoint_path, "depends-on", fallback=None) if dependency: # Look up the parent resource instance directly parent_resource = resource_registry.get(dependency) if not parent_resource: return False, f"Dependency configuration error: {dependency} not found in registry" # Call the parent's get() method directly (internal Python call) # This returns (dict, status_code) data, status_code = await parent_resource.get() if status_code != 200 or not data.get("raw") or data.get("error"): return False, f"Dependency failed: parent {dependency} is OFF or in error state" # Recurse return await self._check_dependencies(dependency) return True, None async def dispatch_request(self, *args, **kwargs): """ Override the default dispatch to enforce dependency checks and validate method existence before execution. Returns ------- Response The Quart response object or a JSON error. """ # Determine the logical path for dependency checking full_path = request.path api_path = full_path.replace("/api", "") # 1. Validate if the class actually implements the requested HTTP method # This prevents the "TypeError: first argument must be callable" handler = getattr(self, request.method.lower(), None) if handler is None: # This will trigger our global 405 JSON error handler from quart import abort abort(405) # 2. Validate dependencies (Topology check) is_available, err_msg = await self._check_dependencies(api_path) if not is_available: return jsonify({ "raw": None, "response": None, "error": [err_msg], "timestamp": self.timestamp, # "hint": "Use ?force=true to bypass this check" }), 424 # 3. Proceed to standard MethodView dispatching return await super().dispatch_request(*args, **kwargs) def register_error_handlers(app): now = datetime.utcnow().isoformat() """ Binds global HTTP errors to a standard JSON format. Parameters ---------- app : Quart The Quart application instance. """ @app.errorhandler(400) async def bad_request(e): return jsonify({"error": ["Bad Request"], "timestamp": now}), 400 return jsonify({"error": ["Bad Request"], "timestamp": datetime.utcnow().isoformat()}), 400 @app.errorhandler(404) async def not_found(e): return jsonify({"error": ["URL not found"], "timestamp": now}), 404 return jsonify({"error": ["URL not found"], "timestamp": datetime.utcnow().isoformat()}), 404 @app.errorhandler(405) async def method_not_allowed(e): return jsonify({"error": ["Method not allowed"], "timestamp": now}), 405 @app.errorhandler(500) async def internal_error(e): return jsonify({"error": [str(e)], "timestamp": now}), 500 async def not_found(e): return jsonify({"error": ["Method not allowed"], "timestamp": datetime.utcnow().isoformat()}), 404 @app.errorhandler(424) async def failed_dependency(e): return jsonify({"error": ["Failed Dependency"], "timestamp": datetime.utcnow().isoformat()}), 424
noctua/api/camera.py +13 −12 Original line number Diff line number Diff line Loading @@ -55,6 +55,19 @@ class CoolerTemperatureSetpoint(BaseResource): res = await self.run_blocking(action) return self.make_response(res) class CoolerWarmup(BaseResource): """Manage the warmup of the CCD.""" async def post(self): """Start the warm up the CCD.""" res = await self.run_blocking(lambda: self.dev.put('cooler', False)) return self.make_response(res) async def delete(self): """Stop the warm up of the CCD.""" return self.make_response("Warmup sequence aborted") class Filters(BaseResource): """Camera filters names.""" Loading Loading @@ -256,18 +269,6 @@ class SnapshotDomeslewing(BaseResource): seq.tpl.domeslewing = False return self.make_response(False) class CoolerWarmup(BaseResource): """Manage the warmup of the CCD.""" async def post(self): """Start the warm up the CCD.""" res = await self.run_blocking(lambda: self.dev.put('cooler', False)) return self.make_response(res) async def delete(self): """Stop the warm up of the CCD.""" return self.make_response("Warmup sequence aborted") class Connection(BaseResource): '''Manage the connection to ASCOM.''' Loading
noctua/api/common.py +46 −0 Original line number Diff line number Diff line # In noctua/api/common.py (o direttamente nell'init delle API) from noctua.api import api_blueprint from noctua.api.__init__ import ends # Il configparser caricato from quart import request, jsonify import asyncio @api_blueprint.route('/all/<namespace>') async def get_all(namespace): """Aggregatore intelligente con gestione della cascata""" endpoints = {} # 1. Filtriamo le sezioni dell'ini per il namespace e ordiniamo per get-priority sections = [] for section in ends.sections(): if section.startswith("/" + namespace): priority = ends[section].getint('get-priority', 999) sections.append((priority, section)) sections.sort() # Ordina per priorità (1, 2, 3...) # 2. Scansione sequenziale for priority, section in sections: # Troviamo l'endpoint corrispondente nel server Quart # In Quart, possiamo simulare una chiamata interna per massima coerenza from quart import current_app client = current_app.test_client() response = await client.get(f"/api{section}") data = await response.get_json() name = section.split("/")[-1] # es. "power" o "light" endpoints[name] = data # 3. LOGICA A CASCATA: # Se un elemento con priorità ha un 'raw' False (o errore), # interrompiamo la scansione per questo namespace if priority < 999: # Solo per elementi critici (power, connection) if data.get("raw") is False or data.get("error") is not None: # Marchiamo i successivi come "unavailable" break return jsonify(endpoints) import asyncio from datetime import datetime from dataclasses import dataclass Loading
