# scopes.core.core """A concurrent cybernetic communication system""" from __future__ import annotations # no check of type annotations at runtime import asyncio from collections.abc import Awaitable, Callable, Coroutine from dataclasses import dataclass, field from typing import Any, Self from scopes.core import actor from scopes.core.message import Message from scopes import util from scopes.util import keywords as k, Symbol, Symbols # keywords (domains, actions, classes, stages, data keys): k('csys ctl env', 'connect create created data default query set value', 'c00 e00 s00 rcvr op', 'initial active retired', 'address pred succ') type Mailbox = asyncio.Queue type Categ = tuple[Symbol, Symbol] type Proc = Callable[[Message, Scope], Awaitable[Scope | None]] type Program = Callable[[Scope], Proc] type ProgramSpec = dict[tuple[Symbol, Symbol], Proc] type Syn = Callable[[Message], Awaitable[None]] type SynOp = Callable[[Message], Message | None] type Handler = Callable[[Message, Scope], Awaitable[Message | None | tuple[Message | None, Scope | None]]] type Handlers = Handler | None | list[Handlers] type Actions = dict[Symbol, Handlers] def neuron(scope: Scope) -> Mailbox: async def process(msg: Message) -> actor.Behavior | None: nonlocal scope scope = await scope.proc(msg, scope) or scope return process return actor.create(process) def synapse(rcvr: Mailbox, op: SynOp = lambda x: x) -> Syn: async def syn(msg: Message) -> None: # op = update_op(msg) (=> return?) # rv = check_return(msg) => return rv msg_new = op(msg) if msg_new is not None: await rcvr.put(msg_new) return syn env_dummy: Mailbox = asyncio.Queue() def notify() -> Handler: async def handle(msg: Message, scope: Scope) -> None: await scope.environ.put(msg) return handle def gen_proc(actions: Actions, default: Handlers = notify(), prepare: Handlers = None, finalize: Handlers = None) -> Proc: async def proc(msg: Message, scope: Scope) -> Scope: acts = actions.get(msg.action, actions.get(k.default, default)) acts = util.flatten(prepare, acts, finalize, ignore_none=True) for act in acts: msg = await act(msg, scope) if isinstance(msg, tuple): msg, new_scope = msg if new_scope is not None: scope = new_scope if msg is None: break return scope return proc def gen_program(spec: ProgramSpec, default: Proc = gen_proc({})) -> Program: def program(scope: Scope) -> Proc: key = (scope.categ[1], scope.stage) proc = spec.get(key, spec.get((key[0], k.default))) if proc is None: proc = spec.get((k.default, key[1]), default) return proc return program @dataclass class Scope: value: Any = 0 categ: Categ = (k.csys, k.c00) stage: Symbol = k.initial proc: Proc = field(init=False) program: Program = gen_program({}) syns: list[Syn] = field(default_factory=list) environ: Mailbox = env_dummy def __post_init__(self): self.set_proc() def set_proc(self) -> Self: self.proc = self.program(self) return self def create_scope(env: Mailbox, cat: Categ = (k.csys, k.c00), **kw) -> Scope: return Scope(environ=env, categ=cat, **kw) def clone_scope(scope: Scope, **kw) -> Scope: value = kw.get('value', 0) categ = kw.get('categ', scope.categ) stage = kw.get('stage', k.initial) return Scope(environ=scope.environ, value=value, categ=categ, stage=stage, program=scope.program, syns=[]) async def create_zero(env: Mailbox) -> Mailbox: scope = create_scope(env, program=gen_program({}, default=std_proc())) zero = neuron(scope) await notify_created(Message(k.csys), zero, scope) return zero # predefined action handlers and Action sets def forward() -> Handler: async def handle(msg: Message, scope: Scope) -> None: for syn in scope.syns: await syn(msg) return handle def value_add() -> Handler: async def handle(msg: Message, scope: Scope) -> tuple[Message, Scope]: return msg, scope return handle def create() -> Handler: async def handle(msg: Message, scope: Scope) -> None: new = neuron(clone_scope(scope, **msg.data)) await notify_created(msg, new, scope) conn = msg.data.get(k.connect) return handle def connect() -> Handler: async def handle(msg: Message, scope: Scope) -> None: return None return handle def basic_actions() -> Actions: acts: Actions = { k.value: value_add(), k.create: create(), k.connect: connect() } return acts # predefined specialized procs def std_proc(actions: Actions | None = None, finalize: Handler = forward(), more: Actions = {}) -> Proc: if actions is None: actions = basic_actions() actions.update(more) return gen_proc(actions, finalize=finalize) # helpers async def notify_created(msg: Message, new: Mailbox, scope: Scope) -> None: msg = Message(msg.domain, k.created, data=msg.data | {k.new: new}) await scope.environ.put(msg)