Automation Pipelines & Triggers: Ring the Doorbell

Introduction

So far your detector sees. Automation is when it also does something, by itself: logs a goal, sounds an alert, saves a clip. Today you'll build a goal detector that rings a doorbell (an event) exactly once per goal, and plug in listeners that react, then test it with pretend matches so you can trust it on the real day.

By the end of this lesson you'll be able to:

Describe the perceive, detect, act pipeline
Explain polling versus event-driven designs and measure the difference
Use a small event bus so detection and actions stay separate
Write a goal trigger with three rules (confirm, come from outside, cooldown) and say what each rule prevents
Test a trigger with scripted situations, including the failures
Log events with timestamps and update a score

Working principle

Perceive, detect, act

camera (perceive) -> YOLO + rules (detect) -> event -> log / sound / score / hardware (act)

Detection answers "what is in the picture". Automation answers "what should happen because of it". The second half needs just as much care.

Polling versus event-driven

	Polling	Event-driven
Idea	Keep asking "has it happened yet?"	Be told when it happens
Delay	Up to one polling interval	Nearly zero
Cost	Busy even when nothing happens	Idle until needed
Code	A loop with `sleep`	Handlers called by an event

The event bus

A tiny object with two methods: on(name, handler) registers someone who cares, and emit(name, **details) announces it. The detector never needs to know about the sound or the log. Adding a new action (say, an LED in lesson 35) is one extra bus.on(...) line, and nothing in the detector changes.

A goal is not "the ball is in the zone"

A raw rule ("fire whenever the ball is inside the goal box") produces false alarms all the time. Our detector adds three rules:

Rule	What it prevents
Confirm: the ball must be inside for 3 frames in a row	A single stray detection becoming a goal
From outside: the ball must have been seen outside the zone first	A ball resting in the net being counted again and again
Cooldown: wait 5 seconds before counting another	A flickering detector counting one goal twice

Each rule trades a risk of false alarms for a risk of missed goals (lesson 19). The numbers (3 frames, 5 seconds) are decisions to be tested, not facts.

What if the ball isn't detected?

The ball is small and often hidden by a net or a player. We made a choice: a missing ball is neutral. It does not reset the count, because the ball may simply be hidden inside the goal. The alternative (reset) would miss goals where the net hides the ball. Which is better depends on the cost of a miss versus a false alarm, and on your camera angle.

Test with scripted situations

Real matches are rare and slow. Write the situations as short lists of ball positions (one per frame) and check the answers automatically. A trigger without tests is a trigger that works only on the day you tried it.

What you'll need

Your cv_course environment with Ultralytics
Your detector from lesson 24 and a match clip (or the scripted test situations below)

Build it

Put each file in the same folder. They are small on purpose.

Cell 1: polling versus events (timed)

import threading
import time
 
# Something happens at a random moment. How long until we notice, polling versus events?
event_time = {}
noticed = {}
 
 
def polling_watcher(interval, stop):
    while not stop.is_set():
        if "t" in event_time and "polling" not in noticed:
            noticed["polling"] = time.perf_counter()
        time.sleep(interval)                       # "has anything happened?" every <interval> seconds
 
 
def run(interval):
    event_time.clear(); noticed.clear()
    stop = threading.Event()
    waiter = threading.Event()                     # the event-driven version: sleep until told
    threading.Thread(target=polling_watcher, args=(interval, stop), daemon=True).start()
 
    def listener():
        waiter.wait()                              # no checking at all, just waiting
        noticed["event"] = time.perf_counter()
    threading.Thread(target=listener, daemon=True).start()
 
    time.sleep(0.137)                              # the goal happens at an awkward moment
    event_time["t"] = time.perf_counter()
    waiter.set()                                   # "emit"
    time.sleep(interval * 1.5)
    stop.set()
    return (noticed["polling"] - event_time["t"]) * 1000, (noticed["event"] - event_time["t"]) * 1000
 
 
for interval in (0.5, 0.1, 0.02):
    p, e = run(interval)
    print(f"check every {interval * 1000:4.0f} ms: polling noticed after {p:6.1f} ms   event-driven noticed after {e:5.2f} ms")

You should see polling noticing the event late by roughly half its interval (or a lot more for the long one), and the event-driven version noticing in a fraction of a millisecond. On your laptop the numbers will vary.

Cell 2: `events.py`

from collections import defaultdict
 
 
class EventBus:
    """A tiny doorbell system. Parts of the program say what happened. Other parts decide what to do about it."""
 
    def __init__(self):
        self.handlers = defaultdict(list)
 
    def on(self, name, handler):
        self.handlers[name].append(handler)
 
    def emit(self, name, **details):
        for handler in self.handlers[name]:
            handler(**details)

Cell 3: `goal_detector.py`

class GoalDetector:
    """
    Decides when a goal has happened, from the ball's position in each frame.
    Rules (each one stops a particular false alarm):
      1. the ball must be INSIDE the goal zone for `confirm_frames` frames in a row    (a single stray detection is not a goal)
      2. it must have been seen OUTSIDE the zone before                               (a ball resting in the net is not a new goal)
      3. after a goal, wait `cooldown` seconds before counting another                 (a flickering detector is not two goals)
    """
 
    def __init__(self, zone, bus, name="goal", confirm_frames=3, cooldown=5.0):
        self.zone, self.bus, self.name = zone, bus, name
        self.confirm_frames, self.cooldown = confirm_frames, cooldown
        self.inside_count, self.seen_outside, self.last_goal = 0, False, -1e9
 
    def inside(self, ball):
        z = self.zone
        return ball is not None and z["x"] <= ball["x"] <= z["x"] + z["w"] and z["y"] <= ball["y"] <= z["y"] + z["h"]
 
    def update(self, ball, t):
        """Call once per frame. ball = {"x":..., "y":...} or None. t = video time in seconds."""
        if self.inside(ball):
            self.inside_count += 1
            if self.inside_count == self.confirm_frames and self.seen_outside and t - self.last_goal >= self.cooldown:
                self.last_goal = t
                self.seen_outside = False                  # must leave and come back for the next one
                self.bus.emit("goal", zone=self.name, t=t)
        elif ball is not None:                             # the ball is visible and OUTSIDE the zone
            self.inside_count = 0
            self.seen_outside = True
        # ball is None (not detected): do nothing. A ball hidden by the net or a player is not evidence either way.

Cell 4: `actions.py` (log, score and sound)

The sound is made with the standard wave module, so you don't need an audio file. Playing it needs pygame (pip install pygame). Without it, the code falls back to the terminal bell.

import csv
import math
import os
import struct
import wave
from datetime import datetime
 
 
def make_goal_sound(path="goal.wav"):
    """Write a short two-tone 'goal!' sound to a WAV file using only the standard library."""
    rate = 22050
    samples = []
    for freq, seconds in ((660, 0.18), (880, 0.28)):
        for i in range(int(rate * seconds)):
            fade = min(1, (seconds * rate - i) / (0.05 * rate))                # fade out so it doesn't click
            samples.append(int(12000 * fade * math.sin(2 * math.pi * freq * i / rate)))
    with wave.open(path, "w") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(rate)
        w.writeframes(b"".join(struct.pack("<h", s) for s in samples))
    return path
 
 
def play_sound(path="goal.wav"):
    try:
        import pygame                                   # pip install pygame
        if not pygame.mixer.get_init():
            pygame.mixer.init()
        pygame.mixer.Sound(path).play()
    except Exception:
        print("\a", end="")                             # no pygame or no audio device: terminal bell instead
 
 
class GoalLog:
    def __init__(self, path="goals.csv"):
        self.path = path
        if not os.path.exists(path):
            with open(path, "w", newline="") as f:
                csv.writer(f).writerow(["wall_clock", "video_time_s", "zone"])
 
    def __call__(self, zone, t):
        with open(self.path, "a", newline="") as f:
            csv.writer(f).writerow([datetime.now().isoformat(timespec="seconds"), round(t, 2), zone])
 
 
class Score:
    def __init__(self):
        self.goals = {}
 
    def __call__(self, zone, t):
        self.goals[zone] = self.goals.get(zone, 0) + 1
        print(f"SCORE  " + "  ".join(f"{z}: {n}" for z, n in sorted(self.goals.items())))

Cell 5: replay the match from the lesson 30 data

Needs detections.json from lesson 30 (make_demo_data.py or export_json.py). Edit goal_zone in the JSON to match your own video.

import json
import sys
 
from actions import GoalLog, Score, make_goal_sound, play_sound
from events import EventBus
from goal_detector import GoalDetector
 
SRC = sys.argv[1] if len(sys.argv) > 1 else "detections.json"          # from lesson 30 (make_demo_data.py or export_json.py)
data = json.load(open(SRC))
 
bus = EventBus()
sound = make_goal_sound()
bus.on("goal", GoalLog())                                              # 1. write a line in goals.csv
bus.on("goal", Score())                                                # 2. update the score
bus.on("goal", lambda zone, t: play_sound(sound))                      # 3. play the sound
bus.on("goal", lambda zone, t: print(f"GOAL at {t:.1f} s in the {zone}"))
 
detector = GoalDetector(data["goal_zone"], bus, name="north goal", confirm_frames=3, cooldown=5.0)
for f in data["frames"]:                                               # in a live system this loop is your detection loop
    detector.update(f["ball"], f["t"])

With the demo data it should print three goals, at about 5.7 s, 12.4 s and 19.1 s, update the score each time, and write goals.csv.

Cell 6: test the trigger

from events import EventBus
from goal_detector import GoalDetector
 
ZONE = {"x": 100, "y": 100, "w": 100, "h": 50}
IN, OUT = {"x": 150, "y": 120}, {"x": 400, "y": 300}
 
 
def run(sequence, **kw):
    """sequence = list of ball positions (or None), one per frame at 30 FPS. Returns the goal times."""
    bus, goals = EventBus(), []
    bus.on("goal", lambda zone, t: goals.append(round(t, 2)))
    d = GoalDetector(ZONE, bus, **kw)
    for i, ball in enumerate(sequence):
        d.update(ball, i / 30)
    return goals
 
 
def check(name, got, expected):
    print(f"{'PASS' if got == expected else 'FAIL'}  {name}: goals at {got} (expected {expected})")
 
 
check("clean goal", run([OUT] * 5 + [IN] * 10), [round(7 / 30, 2)])
check("one stray detection inside the zone", run([OUT] * 5 + [IN] + [OUT] * 5), [])
check("two stray detections inside the zone", run([OUT] * 5 + [IN, IN] + [OUT] * 5), [])
check("ball already resting in the net at the start", run([IN] * 30), [])
check("ball stays in the net after a goal (still one goal)", run([OUT] * 3 + [IN] * 60), [round(5 / 30, 2)])
check("flicker: ball lost and found again, 1 s apart", run([OUT] * 3 + [IN] * 5 + [None] * 3 + [OUT] * 3 + [IN] * 5), [round(5 / 30, 2)])
check("two real goals 6 seconds apart", run([OUT] * 3 + [IN] * 5 + [OUT] * 180 + [IN] * 5), [round(5 / 30, 2), round(190 / 30, 2)])
check("ball hidden for 5 frames half-way into the net, then seen inside again", run([OUT] * 3 + [IN] * 2 + [None] * 5 + [IN] * 5), [round(10 / 30, 2)])

All eight should print PASS. Now add two situations of your own. Predict the answer first.

Questions: what happens if confirm_frames is 1? If the cooldown is 0? A ball that rolls along the goal line but never crosses it: how would you handle it? (Hint: lesson 32.) What if there are two goals on the pitch?

Cell 7: the live version

This connects the same GoalDetector to your YOLO model. The only link between detection and automation is one line: detector.update(ball, time.time() - start). It needs pip install ultralytics, a best.pt and a camera or clip.

import time
 
import cv2
from ultralytics import YOLO
 
from actions import GoalLog, Score, make_goal_sound, play_sound
from events import EventBus
from goal_detector import GoalDetector
 
SOURCE = 0                                        # the webcam, or a video file
GOAL_ZONE = {"x": 540, "y": 20, "w": 200, "h": 90}     # pixels in YOUR frame: lesson 32 shows how to draw this properly
model = YOLO("best.pt")
 
bus = EventBus()
sound = make_goal_sound()
bus.on("goal", GoalLog())
bus.on("goal", Score())
bus.on("goal", lambda zone, t: play_sound(sound))
detector = GoalDetector(GOAL_ZONE, bus, name="north goal")
 
cap = cv2.VideoCapture(SOURCE)
start = time.time()
while True:
    ok, frame = cap.read()
    if not ok:
        break
    result = model(frame, conf=0.3, verbose=False)[0]
    ball = None
    best = 0.0
    for b in result.boxes:                         # take the most confident "ball" in the frame
        if model.names[int(b.cls[0])] == "ball" and float(b.conf[0]) > best:
            x1, y1, x2, y2 = b.xyxy[0].tolist()
            ball, best = {"x": (x1 + x2) / 2, "y": (y1 + y2) / 2}, float(b.conf[0])
    detector.update(ball, time.time() - start)      # <- the only line that connects detection to automation
 
    z = GOAL_ZONE
    cv2.rectangle(frame, (z["x"], z["y"]), (z["x"] + z["w"], z["y"] + z["h"]), (0, 255, 255), 2)
    cv2.imshow("auto goal detector", result.plot())
    if cv2.waitKey(1) & 0xFF == ord("q"):
        break
cap.release()
cv2.destroyAllWindows()

If it goes wrong

Error	Cause	Fix
No goal found with real data	The zone does not match the video, or the ball is rarely detected	Draw the zone on a frame (lesson 32). Check ball detection first.
Goals counted twice	Cooldown too short, or the ball re-enters	Raise the cooldown. Check the `seen_outside` rule.
Goal missed when the net hides the ball	Rules too strict, or hidden frames reset the count	Missing balls are neutral in this design. Lower `confirm_frames` if needed.
No sound	`pygame` missing or no audio device	Install pygame, or accept the bell. Check the volume.
`ModuleNotFoundError: events`	Files not in the same folder	Put all the files together and run from that folder
`goals.csv` keeps growing between runs	Append mode	Delete it for a fresh run, or add a header per session
Live loop is slow	Detection is the bottleneck	Use the lesson 25 speed-ups
Clock mismatch between replay and live	Replay uses video time, live uses wall-clock time	Both are passed in as `t`. Keep that in mind when comparing.

Try it

Add a handler that draws "GOAL!" on the video for two seconds
Add a second goal zone and a score for each side
Write a test for a ball that rolls along the goal line

Check what you learnt

Why use events instead of polling?
What does the cooldown protect against?
Why must the ball be seen outside the goal first?
The ball goes missing for a few frames. What should that mean?
How can you test a trigger without a real match?

Conclusion

Automation means detections cause actions. An event-driven design lets one detector emit an event (like goal) and many listeners react: log it, alert, save a clip. Fire on the change, add a cooldown, treat a missing ball as "don't know", and test with scripted situations before the real match.

What next?

Next: Zone detection & spatial logic. You'll draw zones on the pitch and test who's inside them.