bCNC Phone Alerts: Changes to Sender.py and __main__.py

What each change in bcnc-alerts.diff does, so you can review it, apply it by hand, or adapt it. Testing instructions are at the end.

Overview

bCNC sends a phone notification through ntfy (free, no account) for these events:

The changes are in two files that sit in the same folder. Sender.py gets one new import, one block of helper code near the top, and small additions to four existing methods. __main__.py gets a new --ntfy-topic launch option. They were written against one specific bCNC version; method names and line positions may differ in other versions.

Set up your ntfy topic (required)

The patch contains no topic name. Each person picks their own, so alerts stay private to you.

  1. Install the ntfy app (iOS or Android).
  2. Make up a topic name that nobody could guess. Topics on ntfy.sh are effectively public, so the obscure name acts as your password. For example, yourname-cnc- plus a random string. To generate one:
    python3 -c "import secrets; print('cnc-' + secrets.token_urlsafe(12))"
    Subscribe to that topic in the app.
  3. Tell bCNC the topic, on the machine that runs it. Pick one of three ways:
    • Launch option. Start bCNC with --ntfy-topic=YOUR_TOPIC. This is the best choice when several people share one machine, because each person's launcher or shortcut carries their own topic.
    • Topic file. Put it alone on the first line of a file named .bCNC-ntfy-topic in your home folder:
      echo "YOUR_TOPIC" > ~/.bCNC-ntfy-topic
      (on Windows, create .bCNC-ntfy-topic in C:\Users\YourName with the topic as its only line). This suits a machine where each person has their own login.
    • Environment variable. Set BCNC_NTFY_TOPIC in the environment bCNC is launched from.
    If more than one is set, the launch option wins, then the environment variable, then the file.

If no topic is configured, no alerts are sent and bCNC prints one line saying so to its console. Changing the topic takes effect on the next alert, with no restart.

Applying the changes

Option A, the diff. Back up Sender.py and __main__.py, then from the folder that contains them run:

patch -p1 < bcnc-alerts.diff

Option B, the whole files. Back up your Sender.py and __main__.py and replace them with the patched ones. This only works if your bCNC version matches the files the diff was made from.

Option C, by hand. Follow steps 1 to 8 below.

1. Add an import

Top of Sender.py, with the other imports (os, time and threading are already imported there):

import urllib.request

2. Add the helper code

Directly after the line from CNC import CNC, MSG, UPDATE, WAIT, GCode. Nothing needs editing here; the topic comes from your setup above.

# ----------------------------------------------------------------------
# Phone alerts via ntfy (https://ntfy.sh)
# ----------------------------------------------------------------------
ALARM_REPEATS = 5  # total alarm alerts while the alarm persists
ALARM_REPEAT_SECS = 120  # seconds between repeated alarm alerts
NTFY_TOPIC_FILE = os.path.expanduser("~/.bCNC-ntfy-topic")
_ntfy_warned = [False]


def ntfy_topic():
    """ntfy topic from $BCNC_NTFY_TOPIC (which the --ntfy-topic launch
    option sets), else the first line of ~/.bCNC-ntfy-topic.
    Returns "" when not configured (alerts off)."""
    topic = os.environ.get("BCNC_NTFY_TOPIC", "").strip()
    if not topic:
        try:
            with open(NTFY_TOPIC_FILE) as f:
                topic = f.readline().strip()
        except OSError:
            topic = ""
    return topic


def notify_phone(msg, priority=None, tags=None):
    topic = ntfy_topic()
    if not topic:
        if not _ntfy_warned[0]:
            _ntfy_warned[0] = True
            print("bCNC phone alerts are off: no ntfy topic in", NTFY_TOPIC_FILE)
        return

    def _send():
        try:
            headers = {"Title": "bCNC"}
            if priority:
                headers["Priority"] = priority  # e.g. "high", "urgent"
            if tags:
                headers["Tags"] = tags  # ntfy emoji shortcodes
            req = urllib.request.Request(
                "https://ntfy.sh/" + topic,
                data=msg.encode("utf-8"),
                headers=headers,
                method="POST",
            )
            urllib.request.urlopen(req, timeout=10)
        except Exception:
            pass  # never let a failed alert interrupt the machine

    threading.Thread(target=_send, daemon=True).start()


def job_summary(sender):
    """Returns (file name, duration string or None)."""
    name = (
        os.path.basename(getattr(sender.gcode, "filename", "") or "") or "unnamed job"
    )
    start = getattr(sender, "_job_start", None)
    if not start:
        return name, None
    m, s = divmod(int(time.time() - start), 60)
    h, m = divmod(m, 60)
    return name, ("{}h {}m".format(h, m) if h else "{}m {}s".format(m, s))


_last_problem = [0.0]


def notify_problem(msg):
    # High priority, at most one per 60 seconds, so a burst of
    # errors does not flood the phone.
    now = time.time()
    if now - _last_problem[0] < 60:
        return
    _last_problem[0] = now
    notify_phone(msg, priority="urgent", tags="rotating_light")


_alarm = {"timer": None, "active": False, "last": 0.0}


def alarm_cancel():
    _alarm["active"] = False
    t = _alarm["timer"]
    _alarm["timer"] = None
    if t is not None:
        t.cancel()


def _alarm_schedule(msg, count, delay):
    t = threading.Timer(delay, _alarm_loop, (msg, count))
    t.daemon = True
    _alarm["timer"] = t
    t.start()


def _alarm_loop(msg, count):
    if not _alarm["active"]:
        return
    _alarm["last"] = time.time()
    notify_phone(msg, priority="urgent", tags="rotating_light")
    if count < ALARM_REPEATS:
        _alarm_schedule(msg, count + 1, ALARM_REPEAT_SECS)


def alarm_start(msg):
    # May be called many times while the controller stays in Alarm (or
    # flips in and out of it); only the first call starts a series, and
    # alerts are never closer together than ALARM_REPEAT_SECS.
    if _alarm["active"]:
        return
    _alarm["active"] = True
    wait = _alarm["last"] + ALARM_REPEAT_SECS - time.time()
    _alarm_schedule(msg, 1, max(0.0, wait))

What the helpers do:

3. Record the start time

In initRun, right after self.running = True:

        self.running = True
        self._job_start = time.time()
        self._job_stopped = False

The start time feeds the duration. The stop flag is reset here so each job starts clean.

4. Job finished

In runEnded, inside the if self.running: block, after the three MSG_RUNEND log lines and before the if self._onStop: check:

            if not getattr(self, "_job_stopped", False):
                name, dur = job_summary(self)
                if dur:
                    notify_phone(
                        "{} finished in {}".format(name, dur), tags="white_check_mark"
                    )
                else:
                    notify_phone("{} finished".format(name), tags="white_check_mark")

The stop flag prevents a "finished" message for a job that was stopped.

5. Job stopped

At the very top of stopRun, before self.feedHold():

        if self.running and not getattr(self, "_job_stopped", False):
            self._job_stopped = True
            name, dur = job_summary(self)
            notify_phone(
                "{} stopped{}".format(name, " after " + dur if dur else ""),
                priority="high",
                tags="stop_sign",
            )

In this version of bCNC, stopRun does not call runEnded, so the stop alert is sent from here. It only fires if a job is running, and only once per job.

6. Controller alarm

In controllerStateChange, right after the print(...) call at the top:

        if state == "Alarm":
            if self.running:
                alarm_start("ALARM during {}".format(job_summary(self)[0]))
            else:
                alarm_start("Controller ALARM")
        else:
            alarm_cancel()

bCNC calls this when the controller's reported state changes, and possibly more often than that, which is why the alarm helpers guard against repeats. Entering "Alarm" starts the repeating alert; any other state cancels it, which is how clearing the alarm silences it.

7. G-code evaluation errors

In serialIO, in the except Exception: block that logs expression errors, right after the for s in ...: self.log.put((Sender.MSG_ERROR, s)) loop:

                            notify_problem(
                                "G-code error in {}: {}".format(
                                    job_summary(self)[0], str(sys.exc_info()[1])[:100]
                                )
                            )

This covers errors raised while bCNC evaluates a G-code expression or macro.

8. Add the launch option

In __main__.py, in three places. First, in the usage() function, after the -l | --list line:

    wrt("\t--ntfy-topic #\t\tntfy topic for phone alerts (overrides\n")
    wrt("\t\t\t\t$BCNC_NTFY_TOPIC and ~/.bCNC-ntfy-topic)\n")

Second, in the list of long options passed to getopt.getopt, after "run",:

                "ntfy-topic=",

Third, in the for opt, val in optlist: loop in main(), after the --run branch:

        elif opt == "--ntfy-topic":
            # Read by Sender.ntfy_topic() when an alert is sent
            os.environ["BCNC_NTFY_TOPIC"] = val.strip()

The option stores the topic in the BCNC_NTFY_TOPIC environment variable for this bCNC process, which is where ntfy_topic() in Sender.py looks first. That is why the launch option beats the topic file. Both --ntfy-topic=NAME and --ntfy-topic NAME work.

Not covered: plain GRBL error:N responses (for example, an unsupported command) are parsed in the controller files (such as GRBL1.py in the controllers folder), not in Sender.py, so they don't trigger an alert yet.

Testing

Before you start

Tests

curl -d "test" ntfy.sh/YOUR_TOPIC
Expect: a "test" notification on your phone. If not, the problem is the network or topic name, not bCNC.
Expect: "yourfile.nc finished in Xm Ys".
Expect: a high-priority "yourfile.nc stopped after Xm Ys" and no "finished" message.
Expect: an urgent "Controller ALARM" alert right away, repeating every 2 minutes (up to 5 times) until the alarm is cleared.
Expect: the repeating alerts stop.
Expect: "ALARM during yourfile.nc".
Expect: the "finished" alert arrives on the second topic only, not on the topic from your topic file or environment variable.

If something doesn't fire

Reverting

Copy your backups of the original Sender.py and __main__.py back over the patched ones and restart bCNC.