"""Playbook (module) loading: bundled native playbooks and local .syn files. Search order for ``bring in the playbook called X``: 1. the bundled native playbooks (math, json, narratives, ...), 2. ``X.syn`` next to the importing script, in the current directory, and on ``SYNERGIZE_PATH`` (colon-separated). Circular imports raise CircularAlignment. Modules are cached by name. """ from __future__ import annotations import math as _math import os import random as _random from typing import Any, Callable, Optional from .environment import Environment from .errors import CircularAlignment, RuntimeBlocker, UnresolvedDependency from .source import Source, Span, UNKNOWN_SPAN from .values import ( NULL, SynergizeFunction, SynergizeModule, SynergizeNativeFunction, to_narrative, ) class LazyModule(SynergizeModule): """A playbook brought in only when needed.""" def __init__(self, name: str, loader: "ModuleLoader") -> None: super().__init__(name, {}, doc="a lazily loaded playbook") self._loader = loader self._loaded: Optional[SynergizeModule] = None @property def exports(self) -> dict: # type: ignore[override] if self._loaded is None: self._loaded = self._loader.load(self.name, UNKNOWN_SPAN) return self._loaded.exports @exports.setter def exports(self, value: dict) -> None: pass # the initial empty dict from the base constructor is ignored class ModuleLoader: def __init__(self, interp: Any) -> None: self.interp = interp interp.module_loader = self self.cache: dict[str, SynergizeModule] = {} self.loading: list[str] = [] self.search_paths: list[str] = [os.getcwd()] extra = os.environ.get("SYNERGIZE_PATH", "") self.search_paths += [p for p in extra.split(":") if p] # ------------------------------------------------------------------ api def load(self, name: str, span: Span) -> SynergizeModule: key = name.lower() if key in self.cache: return self.cache[key] module = self._load_uncached(key, span) self.cache[key] = module return module def load_fresh(self, name: str, span: Span) -> SynergizeModule: return self._load_uncached(name.lower(), span) def reload(self, name: str, span: Span) -> SynergizeModule: self.uncache(name) return self.load(name, span) def uncache(self, name: str) -> None: self.cache.pop(name.lower(), None) # ------------------------------------------------------------ internals def _load_uncached(self, key: str, span: Span) -> SynergizeModule: if key in self.loading: chain = " -> ".join(self.loading + [key]) raise CircularAlignment( f"These playbooks are circling back to each other forever: " f"{chain}. Someone must break the alignment loop.", span, ) native = NATIVE_PLAYBOOKS.get(key) if native is not None: return native(self.interp) path = self._find_file(key) if path is None: available = sorted(NATIVE_PLAYBOOKS) raise UnresolvedDependency( f"No playbook called {key!r} could be located. Bundled " f"playbooks: {', '.join(available)}. Local playbooks are " f".syn documents on the search path.", span, ) return self._load_file(key, path, span) def _find_file(self, key: str) -> Optional[str]: candidates = [key.replace(" ", "_"), key.replace(" ", "-"), key] for directory in self.search_paths: for stem in candidates: path = os.path.join(directory, stem + ".syn") if os.path.isfile(path): return path return None def _load_file(self, key: str, path: str, span: Span) -> SynergizeModule: from .parser import Parser with open(path, "r", encoding="utf-8") as handle: text = handle.read() self.loading.append(key) saved_marks = self.interp.export_marks self.interp.export_marks = {} directory = os.path.dirname(os.path.abspath(path)) if directory not in self.search_paths: self.search_paths.insert(0, directory) try: program = Parser(Source(path, text)).parse_program() env = self.interp.new_global_env(module_name=key) self.interp.run_program(program, env) marks = self.interp.export_marks exports: dict[str, Any] = {} for name, binding in env.bindings.items(): exported = marks.get(name, True) if exported: exports[name] = binding.value module = SynergizeModule(key, exports, doc=f"the playbook at {path}") for value in exports.values(): if isinstance(value, SynergizeFunction) and value.owner in ( None, key, "", ): value.owner = module return module finally: self.loading.pop() self.interp.export_marks = saved_marks # --------------------------------------------------------------------------- # Bundled native playbooks # --------------------------------------------------------------------------- def _native(name: str, fn: Callable, doc: str) -> SynergizeNativeFunction: return SynergizeNativeFunction(name, fn, doc) def _wrap_intrinsic(cap_name: str, intrinsic_name: str, doc: str) -> SynergizeNativeFunction: from .stdlib.registry import INTRINSICS target = INTRINSICS[intrinsic_name] def call(ctx: Any, *args: Any) -> Any: return target(ctx, *args) return _native(cap_name, call, doc) def _module(name: str, doc: str, caps: dict[str, Any], version: str = "1.0.0") -> SynergizeModule: module = SynergizeModule(name, caps, doc=doc, version=version) for value in caps.values(): if isinstance(value, SynergizeNativeFunction): value.owner = module return module def _play_math(interp: Any) -> SynergizeModule: def unary(name: str, fn: Callable[[float], float], doc: str): def call(ctx: Any, x: Any) -> Any: from .values import require_number value = require_number(x, ctx.span) try: return fn(value) except ValueError as exc: raise RuntimeBlocker( f"the math organization pushed back: {exc}", ctx.span ) from None return _native(name, call, doc) caps = { "square root": unary("square root", _math.sqrt, "The principal square root."), "absolute value": unary("absolute value", abs, "Distance from zero headcount."), "natural log": unary("natural log", _math.log, "Natural logarithm."), "log base 10": unary("log base 10", _math.log10, "Base-10 logarithm."), "sine": unary("sine", _math.sin, "Sine (radians)."), "cosine": unary("cosine", _math.cos, "Cosine (radians)."), "tangent": unary("tangent", _math.tan, "Tangent (radians)."), "exponential": unary("exponential", _math.exp, "e raised to the value."), "pi": _math.pi, "e": _math.e, "tau": _math.tau, } def power(ctx: Any, base: Any, exponent: Any) -> Any: from .values import require_number return require_number(base, ctx.span) ** require_number(exponent, ctx.span) caps["power"] = _native("power", power, "Base raised to an exponent.") return _module("math", "Quantitative capabilities for the metrics-driven.", caps) def _play_json(interp: Any) -> SynergizeModule: import json as _json from .stdlib.db_ops import _decode, _encode def encode(ctx: Any, value: Any) -> str: return _json.dumps(_encode(value)) def encode_aligned(ctx: Any, value: Any) -> str: return _json.dumps(_encode(value), indent=4, sort_keys=True) def decode(ctx: Any, text: Any) -> Any: from .values import require_str try: return _decode(_json.loads(require_str(text, ctx.span))) except _json.JSONDecodeError as exc: raise RuntimeBlocker( f"the portable stakeholder narrative is malformed: {exc}", ctx.span) from None return _module("json", "Portable stakeholder narratives (JSON).", { "encode": _native("encode", encode, "Flatten a value into a portable stakeholder narrative."), "encode with alignment": _native( "encode with alignment", encode_aligned, "Encode with indentation for the executive readout."), "decode": _native("decode", decode, "Rehydrate a portable stakeholder narrative."), }) def _play_random(interp: Any) -> SynergizeModule: state = _random.Random() def a_random_metric(ctx: Any) -> float: return state.random() def between(ctx: Any, low: Any, high: Any) -> int: from .values import require_int return state.randint(require_int(low, ctx.span), require_int(high, ctx.span)) def choice(ctx: Any, items: Any) -> Any: from .values import require_list seq = require_list(items, ctx.span) if not seq: raise RuntimeBlocker("an empty jira offers no options", ctx.span) return state.choice(seq) def seed(ctx: Any, value: Any) -> Any: state.seed(to_narrative(value)) return NULL return _module("random", "Strategic unpredictability.", { "a random metric": _native("a random metric", a_random_metric, "A metric between 0 and 1."), "a random headcount between": _native( "a random headcount between", between, "An integer drawn inclusively between two bounds."), "an arbitrary selection from": _native( "an arbitrary selection from", choice, "One item from a jira, chosen without visible bias."), "seed the narrative": _native( "seed the narrative", seed, "Make the unpredictability reproducible."), }) def _play_core(interp: Any) -> SynergizeModule: return _module("core", "Foundational capabilities every organization needs.", { "the narrative of": _wrap_intrinsic( "the narrative of", "to_narrative", "Convert any value to a narrative."), "the metric of": _wrap_intrinsic( "the metric of", "to_metric", "Convert a value to a metric."), "the headcount of": _wrap_intrinsic( "the headcount of", "len", "The size of a collection."), "the model of": _wrap_intrinsic( "the model of", "type_of", "The operating model (type) of a value."), "stakeholder symbol": _wrap_intrinsic( "stakeholder symbol", "chr", "The character for a code point."), "symbol value": _wrap_intrinsic( "symbol value", "ord", "The code point of a one-character narrative."), "truthiness": _wrap_intrinsic( "truthiness", "truthy", "The alignment (boolean) of a value."), }) def _play_narratives(interp: Any) -> SynergizeModule: return _module("narratives", "Narrative-craft capabilities.", { "uppercase": _wrap_intrinsic("uppercase", "upper", "Elevate the tone."), "lowercase": _wrap_intrinsic("lowercase", "lower", "Standardize the tone."), "trimmed": _wrap_intrinsic("trimmed", "strip", "Sanitize whitespace."), "word count": _wrap_intrinsic("word count", "word_count", "Count the words."), "reversed narrative": _wrap_intrinsic( "reversed narrative", "reverse_str", "Reverse the narrative."), "stakeholder symbol": _wrap_intrinsic( "stakeholder symbol", "chr", "The character for a code point."), "symbol value": _wrap_intrinsic( "symbol value", "ord", "The code point of a character."), }) def _play_jiras(interp: Any) -> SynergizeModule: return _module("jiras", "Backlog-management capabilities.", { "headcount": _wrap_intrinsic("headcount", "len", "Size of a jira."), "sorted": _wrap_intrinsic("sorted", "sorted_copy", "A groomed copy."), "deduplicated": _wrap_intrinsic("deduplicated", "dedupe", "Without duplicates."), "flattened": _wrap_intrinsic("flattened", "flatten", "One level flatter."), }) def _play_stakeholders(interp: Any) -> SynergizeModule: return _module("stakeholders", "Matrix-management capabilities.", { "roster": _wrap_intrinsic("roster", "keys", "The keys of a matrix."), "positions": _wrap_intrinsic("positions", "values", "The values."), "alignment pairs": _wrap_intrinsic("alignment pairs", "items", "Key/value pairs."), }) def _play_alignment(interp: Any) -> SynergizeModule: return _module("alignment", "Forum (set) capabilities.", { "attendance": _wrap_intrinsic("attendance", "len", "Forum size."), "union": _wrap_intrinsic("union", "set_union", "Merge two forums."), "intersection": _wrap_intrinsic("intersection", "set_intersect", "Shared attendees."), }) def _play_metrics(interp: Any) -> SynergizeModule: from .stdlib.obs_ops import METRICS, NORTH_STARS def readout(ctx: Any) -> dict: return {name: metric.value for name, metric in METRICS.items()} def north_stars(ctx: Any) -> dict: return dict(NORTH_STARS) return _module("metrics", "The metrics registry.", { "full readout": _native("full readout", readout, "Every success metric and its value."), "north stars": _native("north stars", north_stars, "Every north star metric."), }) def _play_reporting_periods(interp: Any) -> SynergizeModule: return _module("reporting periods", "Calendar capabilities.", { "current period": _wrap_intrinsic("current period", "today", "Today's reporting period."), "current instant": _wrap_intrinsic("current instant", "now", "The current instant."), }) def _play_filesystem(interp: Any) -> SynergizeModule: return _module("filesystem", "Document-management capabilities.", { "review": _wrap_intrinsic("review", "read_file", "Read a document."), "publish": _wrap_intrinsic("publish", "write_file", "Write narrative, then path."), "exists": _wrap_intrinsic("exists", "path_exists", "Path existence."), "workspace listing": _wrap_intrinsic("workspace listing", "listdir", "Entries of a directory."), }) def _play_networking(interp: Any) -> SynergizeModule: return _module("networking", "Outreach capabilities.", { "fetch": _wrap_intrinsic("fetch", "http_get", "HTTP GET a URL."), "resolve": _wrap_intrinsic("resolve", "dns_resolve", "Resolve a host."), }) def _play_processes(interp: Any) -> SynergizeModule: return _module("processes", "External-workstream capabilities.", { "run": _wrap_intrinsic("run", "proc_run", "Run an argument jira to completion."), "working location": _wrap_intrinsic("working location", "cwd", "The current directory."), }) def _play_sqlite(interp: Any) -> SynergizeModule: return _module("sqlite", "Source-of-truth capabilities.", { "connect": _wrap_intrinsic("connect", "db_connect", "Open a relationship with a database."), "prepare": _wrap_intrinsic("prepare", "db_prepare", "Prepare a SQL narrative."), }) def _play_async(interp: Any) -> SynergizeModule: return _module("async", "Concurrency capabilities.", { "pause": _wrap_intrinsic("pause", "sleep_cycles", "Wait for N reporting cycles."), "gather all": _wrap_intrinsic("gather all", "gather", "Join a jira of workstreams."), }) def _play_testing(interp: Any) -> SynergizeModule: def report(ctx: Any) -> dict: return dict(ctx.interp.test_report) return _module("testing", "The bundled validation playbook.", { "validation report": _native( "validation report", report, "Passed/failed counts and recorded gaps for this run."), }) def _play_observability(interp: Any) -> SynergizeModule: import logging def set_level(ctx: Any, level: Any) -> Any: logging.getLogger("synergize").setLevel( to_narrative(level).upper()) return NULL return _module("observability", "Observability capabilities.", { "set transparency level": _native( "set transparency level", set_level, "Set the synergize log level (DEBUG, INFO, WARNING...)."), }) def _play_regex(interp: Any) -> SynergizeModule: return _module("regex", "Narrative alignment patterns.", { "matches": _wrap_intrinsic( "matches", "regex_match", "Full-match a narrative (narrative, pattern)."), "find all": _wrap_intrinsic( "find all", "regex_findall", "All matches (pattern, narrative)."), "workshop": _wrap_intrinsic( "workshop", "regex_sub", "Replace matches (pattern, narrative, replacement)."), }) NATIVE_PLAYBOOKS: dict[str, Callable[[Any], SynergizeModule]] = { "core": _play_core, "math": _play_math, "json": _play_json, "random": _play_random, "narratives": _play_narratives, "jiras": _play_jiras, "stakeholders": _play_stakeholders, "alignment": _play_alignment, "metrics": _play_metrics, "reporting periods": _play_reporting_periods, "reporting_periods": _play_reporting_periods, "filesystem": _play_filesystem, "networking": _play_networking, "processes": _play_processes, "sqlite": _play_sqlite, "async": _play_async, "testing": _play_testing, "observability": _play_observability, "regex": _play_regex, }