#!/usr/bin/env python3 import os import tempfile import threading from flask import Flask, jsonify, request, abort, render_template_string app = Flask(__name__) # --- Configuration --- VALUE_FILE = os.environ.get("VALUE_FILE", "value.txt") POLL_INTERVAL_MS = int(os.environ.get("POLL_INTERVAL_MS", "20000")) # client refresh rate _write_lock = threading.Lock() # --- Helpers --- def _clamp_to_range(val: int) -> int: return max(0, min(99, int(val))) def _ensure_file_exists(): """Create the value file with default '0' if missing or invalid.""" if not os.path.exists(VALUE_FILE): _atomic_write("0") return try: _ = _read_value() except Exception: _atomic_write("0") def _read_value() -> int: """Read the current numeric value from disk.""" with open(VALUE_FILE, "r", encoding="utf-8") as f: raw = f.read().strip() # Allow blank/new files to recover to 0 if raw == "": return 0 n = int(raw) return _clamp_to_range(n) def _atomic_write(text: str): """Atomically write text to VALUE_FILE.""" dir_name = os.path.dirname(os.path.abspath(VALUE_FILE)) or "." os.makedirs(dir_name, exist_ok=True) fd, tmp_path = tempfile.mkstemp(prefix=".value-", dir=dir_name, text=True) try: with os.fdopen(fd, "w", encoding="utf-8") as tmp: tmp.write(text) tmp.flush() os.fsync(tmp.fileno()) os.replace(tmp_path, VALUE_FILE) # atomic on POSIX/Windows finally: # If replace succeeded, tmp_path no longer exists; ignore errors. try: if os.path.exists(tmp_path): os.remove(tmp_path) except Exception: pass def _get_mtime() -> float: try: return os.path.getmtime(VALUE_FILE) except FileNotFoundError: return 0.0 # --- Routes --- @app.route("/", methods=["GET"]) def index(): _ensure_file_exists() val = _read_value() mtime = _get_mtime() # Single-file template for convenience html = """ Tiny Value Server

Tiny Value Server

Stores a single integer 0–99 in {{value_file}}. Updates live if the file changes.
{{value}}
last updated: {{mtime_human}}
Auto-refresh every {{poll_ms}} ms.
""" return render_template_string( html, value_file=os.path.abspath(VALUE_FILE), value=val, mtime_human=_fmt_time(mtime), poll_ms=POLL_INTERVAL_MS, ) @app.route("/value", methods=["GET"]) def get_value(): _ensure_file_exists() try: val = _read_value() except Exception: # If the file is corrupt or unreadable, reset to 0 with _write_lock: _atomic_write("0") val = 0 mtime = _get_mtime() return jsonify({"value": val, "mtime": mtime, "mtime_human": _fmt_time(mtime)}) @app.route("/set", methods=["POST"]) def set_value(): if not request.is_json: abort(400, "Expected application/json body") body = request.get_json(silent=True) or {} if "value" not in body: abort(400, "Missing 'value'") try: new_val = _clamp_to_range(int(body["value"])) except Exception: abort(400, "Value must be an integer 0–99") with _write_lock: _atomic_write(str(new_val)) mtime = _get_mtime() return jsonify({"ok": True, "value": new_val, "mtime": mtime, "mtime_human": _fmt_time(mtime)}) # --- Utilities --- import time def _fmt_time(mtime: float) -> str: if not mtime: return "never" return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mtime)) # --- Entry point --- if __name__ == "__main__": _ensure_file_exists() app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8087")), debug=True)