Merge branch 'main' of https://www.ict.inaf.it/gitlab/davide.ricci/software-di-controllo

image: python:3.12

image: python:3.13

variables:

  PROJECT: "noctua"

http://software-di-controllo-830bce.pages.ict.inaf.it/

## Python dependencies:

## Virtual Environment

It is preferable to use a virtual environment. To do so:

```bash

python -m venv venv

source venv/bin/activate

```

Once activated, your terminal prompt will change to indicate that you are inside the virtual environment (e.g., `(venv) your-username@your-machine:...$`).

To deactivate de virtual environment:

```bash

deactivate

```

# For the core routines:

pip3 install requests loguru astropy

pip3 install gnuplotlib pyvantagepro # maybe I'll drop these.

# For the API:

pip3 install flask flask-restx werkzeug

# Tested with flask==3.0.2 werkzeug-3.0.1 flask-restx==1.3.0 

## Python dependencies:

# For the web interface:

pip3 install flask flask-socketio flask-httpauth

# Tested with flask==3.0.2 werkzeug-3.0.1 flask-socketio==5.3.6 flask-httpauth==4.8.0

To install the dependencies:

```

pip install -e .

```

In particular, for the core routines:

`requests loguru astropy` and

`gnuplotlib pyvantagepro`, that will be soon removed.

For the API:

`flask==3.0.2 werkzeug-3.0.1 flask-restx==1.3.0`, that will be soon replaced by `quart`

For the web interface:

`flask==3.0.2 werkzeug-3.0.1 flask-socketio==5.3.6 flask-httpauth==4.8.0`

## Add the nodes of your observatory:

#!/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 noctua.utils.logger import log

from .blocks import blocks_api

from .defaults import defaults_api

from .sequencer import sequencer_api

from .sequencer import sequencer_api, BobRun

api_blueprint = Blueprint('api', __name__)

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)

# # Uncomment to test dependecy errors

# resource_registry = {}

resource_registry['/sequencer/run'] = BobRun(seq)

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", "/templates", "/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

        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}") # noctua.api.camera

        resource_class = getattr(module, res_class_name) # noctua.api.camera.CoolerWarmup

        # 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())]

        # Register dynamic route as MethodView

        # This automatically maps GET/POST/PUT/DELETE to class methods

        view = resource_class.as_view(url_path, dev=dev_instance)

        # Populate internal registry

        resource_registry[url_path] = resource_class(dev=dev)

        # Register route

        view = resource_class.as_view(url_path, dev=dev)

        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__}"

        log.info(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}")

        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", "/templates", "/sequencer")):

        dynamic_import(section)

import configparser

from pathlib import Path

ends = configparser.ConfigParser()

config_path = Path(__file__).parent.parent / "config" / "api.ini"

ends.read(config_path)

resource_registry = {}

#!/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.views import MethodView

# Custom modules

from .deps import dep_checker # Change this import

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 an 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.

        Build a standardized JSON response dict.

        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 of (response dict, 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,

            "timestamp": self.timestamp,

        }

        return jsonify(res), status_code

        return res, status_code

    async def dispatch_request(self, *args, **kwargs):

        """

        Override the default dispatch to enforce cached dependency checks.

        """

        api_path = request.path.replace("/api", "", 1)

        handler = getattr(self, request.method.lower(), None)

        if handler is None:

            from quart import abort

            abort(405)

        force = request.args.get('force') == 'true'

        # Use dep_checker instead of check_dependencies

        is_available, err_msg = await dep_checker.get_status(api_path, force)

        if not is_available:

            return jsonify({

                "raw": None,

                "response": None,

                "error": [err_msg],

                "timestamp": self.timestamp,

            }), 424

        return await super().dispatch_request(*args, **kwargs)

def register_error_handlers(app):

    now = datetime.utcnow().isoformat()

    """

    Bind 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

        return jsonify({"error": ["Method not allowed"], "timestamp": datetime.utcnow().isoformat()}), 405

    @app.errorhandler(424)

    async def failed_dependency(e):

        return jsonify({"error": ["Failed Dependency"], "timestamp": datetime.utcnow().isoformat()}), 424