62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
|
from typing import Dict, List, Any, Optional
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Segment:
|
||
|
|
segment_id: str
|
||
|
|
code: str
|
||
|
|
namespace: Dict[str, Any]
|
||
|
|
|
||
|
|
|
||
|
|
class SegmentRuntimeError(Exception):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class SegmentRuntime:
|
||
|
|
@staticmethod
|
||
|
|
def stitch_segments(segments: List[Segment]) -> str:
|
||
|
|
if not segments:
|
||
|
|
raise SegmentRuntimeError("No segments to stitch")
|
||
|
|
|
||
|
|
lines = []
|
||
|
|
for segment in segments:
|
||
|
|
lines.append(f"# --- Segment {segment.segment_id} ---")
|
||
|
|
lines.append(segment.code)
|
||
|
|
lines.append("")
|
||
|
|
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def merge_namespaces(namespaces: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
|
|
merged = {}
|
||
|
|
for ns in namespaces:
|
||
|
|
for key, value in ns.items():
|
||
|
|
if key in merged and merged[key] != value:
|
||
|
|
raise SegmentRuntimeError(f"Namespace conflict on key: {key}")
|
||
|
|
merged[key] = value
|
||
|
|
return merged
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def execute_segments(segments: List[Segment], debug: bool = False) -> Dict[str, Any]:
|
||
|
|
if not segments:
|
||
|
|
raise SegmentRuntimeError("No segments to execute")
|
||
|
|
|
||
|
|
merged_ns = SegmentRuntime.merge_namespaces([s.namespace for s in segments])
|
||
|
|
stitched_code = SegmentRuntime.stitch_segments(segments)
|
||
|
|
|
||
|
|
globals_dict = {
|
||
|
|
"__name__": "__main__",
|
||
|
|
"__segments__": [s.segment_id for s in segments],
|
||
|
|
}
|
||
|
|
globals_dict.update(merged_ns)
|
||
|
|
|
||
|
|
try:
|
||
|
|
exec(compile(stitched_code, "<segments>", "exec"), globals_dict)
|
||
|
|
except Exception as e:
|
||
|
|
if debug:
|
||
|
|
raise
|
||
|
|
raise SegmentRuntimeError(f"Execution failed: {e}")
|
||
|
|
|
||
|
|
return globals_dict
|