41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Any
|
|
|
|
|
|
@dataclass
|
|
class Node:
|
|
id: str
|
|
action: str
|
|
inputs: Dict[str, Any] = field(default_factory=dict)
|
|
outputs: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class ExecutionGraph:
|
|
def __init__(self) -> None:
|
|
self.nodes: Dict[str, Node] = {}
|
|
self.edges = [] # list of (src, dst)
|
|
|
|
def add_node(self, node: Node) -> None:
|
|
self.nodes[node.id] = node
|
|
|
|
def add_edge(self, src_id: str, dst_id: str) -> None:
|
|
self.edges.append((src_id, dst_id))
|
|
|
|
def compile_from_ir(self, ir: Dict[str, Any]) -> None:
|
|
# Minimal compilation: create a single allocation node if assets present
|
|
assets = getattr(ir.get("objects"), "assets", {}) if isinstance(ir, dict) else {}
|
|
total = sum(assets.values()) if isinstance(assets, dict) else 0.0
|
|
self.nodes["allocate"] = Node(id="allocate", action="allocate", inputs={"assets_total": total})
|
|
|
|
def run(self, assets: Dict[str, float], target_total: float) -> Dict[str, float]:
|
|
# Simple deterministic allocation: scale assets to meet target_total while preserving ratios
|
|
if not assets:
|
|
return {}
|
|
total = sum(assets.values())
|
|
if total == 0:
|
|
return {k: 0.0 for k in assets}
|
|
scale = target_total / total if target_total is not None else 1.0
|
|
return {k: v * scale for k, v in assets.items()}
|