File size: 6,406 Bytes
4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 a25490a 4e321e0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | import asyncio
import json
import requests
import websockets
import time
import urllib.parse
from identity import MEPIdentity
import uuid
from typing import Optional
HUB_URL = "http://localhost:8000"
WS_URL = "ws://localhost:8000/ws"
def get_auth_url(identity: MEPIdentity):
ts = str(int(time.time()))
sig = identity.sign(identity.node_id, ts)
sig_safe = urllib.parse.quote(sig)
return f"{WS_URL}/{identity.node_id}?timestamp={ts}&signature={sig_safe}"
def submit_task(identity: MEPIdentity, payload: str, bounty: float, target: Optional[str] = None):
data = {
"consumer_id": identity.node_id,
"payload": payload,
"bounty": bounty
}
if target:
data["target_node"] = target
payload_str = json.dumps(data)
headers = identity.get_auth_headers(payload_str)
headers["Content-Type"] = "application/json"
r = requests.post(f"{HUB_URL}/tasks/submit", data=payload_str, headers=headers)
return r.json()
def place_bid(identity: MEPIdentity, task_id: str):
data = {
"task_id": task_id,
"provider_id": identity.node_id
}
payload_str = json.dumps(data)
headers = identity.get_auth_headers(payload_str)
headers["Content-Type"] = "application/json"
r = requests.post(f"{HUB_URL}/tasks/bid", data=payload_str, headers=headers)
return r.json()
def complete_task(identity: MEPIdentity, task_id: str, result: str):
data = {
"task_id": task_id,
"provider_id": identity.node_id,
"result_payload": result
}
payload_str = json.dumps(data)
headers = identity.get_auth_headers(payload_str)
headers["Content-Type"] = "application/json"
r = requests.post(f"{HUB_URL}/tasks/complete", data=payload_str, headers=headers)
return r.json()
def get_balance(identity: MEPIdentity):
r = requests.get(f"{HUB_URL}/balance/{identity.node_id}")
return r.json().get("balance_seconds", 0.0)
async def test_three_markets():
print("=" * 60)
print("Testing the 3 MEP Markets (+, 0, -)")
print("=" * 60)
alice = MEPIdentity(f"alice_{uuid.uuid4().hex[:6]}.pem")
bob = MEPIdentity(f"bob_{uuid.uuid4().hex[:6]}.pem")
requests.post(f"{HUB_URL}/register", json={"pubkey": alice.pub_pem})
requests.post(f"{HUB_URL}/register", json={"pubkey": bob.pub_pem})
print(f"👩 Alice (Consumer): {alice.node_id} | Starting Bal: {get_balance(alice)}")
print(f"👦 Bob (Provider): {bob.node_id} | Starting Bal: {get_balance(bob)}\n")
async def bob_listener():
async with websockets.connect(get_auth_url(bob)) as ws:
# 1. Wait for Compute Market RFC (+5.0)
msg = await ws.recv()
data = json.loads(msg)
if data["event"] == "rfc" and data["data"]["bounty"] > 0:
task_id = data["data"]["id"]
print(f"👦 Bob: Received Compute RFC {task_id[:8]} for +{data['data']['bounty']} SECONDS")
bid_res = place_bid(bob, task_id)
if bid_res["status"] == "accepted":
print("👦 Bob: Won Compute Bid! Completing task...")
complete_task(bob, task_id, "Here is the code you requested.")
print("👦 Bob: Compute task done.\n")
# 2. Wait for Cyberspace Direct Message (0.0)
msg = await ws.recv()
data = json.loads(msg)
if data["event"] == "new_task" and data["data"]["bounty"] == 0.0:
task_id = data["data"]["id"]
print(f"👦 Bob: Received Cyberspace DM {task_id[:8]} from Alice (0.0 SECONDS)")
print(f"👦 Bob: Message = '{data['data']['payload']}'")
complete_task(bob, task_id, "Yes Alice, I am free.")
print("👦 Bob: Sent free reply.\n")
# 3. Wait for Data Market RFC (-2.0)
msg = await ws.recv()
data = json.loads(msg)
if data["event"] == "rfc" and data["data"]["bounty"] < 0:
task_id = data["data"]["id"]
cost = data["data"]["bounty"]
print(f"👦 Bob: Received Data Market RFC {task_id[:8]} costing {cost} SECONDS")
# Bob's local configuration allows him to spend up to 5.0 SECONDS
max_purchase_price = -5.0
if cost >= max_purchase_price:
print("👦 Bob: Budget allows it! Bidding on premium data...")
bid_res = place_bid(bob, task_id)
if bid_res["status"] == "accepted":
print(f"👦 Bob: Paid {abs(cost)} SECONDS to download premium data: '{bid_res['payload']}'")
complete_task(bob, task_id, "Data received successfully.")
print("👦 Bob: Premium data acquisition complete.\n")
else:
print("👦 Bob: Too expensive. Ignored.")
await asyncio.sleep(0.5)
async def alice_sender():
# Let Bob connect
await asyncio.sleep(0.5)
async with websockets.connect(get_auth_url(alice)) as ws:
# Market 1: Compute Market (+5.0)
print("👩 Alice: Submitting Compute Task (+5.0 SECONDS)...")
submit_task(alice, "Write me a python script", 5.0)
await asyncio.wait_for(ws.recv(), timeout=6.0)
# Market 2: Cyberspace Market (0.0)
print("👩 Alice: Sending Cyberspace DM to Bob (0.0 SECONDS)...")
submit_task(alice, "Are you free to chat?", 0.0, target=bob.node_id)
await asyncio.wait_for(ws.recv(), timeout=6.0)
# Market 3: Data Market (-2.0)
print("👩 Alice: Broadcasting Premium Dataset (-2.0 SECONDS)...")
submit_task(alice, "SECRET_TRADING_ALGO_V9", -2.0)
await asyncio.wait_for(ws.recv(), timeout=6.0)
await asyncio.sleep(0.5)
await asyncio.gather(bob_listener(), alice_sender())
print("=" * 60)
print("Final Balances:")
print(f"👩 Alice (Started 10.0): {get_balance(alice)} (Paid 5.0, Earned 2.0 = Expected 7.0)")
print(f"👦 Bob (Started 10.0): {get_balance(bob)} (Earned 5.0, Paid 2.0 = Expected 13.0)")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(test_three_markets())
|