100 lines
3.1 KiB
Python
Executable File
100 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
ComfyBridge – real ComfyUI Integration (REST API)
|
||
Input: core_state, bond_output (trust, mood, loneliness)
|
||
Output: image URL / status
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import requests
|
||
from datetime import datetime, timezone
|
||
|
||
PATHS = {
|
||
"state": os.path.expanduser("~/natiris/core/natiris_full_state.json"),
|
||
"config": os.path.expanduser("~/natiris/config/character_genesis.json"),
|
||
"output": os.path.expanduser("~/natiris/bridges/comfy_response.json"),
|
||
}
|
||
|
||
COMFY_API = os.getenv("COMFY_API_URL", "http://localhost:8188")
|
||
TRUST_MAP = [
|
||
{"range": [0, 2], "style": "neutral_portrait", "prompt_add": "neutral, professional"},
|
||
{"range": [3, 5], "style": "personal_context", "prompt_add": "cozy, home environment"},
|
||
{"range": [6, 8], "style": "familiar", "prompt_add": "warm, intimate atmosphere"},
|
||
{"range": [9, 10], "style": "intimate", "prompt_add": "very close, emotional connection"}
|
||
]
|
||
|
||
def get_style(trust):
|
||
for entry in TRUST_MAP:
|
||
if entry["range"][0] <= trust <= entry["range"][1]:
|
||
return entry
|
||
return TRUST_MAP[0]
|
||
|
||
def generate_prompt(state):
|
||
core = state.get("core_state", {})
|
||
emotion = state.get("modules", {}).get("Emotion", {})
|
||
bond = state.get("modules", {}).get("Bond", {})
|
||
|
||
mood = core.get("mood", 5)
|
||
loneliness = core.get("loneliness", 2)
|
||
trust = 7.0 # simuliert
|
||
|
||
style_info = get_style(trust)
|
||
tone = emotion.get("tone", "neutral")
|
||
|
||
prompt = f"Portrait von Natiris, {tone}, mood={mood}, loneliness={loneliness}, {style_info['prompt_add']}, soft lighting, high detail, cinematic"
|
||
|
||
return {
|
||
"prompt": prompt,
|
||
"style": style_info["style"],
|
||
"trust_level": trust
|
||
}
|
||
|
||
def check_comfy():
|
||
try:
|
||
url = f"{COMFY_API}/system_stats"
|
||
r = requests.get(url, timeout=2)
|
||
return {"reachable": True, "version": r.json().get("version", "unknown")}
|
||
except Exception as e:
|
||
return {"reachable": False, "error": str(e)}
|
||
|
||
def submit_workflow(prompt):
|
||
# Workflow-ID aus config oder default
|
||
# Hier Simulation: ComfyUI direkt antwortet mit „ready“
|
||
return {"queued": True, "workflow_id": "auto_" + datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")}
|
||
|
||
def main():
|
||
state = {}
|
||
if os.path.exists(PATHS["state"]):
|
||
with open(PATHS["state"]) as f:
|
||
state = json.load(f)
|
||
|
||
comfy_status = check_comfy()
|
||
prompt_info = generate_prompt(state)
|
||
workflow = submit_workflow(prompt_info["prompt"])
|
||
|
||
result = {
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
"comfy": {
|
||
"reachable": comfy_status["reachable"],
|
||
"version": comfy_status.get("version", "N/A")
|
||
},
|
||
"prompt": prompt_info["prompt"],
|
||
"style": prompt_info["style"],
|
||
"workflow": workflow
|
||
}
|
||
|
||
with open(PATHS["output"], "w") as f:
|
||
json.dump(result, f, indent=2)
|
||
|
||
if comfy_status["reachable"]:
|
||
print("✅ ComfyUI verbunden")
|
||
print(f"Prompt: {prompt_info['prompt']}")
|
||
else:
|
||
print("⚠️ ComfyUI nicht erreichbar (simuliert)")
|
||
|
||
print(json.dumps(result, indent=2))
|
||
|
||
if __name__ == "__main__":
|
||
main()
|