py-scopes/scopes/core/core.py

# 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 env _meta',
  'connect create created data default query remove set value',
  'c00 e00 s00 op rcvr',
  '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[Symbol | 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) -> Symbol | None:
        msg_new = op(msg)
        if msg_new is not None:
            await rcvr.put(msg_new) # return actor.send(rcvr, msg_new)
        return None
    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 | None] = 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 noop(no_forward: bool = False) -> Handler:
    async def handle(msg: Message, scope: Scope) -> Message | None:
        return None if no_forward else msg
    return handle

def forward(no_syns_notify: bool = False) -> Handler:
    do_notify = notify() if no_syns_notify else None
    async def handle(msg: Message, scope: Scope) -> None:
        count = 0
        for idx, syn in enumerate(scope.syns):
            res = await syn(msg) if syn is not None else None
            if res == k.remove:
                scope.syns[idx] = None
            else:
                count += 1
        if count == 0 and do_notify:
            do_notify(msg, scope)
    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 notify()(msg, scope)