130 lines
3.8 KiB
Python
130 lines
3.8 KiB
Python
# 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
|
|
|
|
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]
|
|
|
|
env_dummy: Mailbox = asyncio.Queue()
|
|
|
|
def notify() -> Handler:
|
|
async def handle(msg, 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.head['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 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 create_zero(env: Mailbox) -> Mailbox:
|
|
scope = create_scope(env)
|
|
zero = neuron(scope)
|
|
return zero
|
|
|
|
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
|
|
|
|
def std_proc(actions: Actions) -> Proc:
|
|
return gen_proc(actions, finalize=forward())
|
|
|
|
def basic_actions() -> Actions:
|
|
acts: Actions = {
|
|
k.value: value_add(),
|
|
k.create: create()
|
|
}
|
|
return acts
|
|
|
|
# predefined action handlers
|
|
|
|
def forward() -> Handler:
|
|
async def handle(msg, scope) -> None:
|
|
for syn in scope.syns:
|
|
await syn(msg)
|
|
return handle
|
|
|
|
def value_add() -> Handler:
|
|
async def handle(msg, scope) -> tuple[Message, Scope]:
|
|
return msg, scope
|
|
return handle
|
|
|
|
def create() -> Handler:
|
|
async def handle(msg, scope) -> None:
|
|
return None
|
|
return handle
|
|
|