跳转至

socioverse.env_layers — information-axis helpers

Reusable building blocks for the information side of E: audience-scoped broadcasts and the neighbour feed.

Broadcast delivery

socioverse.env_layers.information

InformationEnvironment — reusable machinery for the information axis of E.

Activates audience-scoped Broadcasts over time and renders them into the macro_information / local_information quadrants. A study's EnvironmentProvider composes one of these and supplies a matcher(selector, agent_ctx) -> bool for local audiences (Chicago: a TractSelector-aware matcher; default: a generic attribute matcher).

Scenario 1 (policy A to all + policy B to a tract at step n) is two Broadcasts: Broadcast(content=A, audience="all", at_step=n) -> macro_information for everyone Broadcast(content=B, audience={...tract...}, at_step=n) -> local_information for matches

InformationEnvironment 

InformationEnvironment(
    program: InformationProgram | None = None,
    matcher: Callable[[dict, dict], bool] | None = None,
)
Source code in socioverse/env_layers/information.py 
60
61
62
63
64
65
66
67
def __init__(
    self,
    program: InformationProgram | None = None,
    matcher: Callable[[dict, dict], bool] | None = None,
):
    self.program = program or InformationProgram()
    self.matcher = matcher or match_selector
    self._active: list[Broadcast] = []

advance_to 

advance_to(t: int) -> list[Broadcast]

Activate broadcasts firing at t, expire those past their ttl. Returns the newly-activated broadcasts (for event logging).

Source code in socioverse/env_layers/information.py 
72
73
74
75
76
77
78
def advance_to(self, t: int) -> list[Broadcast]:
    """Activate broadcasts firing at t, expire those past their ttl. Returns the
    newly-activated broadcasts (for event logging)."""
    newly = [b for b in self.program.broadcasts if b.at_step == t]
    self._active.extend(newly)
    self._active = [b for b in self._active if b.at_step + b.ttl > t]
    return newly

rendered_lines 

rendered_lines(agent_ctx: dict[str, Any]) -> list[str]

Flat text lines visible to an agent (macro + matching local) for LLM prompts.

Source code in socioverse/env_layers/information.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
def rendered_lines(self, agent_ctx: dict[str, Any]) -> list[str]:
    """Flat text lines visible to an agent (macro + matching local) for LLM prompts."""
    lines = []
    for b in self._active:
        visible = b.is_macro or (
            isinstance(b.audience, dict) and self.matcher(b.audience, agent_ctx)
        )
        if visible:
            lines.append(f"[{b.channel}] {b.content}")
    return lines

match_selector 

match_selector(
    selector: dict[str, Any], ctx: dict[str, Any]
) -> bool

Generic audience matcher over an agent context dict. Supported selector keys: - "": exact equality -> ctx[""] == value - "_in": membership -> ctx[""] in value - "_below"/"_above": numeric - "geoid_list": ctx['geoid'|'tract_id'] in value - "index_below": ctx['index'] < value Empty selector matches nobody (a broadcast must scope its local audience).

Source code in socioverse/env_layers/information.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def match_selector(selector: dict[str, Any], ctx: dict[str, Any]) -> bool:
    """Generic audience matcher over an agent context dict. Supported selector keys:
      - "<attr>": exact equality              -> ctx["<attr>"] == value
      - "<attr>_in": membership               -> ctx["<attr>"] in value
      - "<attr>_below"/"<attr>_above": numeric
      - "geoid_list": ctx['geoid'|'tract_id'] in value
      - "index_below": ctx['index'] < value
    Empty selector matches nobody (a broadcast must scope its local audience).
    """
    if not selector:
        return False
    for key, val in selector.items():
        if key.endswith("_in"):
            if ctx.get(key[:-3]) not in val:
                return False
        elif key.endswith("_below"):
            v = ctx.get(key[:-6])
            if v is None or not (v < val):
                return False
        elif key.endswith("_above"):
            v = ctx.get(key[:-6])
            if v is None or not (v > val):
                return False
        elif key == "geoid_list":
            if ctx.get("geoid") not in val and ctx.get("tract_id") not in val:
                return False
        elif key == "index_below":
            if not (ctx.get("index", float("inf")) < val):
                return False
        else:
            if ctx.get(key) != val:
                return False
    return True

Neighbour feed

socioverse.env_layers.neighbor_feed

NeighborFeed — local-information from agent-to-agent messages (scenario 2, HiSim).

Wraps a MessageBus + a neighbours() lookup to turn other agents' post/reply actions into each agent's local_information quadrant. Mediated delivery (not O(N^2) direct calls): posts land on the bus; observers read only their network neighbours' messages.

NeighborFeed 

NeighborFeed(
    neighbors_fn: Callable[[str], list[str]],
    bus: MessageBus | None = None,
)
Source code in socioverse/env_layers/neighbor_feed.py 
18
19
20
21
22
23
24
def __init__(
    self,
    neighbors_fn: Callable[[str], list[str]],
    bus: MessageBus | None = None,
):
    self.neighbors_fn = neighbors_fn
    self.bus: MessageBus = bus or InMemoryMessageBus()

ingest 

ingest(
    actions: list[Action], step: int, round_idx: int = 0
) -> list[dict[str, Any]]

Route message-like actions (kind in {post, reply}) onto the bus. Returns the message dicts (for persistence in the DuckDB messages table).

Source code in socioverse/env_layers/neighbor_feed.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def ingest(self, actions: list[Action], step: int, round_idx: int = 0) -> list[dict[str, Any]]:
    """Route message-like actions (kind in {post, reply}) onto the bus. Returns the
    message dicts (for persistence in the DuckDB `messages` table)."""
    posted = []
    for a in actions:
        if a.kind in ("post", "reply"):
            msg = {
                "author_id": a.agent_id,
                "content": a.payload.get("content", ""),
                "step": step,
                "round": round_idx,
                "channel": a.payload.get("channel", "chat"),
                "audience": a.payload.get("audience", "neighbors"),
            }
            self.bus.post(msg)
            posted.append(msg)
    return posted

News broadcast helpers

socioverse.env_layers.news_broadcast

Standard information-layer declarations + convenience constructors.

The two-axis system means a "macro news broadcast" and a "local ward notice" are the SAME mechanism (an audience-scoped Broadcast) differing only in audience. These helpers give studies ready-made EnvironmentLayer declarations and Broadcast builders so the information axis is one import away.

default_information_layers 

default_information_layers() -> list[EnvironmentLayer]

A macro news channel + a local neighbour/ward channel.

Source code in socioverse/env_layers/news_broadcast.py 
16
17
18
19
20
21
22
23
24
25
26
27
def default_information_layers() -> list[EnvironmentLayer]:
    """A macro news channel + a local neighbour/ward channel."""
    return [
        EnvironmentLayer(
            name="news", modality="information", scope="macro",
            description="city-wide news / policy broadcasts, seen by all agents",
        ),
        EnvironmentLayer(
            name="neighbor_feed", modality="information", scope="local",
            description="local ward notices + neighbour word-of-mouth, audience-scoped",
        ),
    ]

macro_news 

macro_news(
    message_id: str,
    content: str,
    at_step: int,
    ttl: int = 1,
    channel: str = "news",
) -> Broadcast

A broadcast delivered to EVERY agent's macro_information.

Source code in socioverse/env_layers/news_broadcast.py 
30
31
32
33
34
def macro_news(message_id: str, content: str, at_step: int, ttl: int = 1,
               channel: str = "news") -> Broadcast:
    """A broadcast delivered to EVERY agent's macro_information."""
    return Broadcast(message_id=message_id, content=content, channel=channel,
                     at_step=at_step, ttl=ttl, audience="all")

local_notice 

local_notice(
    message_id: str,
    content: str,
    at_step: int,
    audience: dict[str, Any],
    ttl: int = 1,
    channel: str = "neighbor_feed",
) -> Broadcast

A broadcast delivered only to agents matching audience (local_information).

Source code in socioverse/env_layers/news_broadcast.py 
37
38
39
40
41
def local_notice(message_id: str, content: str, at_step: int, audience: dict[str, Any],
                 ttl: int = 1, channel: str = "neighbor_feed") -> Broadcast:
    """A broadcast delivered only to agents matching `audience` (local_information)."""
    return Broadcast(message_id=message_id, content=content, channel=channel,
                     at_step=at_step, ttl=ttl, audience=audience)