Check dependencies fixed by Claude. Ansi Colors on logs (da46e648) · Commits · Davide Ricci / software-di-controllo

noctua/api/baseresource.py

+27 −73

Original line number	Diff line number	Diff line
		@@ -3,9 +3,12 @@

		import asyncio
		from datetime import datetime
		from quart import request, jsonify, current_app
		from quart import request, jsonify
		from quart.views import MethodView
		from .basecontext import ends, resource_registry

		# Custom modules
		from .deps import check_dependencies


		class BaseResource(MethodView):
		"""
		@@ -28,14 +31,13 @@ class BaseResource(MethodView):
		@property
		def timestamp(self):
		"""
		Generate a ISO-formatted UTC timestamp.
		Generate an ISO-formatted UTC timestamp.

		Returns
		-------
		str
		The current timestamp string.
		"""

		return datetime.utcnow().isoformat()

		async def get_payload(self):
		@@ -47,7 +49,6 @@ class BaseResource(MethodView):
		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):
		@@ -69,12 +70,11 @@ class BaseResource(MethodView):
		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.
		Build a standardized JSON response dict.

		Parameters
		----------
		@@ -90,9 +90,8 @@ class BaseResource(MethodView):
		Returns
		-------
		tuple
		A tuple containing (JSON dictionary, status_code).
		A tuple of (response dict, status_code).
		"""

		dev_errors = getattr(self.dev, 'error', [])
		final_errors = errors if errors is not None else dev_errors

		@@ -108,73 +107,28 @@ class BaseResource(MethodView):

		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.
		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", "")
		api_path = request.path.replace("/api", "", 1)

		# 1. Validate if the class actually implements the requested HTTP method
		# This prevents the "TypeError: first argument must be callable"
		# 1. Validate that the class implements the requested HTTP method
		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)
		# 2. Validate the dependency chain (reads ?force=true from query string)
		force = request.args.get('force') == 'true'
		is_available, err_msg = await check_dependencies(api_path, force)

		if not is_available:
		return jsonify({
		@@ -182,15 +136,15 @@ class BaseResource(MethodView):
		"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):
		"""
		Binds global HTTP errors to a standard JSON format.
		Bind global HTTP errors to a standard JSON format.

		Parameters
		----------
		@@ -206,8 +160,8 @@ def register_error_handlers(app):
		return jsonify({"error": ["URL not found"], "timestamp": datetime.utcnow().isoformat()}), 404

		@app.errorhandler(405)
		async def not_found(e):
		return jsonify({"error": ["Method not allowed"], "timestamp": datetime.utcnow().isoformat()}), 404
		async def method_not_allowed(e):
		return jsonify({"error": ["Method not allowed"], "timestamp": datetime.utcnow().isoformat()}), 405

		@app.errorhandler(424)
		async def failed_dependency(e):

noctua/api/deps.py

0 → 100644

+52 −0

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

		'''Shared dependency resolution for API resources and the stream layer.'''

		# Custom modules
		from noctua.api.basecontext import ends, resource_registry


		async def check_dependencies(path: str, force: bool = False) -> tuple[bool, str \| None]:
		"""
		Recursively verify the dependency chain for a given endpoint path.

		Walks the ``depends-on`` chain declared in ``api.ini`` and calls each
		parent resource's ``get()`` method directly, so the check always reflects
		the live hardware state rather than any cached value.

		Parameters
		----------
		path : str
		The API endpoint path to check (e.g. ``/camera/snapshot``).
		force : bool, optional
		When ``True`` the entire dependency chain is bypassed and the function
		returns ``(True, None)`` immediately. Defaults to ``False``.

		Returns
		-------
		tuple[bool, str \| None]
		``(True, None)`` if the chain is satisfied, or
		``(False, error_message)`` at the first failing link.
		"""

		if force:
		return True, None

		dependency = ends.get(path, "depends-on", fallback=None)

		if not dependency:
		return True, None

		parent_resource = resource_registry.get(dependency)

		if not parent_resource:
		return False, f"Dependency configuration error: {dependency} not found in registry"

		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 up the chain
		return await check_dependencies(dependency, force)

noctua/api/stage.py

+32 −33

Original line number	Diff line number	Diff line
		@@ -6,7 +6,22 @@
		# Custom modules
		from .baseresource import BaseResource

		class Position(BaseResource):

		class BaseStageResource(BaseResource):
		"""
		Shared base for all Stage resources.

		Provides the ``delete`` handler that aborts the current stage movement,
		which is identical across every Stage endpoint.
		"""

		async def delete(self):
		"""Abort the current stage movement."""
		res = await self.run_blocking(lambda: self.dev.abort)
		return self.make_response(res)


		class Position(BaseStageResource):
		"""Manage the absolute position of the stage in mm."""

		async def get(self):
		@@ -23,13 +38,9 @@ class Position(BaseResource):
		res = await self.run_blocking(action)
		return self.make_response(res)

		async def delete(self):
		"""Check if the stage is currently moving."""
		res = await self.run_blocking(lambda: self.dev.abort)
		return self.make_response(res)

		class Named(BaseResource):
		"""Manage named positions (e.g., 'imaging', 'spectroscopy')."""
		class Named(BaseStageResource):
		"""Manage named positions (e.g. 'imaging', 'spectroscopy')."""

		async def get(self):
		"""Get the name of the current position if matched."""
		@@ -40,14 +51,17 @@ class Named(BaseResource):
		"""Move the stage to a named position."""
		target_name = await self.get_payload() # expected string
		named_positions = self.dev.named_positions

		def action():
		if target_name in self.dev.named_positions:
		self.dev.named = target_name
		return self.dev.named
		else:
		return self.make_response(None,
		return self.make_response(
		None,
		errors=[f"Not in: {named_positions}"],
		status_code=500)
		status_code=500,
		)

		try:
		res = await self.run_blocking(action)
		@@ -55,14 +69,9 @@ class Named(BaseResource):
		except KeyError as e:
		return self.make_response(None, errors=[str(e)], status_code=400)

		async def delete(self):
		"""Check if the stage is currently moving."""
		res = await self.run_blocking(lambda: self.dev.abort)
		return self.make_response(res)


		class Initialization(BaseResource):
		"""Handle stage homing and status."""
		class Initialization(BaseStageResource):
		"""Handle stage homing and initialization status."""

		async def get(self):
		"""Check if the stage is initialized and referenced."""
		@@ -75,13 +84,8 @@ class Initialization(BaseResource):
		res = await self.run_blocking(lambda: self.dev.is_init)
		return self.make_response(res)

		async def delete(self):
		"""Check if the stage is currently moving."""
		res = await self.run_blocking(lambda: self.dev.abort)
		return self.make_response(res)


		class Movement(BaseResource):
		class Movement(BaseStageResource):
		"""Monitor stage motion status."""

		async def get(self):
		@@ -89,13 +93,8 @@ class Movement(BaseResource):
		res = await self.run_blocking(lambda: self.dev.is_moving)
		return self.make_response(res)

		async def delete(self):
		"""Check if the stage is currently moving."""
		res = await self.run_blocking(lambda: self.dev.abort)
		return self.make_response(res)


		class Limits(BaseResource):
		class Limits(BaseStageResource):
		"""Manage software movement limits."""

		async def get(self):

noctua/config/api.ini

+3 −3

Original line number	Diff line number	Diff line
		@@ -169,9 +169,9 @@ resource = CoordinatesTracking
		device = tel
		depends-on = /telescope/clock

		[/telescope/connection]
		resource = Connection
		device = tel
		# [/telescope/connection]
		# resource = Connection
		# device = tel

		[/telescope/error]
		resource = Error

noctua/web/pages/base.html

+87 −16

Original line number	Diff line number	Diff line
		@@ -2,38 +2,109 @@
		<html lang="it" data-bs-theme="dark">
		<head>
		<meta charset="UTF-8">
		<meta name="viewport" content="width=device-width, initial-scale=1">
		<title>Noctua Control</title>

		<!-- Favicon -->
		<link rel="icon" type="image/x-icon"
		href="{{ url_for('web.static', filename='img/favicon.ico') }}">
		<link rel="icon" type="image/png" sizes="192x192"
		href="{{ url_for('web.static', filename='img/favicon-192.png') }}">
		<link rel="apple-touch-icon" sizes="192x192"
		href="{{ url_for('web.static', filename='img/favicon-192.png') }}">

		<link rel="stylesheet" href="{{ url_for('web.static', filename='css/vendor/bootstrap.min.css') }}">
		<link rel="stylesheet" href="{{ url_for('web.static', filename='css/style.css') }}">
		</head>
		<body class="d-flex flex-column h-100" style="padding-bottom: 180px;">

		<div class="container">
		<body class="d-flex flex-column" style="padding-bottom: 180px;">

		<!-- ── Navbar ─────────────────────────────────────────────────────────── -->
		<nav class="navbar navbar-expand-sm bg-dark border-bottom border-secondary mb-4 px-3" style="padding-top:6px;padding-bottom:6px;">

		<!-- Logo + brand -->
		<a class="navbar-brand d-flex align-items-center gap-2 p-0" href="{{ url_for('web.init_panel') }}">
		<img src="{{ url_for('web.static', filename='img/noctua.svg') }}"
		width="40" height="40"
		alt="Noctua logo"
		style="border-radius:6px;">
		<span class="fw-semibold" style="letter-spacing:0.05em;">NOCTUA</span>
		</a>

		<!-- Hamburger for mobile -->
		<button class="navbar-toggler" type="button"
		data-bs-toggle="collapse" data-bs-target="#mainNav"
		aria-controls="mainNav" aria-expanded="false" aria-label="Toggle navigation">
		<span class="navbar-toggler-icon"></span>
		</button>

		<!-- Nav links -->
		<div class="collapse navbar-collapse" id="mainNav">
		<ul class="navbar-nav ms-3 gap-1">
		<li class="nav-item">
		<a class="nav-link {% if request.endpoint == 'web.init_panel' %}active{% endif %}"
		href="{{ url_for('web.init_panel') }}">
		Init
		</a>
		</li>
		<li class="nav-item">
		<a class="nav-link {% if request.endpoint == 'web.status_monitor' %}active{% endif %}"
		href="{{ url_for('web.status_monitor') }}">
		Status
		</a>
		</li>
		<li class="nav-item">
		<a class="nav-link {% if request.endpoint == 'web.subsystems' or request.endpoint == 'web.subsystem_page' %}active{% endif %}"
		href="{{ url_for('web.subsystems') }}">
		Subsystems
		</a>
		</li>
		</ul>

		<!-- Logo grande, opzionale, visibile solo su schermi larghi -->
		<div class="ms-auto d-none d-md-block" style="opacity:0.08; pointer-events:none;">
		<img src="{{ url_for('web.static', filename='img/noctua.svg') }}"
		width="100" height="100" alt="">
		</div>
		</div>
		</nav>

		<!-- ── Page content ───────────────────────────────────────────────────── -->
		<div class="container-fluid px-4">
		{% block content %}{% endblock %}
		</div>

		<!-- Footer Notification Area -->
		<div class="toast-container position-fixed bottom-0 end-0 p-3" id="notification-container" style="z-index: 1100; margin-bottom: 160px;"></div>
		<!-- ── Toast notifications ────────────────────────────────────────────── -->
		<div class="toast-container position-fixed bottom-0 end-0 p-3"
		id="notification-container"
		style="z-index:1100; margin-bottom:160px;"></div>

		<!-- Unified Sticky Footer Log Terminal -->
		<footer class="footer fixed-bottom bg-black border-top border-secondary shadow-lg" style="height: 160px;">
		<!-- ── Sticky footer log terminal ────────────────────────────────────── -->
		<footer class="footer fixed-bottom bg-black border-top border-secondary shadow-lg"
		style="height:160px;">
		<div class="d-flex justify-content-between align-items-center px-3 py-1 bg-dark border-bottom border-secondary">
		<small class="text-muted fw-bold">SYSTEM LOGS</small>
		<div class="d-flex gap-3 align-items-center">
		<small id="ws-status" class="text-muted">connecting...</small>
		<button class="btn btn-sm text-muted" onclick="document.getElementById('log-terminal').innerHTML=''">× Clear</button>
		<small id="ws-status" class="text-muted">connecting…</small>
		<button class="btn btn-sm text-muted"
		onclick="document.getElementById('log-terminal').innerHTML=''">× Clear</button>
		</div>
		</div>
		<div id="log-terminal" class="p-2 overflow-auto"
		style="height: 125px; font-family: 'Courier New', monospace; font-size: 0.75rem; color: #00ff00;">
		<div id="log-terminal"
		class="p-2 overflow-auto"
		style="height:125px; font-family:'Courier New',monospace; font-size:0.75rem;">
		</div>
		</footer>

		<!-- Vendor Scripts -->
		<!-- ── Vendor scripts ─────────────────────────────────────────────────── -->
		<script src="{{ url_for('web.static', filename='js/vendor/bootstrap.bundle.min.js') }}"></script>
		<script src="{{ url_for('web.static', filename='js/vendor/ansi_up.js') }}"></script>

		<!-- ── UI primitives (must come before actions / status scripts) ──────── -->
		<script src="{{ url_for('web.static', filename='js/ui.js') }}"></script>

		<!-- Unified WebSocket Client (Loaded on every page) -->
		<script src="{{ url_for('web.static', filename='js/websocket-client.js') }}"></script>
		<!-- ── Unified WebSocket client ───────────────────────────────────────── -->
		<script src="{{ url_for('web.static', filename='js/ws-client.js') }}"></script>

		{% block scripts %}{% endblock %}
		</body>