minecraft watcher

readme.adoc

Minecraft Watcher

Periodically checks a Minecraft server for changes in online players, logs those changes, and sends desktop notifications (using libnotify) for those changes.

Setup

The following setup assumes python-gobject (Python 3) is installed by your package manager. This dependency can be satisified by pip install PyGObject from within the venv, but compile issues might occur (and is outside the scope of this article).

python3 -m venv ~/.venv/mc/ --system-site-packages
source ~/.venv/mc/bin/activate
pip install mcstatus

Update the address in the main() function to point to the server you want to monitor.

Execution

You need to execute this script with the Python from the venv you created, e.g.:

~/.venv/mc/python watch_mc.py

The script will keep running until terminated by Ctrl-C or some other means. You may want to use a terminal multiplexer to keep it running long term (or just & it if you don’t care about output).

tmux new -n mc_watch -d 'cd ~/script/directory; ~/.venv/mc/bin/python watch_mc.py'

Logs

By default, logs are generated per day in the running directory at minecraft_watch-$time.json where $time is the timestamp of the first second of the day.

watch_mc.py

#!/usr/bin/env python
import os
import time
import datetime as dt
import json
from mcstatus import MinecraftServer
from gi.repository import Notify

def sleep_until_next_min():
    now = dt.datetime.now()
    sec_until_next_min = 60 - (now.second + now.microsecond/1000000.0)
    time.sleep(sec_until_next_min)

def generate_diff_events(timestamp, old, new):
    events = []

    for player in new:
        if player not in old:
            events.append({
                "type": "player_join",
                "player_name": player,
                "timestamp": timestamp
            })

    for player in old:
        if player not in new:
            events.append({
                "type": "player_quit",
                "player_name": player,
                "timestamp": timestamp
            })

    return events

def print_event(entry, notify=False):
    event_time = time.strftime('%m.%d %X', time.localtime(entry["timestamp"]/1000))

    if entry["type"] == "player_join":
        message = "{} joined.".format(entry["player_name"])
    elif entry["type"] == "player_quit":
        message = "{} quit.".format(entry["player_name"])
    elif entry["type"] == "player_present":
        message = "{} was already present when log was opened.".format(entry["player_name"])
    elif entry["type"] == "log_open":
        message = "Log opened."
    elif entry["type"] == "log_close":
        message = "Log closed."

    print("({}) {}".format(event_time, message))

    if notify:
        Notify.Notification.new("Minecraft: {}".format(message)).show()

def main():
    address = "localhost"
    log_prefix = "minecraft_watch"

    os.environ['TZ'] = "America/Los_Angeles"
    time.tzset()

    server = MinecraftServer(address)
    Notify.init("Minecraft Watcher")

    while True:
        today = dt.date.today()
        today_start_sec = today.strftime('%s')
        tomorrow = today + dt.timedelta(days=1)

        try:
            with open('{}_{}.json'.format(log_prefix, today_start_sec), 'r') as f:
                events = json.load(f)
                for entry in events:
                    print_event(entry)
        except FileNotFoundError:
            events = []

        ts = int(dt.datetime.now().timestamp())
        events.append({"type": "log_open", "timestamp": ts})

        active_players = []
        status = server.status()
        if status.players.sample is not None:
            active_players = [player.name for player in status.players.sample]
            for player in active_players:
                events.append({
                    "type": "player_present",
                    "player_name": player,
                    "timestamp": ts
                })
                print_event(events[-1], True)

        sleep_until_next_min()

        while dt.date.today() < tomorrow:
            ts = int(dt.datetime.now().timestamp())
            status = server.status()
            player_update = []
            if status.players.sample is not None:
                player_update = [player.name for player in status.players.sample]

            new_events = generate_diff_events(ts, active_players, player_update)
            if new_events:
                for entry in new_events:
                    events.append(entry)
                    print_event(entry, True)

            active_players = player_update

            with open('{}_{}.json'.format(log_prefix, today_start_sec), 'w+') as f:
                f.write(json.dumps(events + [{"type": "log_close", "timestamp": ts}]))

            sleep_until_next_min()


if __name__ == "__main__":
    main()