Python logger

Create meter_logger.py:

#!/usr/bin/env python3
import csv
import os
import signal
import sys
import time
from datetime import datetime, timezone

import RPi.GPIO as GPIO

GPIO_PIN = 17
CSV_PATH = os.path.expanduser("~/utility-meter-logger/meter_log.csv")
STATE_PATH = os.path.expanduser("~/utility-meter-logger/state.txt")

# Example: 1000 pulses per kWh for an electricity meter.
# Change this value to match your utility meter.
PULSES_PER_UNIT = 1000.0
UNIT_NAME = "kWh"

# Debounce time in milliseconds. Increase if your contact bounces.
BOUNCE_MS = 50

running = True
pulse_count = 0


def load_state(path: str) -> int:
    if not os.path.exists(path):
        return 0
    with open(path, "r", encoding="utf-8") as f:
        text = f.read().strip()
    if text == "":
        return 0
    return int(text)


def save_state(path: str, count: int) -> None:
    tmp_path = path + ".tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        f.write(f"{count}\n")
    os.replace(tmp_path, path)


def ensure_csv_header(path: str) -> None:
    if os.path.exists(path) and os.path.getsize(path) > 0:
        return
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(
            ["timestamp_utc", "pulse_count_total", "delta_pulses", "consumption_total", "unit"]
        )


def append_csv(path: str, count: int, delta: int) -> None:
    consumption = count / PULSES_PER_UNIT
    timestamp = datetime.now(timezone.utc).isoformat()
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([timestamp, count, delta, f"{consumption:.6f}", UNIT_NAME])


def handle_pulse(channel: int) -> None:
    global pulse_count
    pulse_count += 1
    save_state(STATE_PATH, pulse_count)
    append_csv(CSV_PATH, pulse_count, 1)
    print(
        f"{datetime.now(timezone.utc).isoformat()} pulse={pulse_count} "
        f"consumption={pulse_count / PULSES_PER_UNIT:.6f} {UNIT_NAME}",
        flush=True,
    )


def handle_signal(signum, frame) -> None:
    global running
    running = False


def main() -> int:
    global pulse_count

    pulse_count = load_state(STATE_PATH)
    ensure_csv_header(CSV_PATH)

    GPIO.setmode(GPIO.BCM)
    GPIO.setup(GPIO_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)

    signal.signal(signal.SIGINT, handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)

    GPIO.add_event_detect(
        GPIO_PIN,
        GPIO.FALLING,
        callback=handle_pulse,
        bouncetime=BOUNCE_MS,
    )

    print(
        f"Utility meter pulse logger started on GPIO{GPIO_PIN}. "
        f"Current total pulses: {pulse_count}",
        flush=True,
    )

    try:
        while running:
            time.sleep(1.0)
    finally:
        GPIO.cleanup()

    return 0


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

Make it executable:

chmod +x ~/utility-meter-logger/meter_logger.py