Log on footer

        return self.make_response(res)

class Connection(BaseResource):

    '''Manage the connection to ASCOM.'''

# class Connection(BaseResource):

#     '''Manage the connection to ASCOM.'''

    async def get(self):

        '''Check if the telescope is connected to ASCOM.'''

#     async def get(self):

#         '''Check if the telescope is connected to ASCOM.'''

        res = await self.run_blocking(lambda: self.dev.connection)

        return self.make_response(constants.yes_no.get(res, "N/A"), raw_data=res)

#         res = await self.run_blocking(lambda: self.dev.connection)

#         return self.make_response(constants.yes_no.get(res, "N/A"), raw_data=res)

    async def put(self):

        '''Connect or disconnect the telescope to ASCOM.'''

#     async def put(self):

#         '''Connect or disconnect the telescope to ASCOM.'''

        connection = await self.get_payload()

        def action():

            self.dev.connection = connection

            return self.dev.connection

        res = await self.run_blocking(action)

        return self.make_response(res, raw_data=res)

#         connection = await self.get_payload()

#         def action():

#             self.dev.connection = connection

#             return self.dev.connection

#         res = await self.run_blocking(action)

#         return self.make_response(res, raw_data=res)

class Focuser(BaseResource):

    def park(self):

        """Puts the dome in its park position."""

        self.put("park")

        # self.put("park")

        self.slave = False

        self.azimuth = self.PARK

    def sync(self, az):

        """

    def is_parked(self):

        """bool or None: Checks if the dome is at its park position."""

        res = self.get("atpark")

        if self.error:

            return None

        # res = self.get("atpark")

        # if self.error:

        #     return None

        # return res

        try:

            res = abs(self.PARK-self.azimuth) < self.TOLERANCE

        except TypeError as e:

            res = None

        return res

    @property

# System modules

import shutil

from pathlib import Path

import datetime

# Third-party modules

from astropy.io import fits

                                LOG_FOLDER, dateobs, dir_type, frame_number,

                                imagetyp)

PROJECT_ROOT = Path(__file__).parent.parent.parent

PACKAGE_ROOT = Path(__file__).parent.parent.absolute()

PROJECT_ROOT = PACKAGE_ROOT.parent

def current_log_path():

    """

    Get the full path to the active system log file.

    Returns

    -------

    str

        The absolute path to OARPAF.log.

    """

    # We look for the data folder in the project root

    log_dir = PROJECT_ROOT / DATA_FOLDER / LOG_FOLDER

    filename = f"{FILE_PREFIX}.{LOG_EXT}"

    full_path = log_dir / filename

    return str(full_path)

def date_folder():

# System modules

import os

import asyncio

import json

# Custom modules

from noctua.api.basecontext import ends

from noctua.utils.logger import log

from noctua.utils.structure import current_log_path

from .stream import BroadcastManager, StreamManager

web_blueprint = Blueprint('web', __name__, template_folder='pages', static_folder='static')

# Initialize managers

broadcaster = BroadcastManager()

streamer = StreamManager(broadcaster)

@web_blueprint.before_app_serving

async def start_background_tasks():

    """

    Start telemetry loops for each subsystem and monitor tasks.

    Launch independent telemetry and monitoring loops.

    """

    # Identify subsystems to monitor from api.ini

    subsystems = set()

    for section in ends.sections():

        parts = section.strip('/').split('/')

        if parts:

            subsystems.add(parts[0])

    # Start an independent loop for each subsystem

    for subsys in subsystems:

        if subsys not in ("blocks", "sequencer", "templates"):

            asyncio.create_task(streamer.subsystem_poll_loop(subsys))

    # Start other general loops

    asyncio.create_task(streamer.log_loop())

    asyncio.create_task(streamer.image_loop())

@web_blueprint.websocket('/socket')

async def ws():

    """

    Handle WebSocket lifecycle and incoming control messages.

    Handle WebSocket connection and initial log synchronization.

    """

    await broadcaster.register(websocket._get_current_object())

    real_ws = websocket._get_current_object()

    await broadcaster.register(real_ws)

    # Send history from OARPAF.log

    path = current_log_path()

    if os.path.exists(path):

        try:

            with open(path, 'r') as f:

                lines = f.readlines()

                last_10 = lines[-10:]

                if last_10:

                    await real_ws.send(json.dumps({"name": "new-lines", "data": last_10}))

        except Exception as e:

            log.error(f"WebSocket: Initial log read failed: {e}")

    try:

        while True:

            message_raw = await websocket.receive()

            msg = json.loads(message_raw)

            # Handle control commands from the UI

            if msg.get("action") == "set_force":

                streamer.force_enabled = bool(msg.get("value"))

                log.info(f"WebSocket: Telemetry Force mode set to {streamer.force_enabled}")

                log.info(f"WebSocket: Force set to {streamer.force_enabled}")

    except asyncio.CancelledError:

        pass

    finally:

        await broadcaster.unregister(websocket._get_current_object())

        await broadcaster.unregister(real_ws)

@web_blueprint.route('/init')

async def init_panel():

    """

    Render the initialization dashboard.

    Render the main dashboard for observatory initialization.

    Returns

    -------

    str

        The rendered HTML content.

    """

    return await render_template('init.html')

@web_blueprint.route('/status')

async def status_monitor():

    """

    Render a real-time JSON monitor for all subsystems.

    Render the real-time JSON telemetry monitor.

    Returns

    -------

    str

        The rendered HTML template.

        The rendered HTML content.

    """

    return await render_template('status.html')

@web_blueprint.route('/subsystem')

@web_blueprint.route('/subsystem/')

async def subsystems():

    """

    List all available subsystems dynamically from configuration.

    Returns

    -------

    str

        The rendered HTML list.

    """

    sub_set = set()

    for section in ends.sections():

        parts = section.strip('/').split('/')

        if parts:

            sub_set.add(parts[0])

    return await render_template('subsystems.html', subsystems=sorted(list(sub_set)))

@web_blueprint.route('/subsystem/<string:subsystem_name>')

@web_blueprint.route('/subsystem/<string:subsystem_name>/<path:endpoint_name>')

async def subsystem_page(subsystem_name, endpoint_name=None):

    """

    Render a specific subsystem control panel or a single widget.

    Parameters

    ----------

    subsystem_name : str

        The name of the subsystem group.

    endpoint_name : str, optional

        A specific API path within the subsystem.

    Returns

    -------

    str

        The rendered HTML content.

    """

    subsys_endpoints = {}

    for section in ends.sections():

        is_match = False

        if endpoint_name:

            if section == f"/{subsystem_name}/{endpoint_name}":

                is_match = True

        else:

            if section.startswith(f"/{subsystem_name}/"):

                is_match = True

        if is_match:

            key = section.split("/")[-1]

            subsys_endpoints[key] = {

                "path": section,

                "resource": ends.get(section, "resource", fallback="Unknown")

            }

    return await render_template(

        'subsystem.html',

        subsystem=subsystem_name,

        endpoints=subsys_endpoints

    )

<html lang="it" data-bs-theme="dark">

<head>

    <meta charset="UTF-8">

    <meta name="viewport" content="width=device-width, initial-scale=1.0">

    <title>Noctua - {{ title | default('Control') }}</title>

    <title>Noctua Control</title>

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

    <nav class="navbar navbar-expand-lg navbar-dark bg-dark mb-4">

        <div class="container-fluid">

            <a class="navbar-brand" href="#">Noctua Control</a>

        </div>

    </nav>

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

    <div class="container">

    <div class="container mt-4">

        {% block content %}{% endblock %}

    </div>

    <!-- Footer Notification Area (Toasts) -->

    <div class="toast-container position-fixed bottom-0 end-0 p-3" id="notification-container" style="z-index: 1100;">

      <!-- Toasts injected by actions.js -->

    <!-- Notification Container -->

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

    <!-- 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>

            <button class="btn btn-sm text-muted" onclick="document.getElementById('log-terminal').innerHTML=''">&times; Clear</button>

        </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>

    </footer>

    <!-- Bootstrap Bundle with Popper -->

    <script src="{{ url_for('web.static', filename='js/vendor/bootstrap.bundle.min.js') }}"></script>

    {% block scripts %}{% endblock %}

</body>

</html>