Skip to content
Closed
174 changes: 174 additions & 0 deletions hall_of_fame_machine.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
# SPDX-License-Identifier: MIT

from flask import Flask, request, jsonify, render_template_string
import sqlite3
import json
from datetime import datetime

app = Flask(__name__)

DB_PATH = 'rustchain.db'

def init_hall_of_fame_db():
"""Initialize hall_of_fame table if it doesn't exist"""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS hall_of_fame (
fingerprint_hash TEXT PRIMARY KEY,
machine_name TEXT,
first_seen TIMESTAMP,
total_attestations INTEGER DEFAULT 0,
rust_score INTEGER DEFAULT 0,
fleet_rank INTEGER
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS attestations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
fingerprint_hash TEXT,
epoch INTEGER,
rust_score INTEGER,
timestamp TIMESTAMP,
FOREIGN KEY (fingerprint_hash) REFERENCES hall_of_fame(fingerprint_hash)
)
''')
conn.commit()

def get_machine_details(fingerprint_hash):
"""Get detailed machine information"""
init_hall_of_fame_db() # Ensure tables exist

with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

# Get machine basic info
cursor.execute('''
SELECT fingerprint_hash, machine_name, first_seen,
total_attestations, rust_score, fleet_rank
FROM hall_of_fame
WHERE fingerprint_hash = ?
''', (fingerprint_hash,))
machine = cursor.fetchone()

if not machine:
return None

# Get attestation history
cursor.execute('''
SELECT epoch, rust_score, timestamp
FROM attestations
WHERE fingerprint_hash = ?
ORDER BY epoch DESC
LIMIT 50
''', (fingerprint_hash,))
attestation_history = cursor.fetchall()

# Get fleet averages for comparison
cursor.execute('''
SELECT AVG(rust_score) as avg_score,
COUNT(*) as total_machines
FROM hall_of_fame
''')
fleet_stats = cursor.fetchone()

return {
'machine': dict(machine),
'attestation_history': [dict(row) for row in attestation_history],
'fleet_stats': dict(fleet_stats)
}

@app.route('/hall-of-fame/machine')
def machine_detail_page():
machine_id = request.args.get('id')
if not machine_id:
return "Machine ID required", 400

machine_data = get_machine_details(machine_id)
if not machine_data:
return "Machine not found", 404

machine = machine_data['machine']
history = machine_data['attestation_history']
fleet_stats = machine_data['fleet_stats']

# Calculate performance metrics
avg_score = sum(h['rust_score'] for h in history) / len(history) if history else 0

html_template = '''
<!DOCTYPE html>
<html>
<head>
<title>Machine Details - {{ machine.machine_name }}</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; background: #1a1a1a; color: #fff; }
.machine-card { background: #2a2a2a; padding: 20px; border-radius: 8px; margin-bottom: 20px; }
.metric { display: inline-block; margin-right: 20px; }
.history-table { width: 100%; border-collapse: collapse; background: #333; }
.history-table th, .history-table td { padding: 8px; border: 1px solid #555; }
.history-table th { background: #444; }
.back-link { color: #4CAF50; text-decoration: none; }
</style>
</head>
<body>
<a href="/hall-of-fame" class="back-link">← Back to Hall of Fame</a>

<div class="machine-card">
<h1>{{ machine.machine_name or 'Machine-' + machine.fingerprint_hash[:8] }}</h1>
<p><strong>Fingerprint:</strong> {{ machine.fingerprint_hash }}</p>
<p><strong>First Seen:</strong> {{ machine.first_seen }}</p>

<div class="metrics">
<div class="metric">
<strong>Current Rust Score:</strong> {{ machine.rust_score }}
</div>
<div class="metric">
<strong>Fleet Rank:</strong> #{{ machine.fleet_rank }}
</div>
<div class="metric">
<strong>Total Attestations:</strong> {{ machine.total_attestations }}
</div>
</div>
</div>

{% if history %}
<div class="machine-card">
<h2>Recent Attestation History</h2>
<table class="history-table">
<thead>
<tr>
<th>Epoch</th>
<th>Rust Score</th>
<th>Timestamp</th>
</tr>
</thead>
<tbody>
{% for attestation in history %}
<tr>
<td>{{ attestation.epoch }}</td>
<td>{{ attestation.rust_score }}</td>
<td>{{ attestation.timestamp }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% endif %}

<div class="machine-card">
<h2>Fleet Comparison</h2>
<p><strong>Fleet Average Score:</strong> {{ "%.1f"|format(fleet_stats.avg_score or 0) }}</p>
<p><strong>Total Fleet Machines:</strong> {{ fleet_stats.total_machines }}</p>
</div>
</body>
</html>
'''

return render_template_string(html_template,
machine=machine,
history=history,
fleet_stats=fleet_stats)

if __name__ == '__main__':
app.run(debug=True)
91 changes: 91 additions & 0 deletions hall_of_fame_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
# SPDX-License-Identifier: MIT

import sqlite3
import hashlib
from datetime import datetime, timedelta
import json

DB_PATH = 'rustchain.db'

def get_machine_details(machine_id):
"""Get comprehensive machine details for profile page."""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()

# Get machine basic info
cursor.execute("""
SELECT fingerprint_hash, nickname, first_seen, last_seen,
total_attestations, rust_score, uptime_percentage
FROM machines
WHERE fingerprint_hash = ? OR id = ?
""", (machine_id, machine_id))

machine_data = cursor.fetchone()
if not machine_data:
return None

fingerprint_hash, nickname, first_seen, last_seen, total_attestations, rust_score, uptime_pct = machine_data

# Get recent attestation history
cursor.execute("""
SELECT epoch, timestamp, rust_score, block_height
FROM attestations
WHERE machine_id = (SELECT id FROM machines WHERE fingerprint_hash = ?)
ORDER BY epoch DESC LIMIT 100
""", (fingerprint_hash,))

attestation_history = cursor.fetchall()

# Get machine specs if available
cursor.execute("""
SELECT cpu_model, ram_gb, storage_gb, os_info
FROM machine_specs
WHERE machine_id = (SELECT id FROM machines WHERE fingerprint_hash = ?)
""", (fingerprint_hash,))

specs = cursor.fetchone()

return {
'fingerprint_hash': fingerprint_hash,
'nickname': nickname or f"Machine-{fingerprint_hash[:8]}",
'first_seen': first_seen,
'last_seen': last_seen,
'total_attestations': total_attestations,
'rust_score': rust_score,
'uptime_percentage': uptime_pct,
'attestation_history': attestation_history,
'specs': specs
}

def calculate_machine_rank(fingerprint_hash):
"""Calculate machine's rank in the fleet by rust score."""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()

cursor.execute("""
SELECT COUNT(*) + 1 as rank
FROM machines m1
WHERE m1.rust_score > (
SELECT m2.rust_score FROM machines m2
WHERE m2.fingerprint_hash = ?
)
""", (fingerprint_hash,))

result = cursor.fetchone()
return result[0] if result else None

def get_machine_performance_trend(fingerprint_hash, days=30):
"""Get machine performance trend over specified days."""
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()

cursor.execute("""
SELECT DATE(timestamp) as date, AVG(rust_score) as avg_score
FROM attestations
WHERE fingerprint_hash = ?
AND timestamp >= datetime('now', '-{} days')
GROUP BY DATE(timestamp)
ORDER BY date
""".format(days), (fingerprint_hash,))

return cursor.fetchall()
112 changes: 112 additions & 0 deletions tests/test_hall_of_fame_machine.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
# SPDX-License-Identifier: MIT

import unittest
import sqlite3
import tempfile
import os
import sys
from unittest.mock import patch, MagicMock

# Add the parent directory to the path to import the app
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from hall_of_fame_machine import app, DB_PATH

class TestHallOfFameMachine(unittest.TestCase):

def setUp(self):
"""Set up test database and Flask test client."""
self.db_fd, self.test_db_path = tempfile.mkstemp()
app.config['TESTING'] = True
self.client = app.test_client()

# Mock the DB_PATH to use our test database
self.original_db_path = DB_PATH
import hall_of_fame_machine
hall_of_fame_machine.DB_PATH = self.test_db_path

# Create test database with sample data
with sqlite3.connect(self.test_db_path) as conn:
cursor = conn.cursor()

# Create hall_of_fame table
cursor.execute('''
CREATE TABLE hall_of_fame (
fingerprint_hash TEXT PRIMARY KEY,
machine_name TEXT,
first_seen TIMESTAMP,
total_attestations INTEGER DEFAULT 0,
rust_score INTEGER DEFAULT 0,
fleet_rank INTEGER
)
''')

# Create attestations table
cursor.execute('''
CREATE TABLE attestations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
fingerprint_hash TEXT,
epoch INTEGER,
rust_score INTEGER,
timestamp TIMESTAMP,
FOREIGN KEY (fingerprint_hash) REFERENCES hall_of_fame(fingerprint_hash)
)
''')

# Insert test data
cursor.execute('''
INSERT INTO hall_of_fame VALUES
('abc123def456', 'Mining Rig Alpha', '2024-01-15 10:30:00', 142, 8750, 1),
('def789ghi012', 'Beta Node', '2024-02-01 09:15:00', 98, 7200, 2),
('ghi345jkl678', 'Gamma Processor', '2023-12-01 08:00:00', 201, 9100, 3)
''')

# Insert attestation data
cursor.execute('''
INSERT INTO attestations (fingerprint_hash, epoch, rust_score, timestamp) VALUES
('abc123def456', 1001, 8750, '2024-03-10 14:22:00'),
('abc123def456', 1000, 8600, '2024-03-09 14:22:00'),
('def789ghi012', 999, 7200, '2024-02-28 16:45:00')
''')

conn.commit()

def tearDown(self):
"""Clean up test database."""
# Restore original DB_PATH
import hall_of_fame_machine
hall_of_fame_machine.DB_PATH = self.original_db_path

os.close(self.db_fd)
os.unlink(self.test_db_path)

def test_machine_detail_page_success(self):
"""Test machine detail page for existing machine."""
response = self.client.get('/hall-of-fame/machine?id=abc123def456')
self.assertEqual(response.status_code, 200)
self.assertIn(b'Mining Rig Alpha', response.data)
self.assertIn(b'abc123def456', response.data)
self.assertIn(b'8750', response.data) # rust score

def test_machine_detail_page_nonexistent(self):
"""Test machine detail page for nonexistent machine."""
response = self.client.get('/hall-of-fame/machine?id=nonexistent')
self.assertEqual(response.status_code, 404)
self.assertIn(b'Machine not found', response.data)

def test_machine_detail_page_no_id(self):
"""Test machine detail page without machine ID."""
response = self.client.get('/hall-of-fame/machine')
self.assertEqual(response.status_code, 400)
self.assertIn(b'Machine ID required', response.data)

def test_machine_detail_page_with_history(self):
"""Test machine detail page includes attestation history."""
response = self.client.get('/hall-of-fame/machine?id=abc123def456')
self.assertEqual(response.status_code, 200)
self.assertIn(b'Recent Attestation History', response.data)
self.assertIn(b'1001', response.data) # epoch
self.assertIn(b'1000', response.data) # previous epoch

if __name__ == '__main__':
unittest.main()