154 lines
6.5 KiB
Python
154 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Character Status Node
|
|
|
|
This node represents the character's status. It has seven output ports:
|
|
- HP: Current, HP: Total, MP: Current, MP: Total, FP: Current, FP: Total, EXP.
|
|
It polls an API endpoint (http://127.0.0.1:5000/data) every 500 ms to update its values.
|
|
If the API call is successful, the node's title is set to "Character Status (API Connected)".
|
|
If the API is down or returns an error, the title is set to "Character Status (API Disconnected)".
|
|
"""
|
|
|
|
from NodeGraphQt import BaseNode
|
|
from Qt import QtCore, QtGui
|
|
import requests
|
|
import traceback
|
|
|
|
def get_draw_stat_port(color, border_color=None, alpha=127):
|
|
"""
|
|
Returns a custom port painter function that draws a circular port with a
|
|
semi-transparent fill and then draws text (port label and current value)
|
|
next to it.
|
|
"""
|
|
if border_color is None:
|
|
border_color = color
|
|
|
|
def painter_func(painter, rect, info):
|
|
painter.save()
|
|
pen = QtGui.QPen(QtGui.QColor(*border_color))
|
|
pen.setWidth(1.8)
|
|
painter.setPen(pen)
|
|
semi_transparent_color = QtGui.QColor(color[0], color[1], color[2], alpha)
|
|
painter.setBrush(semi_transparent_color)
|
|
painter.drawEllipse(rect)
|
|
port = info.get('port')
|
|
if port is not None:
|
|
node = port.node()
|
|
stat = port.name()
|
|
value = node.values.get(stat, "N/A") if hasattr(node, 'values') else "N/A"
|
|
text_rect = rect.adjusted(rect.width() + 4, 0, rect.width() + 70, 0)
|
|
painter.setPen(QtGui.QColor(0, 0, 0))
|
|
painter.drawText(text_rect, QtCore.Qt.AlignVCenter | QtCore.Qt.AlignLeft,
|
|
f"{stat}: {value}")
|
|
painter.restore()
|
|
return painter_func
|
|
|
|
class CharacterStatusNode(BaseNode):
|
|
__identifier__ = 'io.github.nicole.status'
|
|
NODE_NAME = 'Character Status'
|
|
|
|
def __init__(self):
|
|
super(CharacterStatusNode, self).__init__()
|
|
|
|
# Define exact expected keys to avoid transformation mismatches
|
|
self.values = {
|
|
"HP: Current": "N/A", "HP: Total": "N/A",
|
|
"MP: Current": "N/A", "MP: Total": "N/A",
|
|
"FP: Current": "N/A", "FP: Total": "N/A",
|
|
"EXP": "N/A"
|
|
}
|
|
|
|
# Add output ports
|
|
self.add_output("HP: Current", painter_func=get_draw_stat_port((217, 36, 78)))
|
|
self.add_output("HP: Total", painter_func=get_draw_stat_port((217, 36, 78)))
|
|
self.add_output("MP: Current", painter_func=get_draw_stat_port((35, 124, 213)))
|
|
self.add_output("MP: Total", painter_func=get_draw_stat_port((35, 124, 213)))
|
|
self.add_output("FP: Current", painter_func=get_draw_stat_port((36, 197, 28)))
|
|
self.add_output("FP: Total", painter_func=get_draw_stat_port((36, 197, 28)))
|
|
self.add_output("EXP", painter_func=get_draw_stat_port((52, 195, 250)))
|
|
|
|
self.set_name("Character Status (API Disconnected)")
|
|
|
|
# Start polling timer
|
|
self.timer = QtCore.QTimer()
|
|
self.timer.timeout.connect(self.poll_api)
|
|
self.timer.start(500)
|
|
|
|
def poll_api(self):
|
|
"""
|
|
Polls the API endpoint to retrieve the latest character stats and updates
|
|
the node's internal values.
|
|
"""
|
|
try:
|
|
response = requests.get("http://127.0.0.1:5000/data", timeout=1)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
if isinstance(data, dict):
|
|
try:
|
|
for key, value in data.items():
|
|
# Ensure the keys match the expected ones exactly
|
|
formatted_key = {
|
|
"hp_current": "HP: Current",
|
|
"hp_total": "HP: Total",
|
|
"mp_current": "MP: Current",
|
|
"mp_total": "MP: Total",
|
|
"fp_current": "FP: Current",
|
|
"fp_total": "FP: Total",
|
|
"exp": "EXP"
|
|
}.get(key, key) # Use mapping or fallback to raw key
|
|
|
|
if formatted_key in self.values:
|
|
self.values[formatted_key] = str(value)
|
|
else:
|
|
print(f"[WARNING] Unexpected API key: {key} (not mapped)")
|
|
|
|
self.set_name("Character Status (API Connected)")
|
|
self.update()
|
|
self.transmit_data()
|
|
|
|
except Exception as e:
|
|
print("[ERROR] Error processing API response data:", e)
|
|
print("[ERROR] Stack Trace:\n", traceback.format_exc())
|
|
else:
|
|
print("[ERROR] Unexpected API response format (not a dict):", data)
|
|
self.set_name("Character Status (API Disconnected)")
|
|
|
|
else:
|
|
print(f"[ERROR] API request failed with status code {response.status_code}")
|
|
self.set_name("Character Status (API Disconnected)")
|
|
|
|
except Exception as e:
|
|
self.set_name("Character Status (API Disconnected)")
|
|
print("[ERROR] Error polling API in CharacterStatusNode:", str(e))
|
|
print("[ERROR] Stack Trace:\n", traceback.format_exc())
|
|
|
|
def transmit_data(self):
|
|
"""
|
|
Sends the updated character stats to connected nodes.
|
|
"""
|
|
for stat, value in self.values.items():
|
|
try:
|
|
port = self.get_output(stat)
|
|
|
|
if port is None:
|
|
print(f"[ERROR] Port '{stat}' not found in node outputs. Skipping...")
|
|
continue
|
|
|
|
if port.connected_ports():
|
|
for connected_port in port.connected_ports():
|
|
connected_node = connected_port.node()
|
|
if hasattr(connected_node, 'receive_data'):
|
|
try:
|
|
connected_node.receive_data(value, stat)
|
|
except Exception as e:
|
|
print(f"[ERROR] Error transmitting data to {connected_node}: {e}")
|
|
print("[ERROR] Stack Trace:\n", traceback.format_exc())
|
|
else:
|
|
print(f"[WARNING] Connected node {connected_node} does not have receive_data method.")
|
|
|
|
except Exception as e:
|
|
print(f"[ERROR] Error while handling port {stat}: {e}")
|
|
print("[ERROR] Stack Trace:\n", traceback.format_exc())
|