util: Symbols / keywords improvements, define symbol lists with namespaces

This commit is contained in:
Helmut Merz 2026-10-23 23:12:33 +02:00
parent 717ce25372
commit 9ee4dd73aa
6 changed files with 37 additions and 14 deletions

View file

@ -11,7 +11,16 @@ 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
from scopes.util import keywords as k, Symbol, Symbols
# keywords (domains, actions, categories, stages):
k('csys', 'env',
'connect', 'create', 'created', 'data', 'default', 'value',
'c00', 'e00', 's00',
'initial', 'active', 'retired')
dom = util.Symbols('dom')('csys', 'env')
cat = util.Symbols('cat')('c00', 'e00', 's00')
type Mailbox = asyncio.Queue
type Categ = tuple[Symbol, Symbol]
@ -52,7 +61,7 @@ def notify() -> Handler:
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 = 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)
@ -78,7 +87,7 @@ def gen_program(spec: ProgramSpec, default: Proc = gen_proc({})) -> Program:
class Scope:
value: Any = 0
categ: Categ = (k.csys, k.c00)
categ: Categ = (k.csys, cat.c00)
stage: Symbol = k.initial
proc: Proc = field(init=False)
program: Program = gen_program({})
@ -92,7 +101,7 @@ class Scope:
self.proc = self.program(self)
return self
def create_scope(env: Mailbox, cat: Categ = (k.csys, k.c00), **kw) -> Scope:
def create_scope(env: Mailbox, cat: Categ = (k.csys, cat.c00), **kw) -> Scope:
return Scope(environ=env, categ=cat, **kw)
def clone_scope(scope: Scope, **kw) -> Scope:
@ -146,6 +155,6 @@ def std_proc(actions: Actions | None = None, finalize: Handler = forward(),
# helpers
async def notify_created(msg: Message, new: Mailbox, scope: Scope) -> None:
msg = Message(msg.head['domain'], k.created, data=msg.data | {k.new: new})
msg = Message(msg.head.domain, k.created, data=msg.data | {k.new: new})
await scope.environ.put(msg)

View file

@ -18,6 +18,9 @@ class Record:
self.set(self.headFields[ix], hv)
self.data = data or {}
def __getattr__(self, k):
raise AttributeError(k)
@property
def headValues(self):
return [self.head[k] for k in self.headFields]
@ -27,7 +30,7 @@ class Record:
if value is None:
value = self.defaultHeadValue
elif attr in self.headFieldsAsKeywords:
value = k(value)
value = k[value]
self.head[attr] = value
setattr(self, attr, value)
else:

View file

@ -15,12 +15,20 @@ class Symbol(str):
class Symbols:
def __getattr__(self, k):
sym = Symbol(k)
setattr(self, k, sym)
return sym
def __init__(self, prefix=None):
self.__prefix = prefix
def __call__(self, k):
def __call__(self, *ks):
prefix = self.__prefix + '.' if self.__prefix else ''
for k in ks:
if not hasattr(self, k):
setattr(self, k, Symbol(prefix + k))
return self
def __getattr__(self, k):
raise AttributeError(k)
def __getitem__(self, k):
return getattr(self, k)
keywords = Symbols()

View file

@ -9,6 +9,7 @@ from scopes import core
from scopes.core import actor, environ
from scopes.core.message import Message
from scopes.util import keywords as k
k('data', 'hello', 'person', 'show', 'switch', 'quit', 'test')
@pytest.fixture
def collector():
@ -71,6 +72,7 @@ async def test_core(meta_env, collector):
await env.put(Message(k.env, k.hello))
await zero.put(Message(k.csys, k.hello))
await zero.put(Message(k.csys, k.create))
#data={k.connect: k.pred, k.address: (1, 1)}))
await asyncio.sleep(0.01)
assert len(collector) == 4
assert collector[0].action is k.created

View file

@ -5,13 +5,14 @@
import pytest
from scopes.util import keywords as k, flatten
k('dummy', 'other', 'cd', 'new')
def test_keywords():
assert k.dummy is k.dummy
assert k.dummy is not k.other
assert getattr(k, "abcd"[-2:]) is k.cd
assert k('dummy') is k.dummy
assert k('new') is k('new')
assert k['dummy'] is k.dummy
assert k['new'] is k['new']
def test_flatten():
lst = (1, (None, 2), (3, (4, 5)))