41 lines
1.2 KiB
Python
41 lines
1.2 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 AsyncGenerator, Callable, Coroutine
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from scopes.core.message import Message
|
|
from scopes.util import keywords as k, Symbol
|
|
|
|
type Mailbox = asyncio.Queue
|
|
type Categ = tuple[Symbol, Symbol]
|
|
type Proc = Callable[[Scope, Message], AsyncGenerator[Scope]]
|
|
type Program = Callable[[Scope], Proc]
|
|
type Syn = Callable[[Message], Message]
|
|
|
|
async def proc_dummy(scope: Scope, msg: Message) -> AsyncGenerator[Scope]:
|
|
yield scope
|
|
|
|
@dataclass
|
|
class Scope:
|
|
|
|
value: Any = 0
|
|
categ: Categ = (k.csys, k.c00)
|
|
stage: Symbol = k.initial
|
|
proc: Proc = proc_dummy
|
|
program: Program = lambda x: proc_dummy
|
|
syns: list[Syn] = field(default_factory=list)
|
|
environ: Mailbox = field(default_factory=asyncio.Queue)
|
|
|
|
def set_proc(self) -> Scope:
|
|
self.proc = self.program(self)
|
|
return self
|
|
|
|
def create_scope(env: Mailbox, cat=(k.csys, k.c00), **kw):
|
|
scope = Scope(environ=env, categ=cat, **kw)
|
|
return scope.set_proc()
|
|
|