forgefirm/meta-forgefirm/recipes-forgefirm/gfhome/files/gfhome.py

#!/usr/bin/python3
"""
gfhome - one-shot Glowforge web-service homing for ForgeFIRM

Connects the machine to the Glowforge web service just long enough for
the service to run its camera-based homing sequence (settings report ->
hunt (Z/lens) -> lid image -> move to the home corner -> lid image),
then parks the lens at the hall-sensor reference, disconnects, and
exits. All three axes end at the factory home position: X/Y at the
back-left home corner, Z at the top-of-travel hall trigger.

The grblHAL-glowforge controller invokes this for $H when
homing_mode = gfcloud is set in /data/forgefirm.conf, releasing
/dev/glowforge for the duration of the run. It can also be run by hand
(with the controller stopped or its homing session active).

The service ends the sequence silently - there is no completion
message - so the run is considered homed once a hunt and at least one
motion have completed and the service has been quiet for --quiet
seconds.

Exit codes: 0 = homed, 1 = configuration/connection failure,
2 = homing did not complete.

(C) Copyright 2026
Scott Wiederhold, s.e.wiederhold@gmail.com
SPDX-License-Identifier: MIT
"""
import argparse
import json
import logging
import queue
import shutil
import signal
import sys
import time
from pathlib import Path
from queue import Queue

from gfutilities.configuration import parse, get_cfg, log_level
from gfutilities.service.authentication import authenticate_machine
from gfutilities.service.websocket import get_session, WsClient

CONF = '/data/etc/gfhome.conf'
CONF_SAMPLE = '/etc/gfhome.conf.sample'

logging.basicConfig(format='(%(levelname)s) %(module)s:%(funcName)s %(message)s')
logger = logging.getLogger('openglow')


def load_config(path: str) -> bool:
    if path == CONF and not Path(CONF).is_file() and Path(CONF_SAMPLE).is_file():
        Path(CONF).parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(CONF_SAMPLE, CONF)
    if not Path(path).is_file():
        logger.error('config file %s not found', path)
        return False
    parse(path)
    if not get_cfg('SERVICE.SERVER_URL'):
        logger.error('config %s has no SERVICE section', path)
        return False
    if get_cfg('LOGGING.FILE'):
        Path(get_cfg('LOGGING.FILE')).parent.mkdir(parents=True, exist_ok=True)
        fh = logging.FileHandler(get_cfg('LOGGING.FILE'))
        fh.setLevel(log_level(get_cfg('LOGGING.LEVEL')))
        fh.setFormatter(logging.Formatter(
            '%(asctime)s (%(levelname)s) %(module)s:%(funcName)s %(message)s'))
        logger.addHandler(fh)
    logger.setLevel(logging.DEBUG)
    return True


def dispatch(machine, msg: dict) -> str:
    """Route one service action to the machine (the GFUIService dispatch
    table, minus print - a print must never run inside a homing session).
    Returns the action type if a puls action was accepted for run."""
    action = msg.get('action_type', '')
    if action == 'settings' and msg.get('status') == 'ready':
        machine.run_settings_report(msg)
    elif 'image' in action:
        machine.run_capture(msg)
    elif action in ('hunt', 'motion'):
        machine.run_puls(msg)
        if machine.running_action_id == msg.get('id'):
            return action
    elif action == 'print':
        logger.warning('ignoring print action during homing')
    else:
        logger.info('ignoring action %s', action)
    return ''


def home(machine, args) -> int:
    q_rx: Queue = Queue()
    q_tx: Queue = Queue()

    session = get_session()
    if not authenticate_machine(session):
        logger.error('sign-in to %s failed', get_cfg('SERVICE.SERVER_URL'))
        return 1

    # ws_connect() discards the client object, which makes a clean
    # disconnect impossible - build the client directly instead.
    ws = WsClient(q_rx, q_tx)
    ws.start()
    deadline = time.monotonic() + 16
    while not ws.ready and time.monotonic() < deadline:
        time.sleep(0.5)
    if not ws.ready:
        logger.error('web socket connection failed')
        ws.stop = True
        return 1

    result = 2
    try:
        machine.start(session, q_tx)

        from gfhardware._common import InputSwitch
        switches = machine._sw_thread.all_switches()
        if not switches[InputSwitch.SW_DOORS]:
            logger.error('lid is open - close it and re-home')
            return 2
        if not switches[InputSwitch.SW_ESTOP]:
            logger.error('e-stop is tripped')
            return 2

        t0 = time.monotonic()
        last_activity = t0
        in_flight = ''
        done = set()

        while True:
            now = time.monotonic()
            if now - t0 > args.timeout:
                logger.error('homing timed out after %ds (completed: %s)',
                             args.timeout, sorted(done) or 'nothing')
                return 2
            if 'hunt' not in done and now - t0 > args.start_timeout:
                logger.error('service did not start homing within %ds',
                             args.start_timeout)
                return 2

            busy = bool(machine.running_action_id)
            if busy:
                last_activity = now
            elif in_flight:
                logger.info('%s completed', in_flight)
                done.add(in_flight)
                in_flight = ''

            if ('hunt' in done and 'motion' in done and not busy
                    and now - last_activity >= args.quiet):
                logger.info('homing complete (service quiet %.0fs)', args.quiet)
                result = 0
                break

            try:
                msg = json.loads(q_rx.get(timeout=0.5))
            except queue.Empty:
                continue
            except ValueError:
                logger.warning('unparseable service message')
                continue
            last_activity = time.monotonic()
            logger.info('service action: %s (%s)',
                        msg.get('action_type'), msg.get('status'))
            started = dispatch(machine, msg)
            if started:
                in_flight = started
    finally:
        if result == 0:
            # Deterministic Z: the hunt file leaves the lens wherever its
            # pattern ends; re-reference against the hall sensor so the
            # controller can trust top-of-travel.
            try:
                from gfhardware.z_axis import ZAxis
                ZAxis.home()
            except Exception:
                logger.exception('final Z reference failed')
                result = 2
        ws.stop = True
        try:
            machine.stop()
        except Exception:
            logger.exception('machine shutdown failed')
    return result


def main() -> int:
    ap = argparse.ArgumentParser(description='ForgeFIRM one-shot Glowforge cloud homing')
    ap.add_argument('-c', '--config', default=CONF, help='config file (default %s)' % CONF)
    ap.add_argument('--timeout', type=int, default=240,
                    help='overall time budget in seconds (default 240)')
    ap.add_argument('--start-timeout', type=int, default=120,
                    help='max seconds to wait for the service to begin homing (default 120)')
    ap.add_argument('--quiet', type=int, default=10,
                    help='silence after the last action that means done (default 10)')
    args = ap.parse_args()

    signal.signal(signal.SIGTERM, lambda *_: sys.exit(2))

    if not load_config(args.config):
        return 1

    # Machine() reads the OCOTP identity and head info; it fails cleanly
    # when the controller still owns /dev/glowforge.
    try:
        from gfhardware import Machine
        machine = Machine()
    except Exception:
        logger.exception('machine init failed (is the motion controller '
                         'still holding /dev/glowforge?)')
        return 1

    rc = home(machine, args)
    logger.info('exit %d (%s)', rc, 'homed' if rc == 0 else 'not homed')
    return rc


if __name__ == '__main__':
    sys.exit(main())