見出し画像

【観測&予測】にゃんぱすーボタンの1分/1時間/1日のカウント増加予測を表示できるプログラム

にゃんぱすー

読者のみなさま、にゃんぱすー。今回はCUI/GUI両方でにゃんぱすーボタンのカウントを表示/推測するプログラムを作りました。
GUI版のほう、体感重いので定期的に再起動してやってください。
それと、今回から必要なpipをちゃんと記載しようと思います。今まで不便でごめんねー。
まずは、CUI版に必要なpipがこれです。

pip install aiohttp

GUI版がこれです。

pip install aiohttp PySide6 matplotlib

コード、ダウンロード(CUI版)

import asyncio
import aiohttp
import sys
from collections import deque
from datetime import datetime

API_URL = "https://nyanpass.com/api/get_count"
INTERVAL = 0.1

short_history = deque(maxlen=int(60 / INTERVAL))
medium_history = deque(maxlen=int(3600 / INTERVAL))
long_history = deque(maxlen=int(86400 / INTERVAL))

async def fetch_count(session):
    try:
        async with session.get(API_URL) as resp:
            data = await resp.json()
            count = int(data.get("count", 0))
            return count
    except Exception:
        return None

def calculate_rate(history):
    if len(history) < 2:
        return 0
    old_time, old_count = history[0]
    new_time, new_count = history[-1]
    elapsed_sec = (new_time - old_time).total_seconds()
    delta_count = new_count - old_count
    return delta_count / elapsed_sec if elapsed_sec > 0 else 0

async def display_loop():
    async with aiohttp.ClientSession() as session:
        while True:
            count = await fetch_count(session)
            now = datetime.now()
            if count is not None:
                short_history.append((now, count))
                medium_history.append((now, count))
                long_history.append((now, count))

                rate_short = calculate_rate(short_history)
                rate_medium = calculate_rate(medium_history)
                rate_long = calculate_rate(long_history)

                per_minute = rate_short * 60
                per_hour = rate_medium * 3600
                per_day = rate_long * 86400

                sys.stdout.write(
                    f"\rにゃんぱすーされた回数: {count:,}回 | "
                    f"予測: 1分 {per_minute:,.1f}回 / "
                    f"1時間 {per_hour:,.0f}回 / "
                    f"1日 {per_day:,.0f}回"
                )
                sys.stdout.flush()
            await asyncio.sleep(INTERVAL)

if __name__ == "__main__":
    try:
        asyncio.run(display_loop())
    except KeyboardInterrupt:
        print("\n終了しました。")

ダウンロードもこちらでどうぞ。Py版とexe版です。


コード、ダウンロード(GUI版)

import sys
import asyncio
from datetime import datetime
from collections import deque
import platform

import aiohttp
from PySide6.QtWidgets import QApplication, QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import QTimer, Qt
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
from matplotlib.ticker import FuncFormatter
from matplotlib import rcParams

rcParams['font.family'] = 'MS Gothic'

def is_dark_mode_windows():
    try:
        import winreg
        registry = winreg.ConnectRegistry(None, winreg.HKEY_CURRENT_USER)
        key = winreg.OpenKey(registry, r"Software\Microsoft\Windows\CurrentVersion\Themes\Personalize")
        value, _ = winreg.QueryValueEx(key, "AppsUseLightTheme")
        winreg.CloseKey(key)
        return value == 0
    except Exception:
        return False

API_URL = "https://nyanpass.com/api/get_count"
INTERVAL = 0.1

short_history = deque(maxlen=int(60 / INTERVAL))
medium_history = deque(maxlen=int(3600 / INTERVAL))
long_history = deque(maxlen=int(86400 / INTERVAL))

class NyanpassGUI(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("にゃんぱすー回数モニター")
        self.setGeometry(100, 100, 1400, 900)
        layout = QVBoxLayout()

        self.label = QLabel("にゃんぱすー回数: 0回\n予測: 1分 0回 / 1時間 0回 / 1日 0回")
        self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.label.setStyleSheet("font-size: 20px;")
        layout.addWidget(self.label)

        self.figure = Figure(figsize=(14,9))
        self.canvas = FigureCanvas(self.figure)
        layout.addWidget(self.canvas)

        self.ax_short = self.figure.add_subplot(311)
        self.ax_medium = self.figure.add_subplot(312)
        self.ax_long = self.figure.add_subplot(313)

        self.dark_mode = platform.system() == "Windows" and is_dark_mode_windows()
        self.apply_theme()

        self.setup_axes(self.ax_short, "1分平均増加数")
        self.setup_axes(self.ax_medium, "1時間平均増加数")
        self.setup_axes(self.ax_long, "1日平均増加数")

        self.line_short, = self.ax_short.plot([], [], 'g')
        self.line_medium, = self.ax_medium.plot([], [], 'b')
        self.line_long, = self.ax_long.plot([], [], 'orange')

        self.setLayout(layout)

        self.timer = QTimer()
        self.timer.setInterval(int(INTERVAL * 1000))
        self.timer.timeout.connect(self.update_gui)
        self.timer.start()

        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self.session = None
        self.loop.create_task(self.fetch_loop())

    def apply_theme(self):
        if self.dark_mode:
            self.setStyleSheet("background-color: #121212; color: white;")
            self.figure.patch.set_facecolor('#121212')
            self.text_color = 'white'
            self.grid_color = 'gray'
        else:
            self.setStyleSheet("background-color: white; color: black;")
            self.figure.patch.set_facecolor('white')
            self.text_color = 'black'
            self.grid_color = 'gray'

    def setup_axes(self, ax, title):
        ax.set_title(title, fontsize=14, color=self.text_color)
        ax.set_xlabel("取得回数", fontsize=12, color=self.text_color)
        ax.set_ylabel("増加数", fontsize=12, color=self.text_color)
        ax.set_facecolor('#1e1e1e' if self.dark_mode else 'white')
        ax.grid(True, color=self.grid_color)
        ax.tick_params(axis='x', colors=self.text_color)
        ax.tick_params(axis='y', colors=self.text_color)

        def human_format(x, pos):
            if x >= 1_000_000:
                return f"{x/1_000_000:.1f}M"
            elif x >= 1_000:
                return f"{x/1_000:.1f}K"
            else:
                return f"{int(x)}"
        ax.yaxis.set_major_formatter(FuncFormatter(human_format))

    async def fetch_loop(self):
        async with aiohttp.ClientSession() as session:
            self.session = session
            while True:
                await self.fetch_count()
                await asyncio.sleep(INTERVAL)

    async def fetch_count(self):
        try:
            async with self.session.get(API_URL) as resp:
                data = await resp.json()
                count = int(data.get("count", 0))
                now = datetime.now()
                short_history.append((now, count))
                medium_history.append((now, count))
                long_history.append((now, count))
        except Exception:
            pass

    def calculate_average_rate(self, history, scale_seconds=1):
        if len(history) < 2:
            return 0
        delta_count = history[-1][1] - history[0][1]
        delta_time = (history[-1][0] - history[0][0]).total_seconds()
        return (delta_count / delta_time) * scale_seconds if delta_time > 0 else 0

    def update_gui(self):
        if short_history:
            count = short_history[-1][1]

            per_minute = self.calculate_average_rate(short_history, 60)
            per_hour = self.calculate_average_rate(medium_history, 3600)
            per_day = self.calculate_average_rate(long_history, 86400)

            self.label.setText(
                f"にゃんぱすー回数: {count:,}回\n"
                f"予測: 1分 {int(per_minute):,}回 / "
                f"1時間 {int(per_hour):,}回 / "
                f"1日 {int(per_day):,}回"
            )

            y_short = [self.calculate_average_rate(list(short_history)[:i+1], 60) for i in range(len(short_history))]
            y_medium = [self.calculate_average_rate(list(medium_history)[:i+1], 3600) for i in range(len(medium_history))]
            y_long = [self.calculate_average_rate(list(long_history)[:i+1], 86400) for i in range(len(long_history))]

            x_short = list(range(len(y_short)))
            x_medium = list(range(len(y_medium)))
            x_long = list(range(len(y_long)))

            self.line_short.set_data(x_short, y_short)
            self.line_medium.set_data(x_medium, y_medium)
            self.line_long.set_data(x_long, y_long)

            for ax in [self.ax_short, self.ax_medium, self.ax_long]:
                ax.relim()
                ax.autoscale_view()
                ax.tick_params(axis='x', rotation=45)

            self.figure.tight_layout(pad=3.0)
            self.canvas.draw()

    def closeEvent(self, event):
        self.loop.stop()
        event.accept()

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = NyanpassGUI()
    window.show()
    import threading
    t = threading.Thread(target=window.loop.run_forever, daemon=True)
    t.start()
    sys.exit(app.exec())

こちらもダウンロードは以下からどぞ!

容量がデカすぎてexeはギガファイル便じゃないとだめでした。てわけでこっちでどうぞ


いいなと思ったら応援しよう!