Skip to content

Explore

tangram_explore

EXPLORE_CHANNEL module-attribute 

EXPLORE_CHANNEL = 'explore'

EXPLORE_EVENT module-attribute 

EXPLORE_EVENT = 'layers'

router module-attribute 

router = APIRouter(prefix='/explore', tags=['explore'])

LayerId module-attribute 

LayerId: TypeAlias = str

Unique UUID for a layer.

ParquetBytes module-attribute 

ParquetBytes: TypeAlias = bytes

LayerConfig module-attribute 

LayerConfig: TypeAlias = dict[str, Any]

Serialised layer configuration without data or label.

plugin module-attribute 

plugin = Plugin(
    frontend_path="dist-frontend",
    routers=[router],
    config_class=ExploreConfig,
    frontend_config_class=ExploreFrontendConfig,
    into_frontend_config_function=into_frontend,
    lifespan=lifespan,
)

ExploreState dataclass

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
44
45
46
47
48
@dataclass
class ExploreState:
    data: dict[LayerId, ParquetBytes] = field(default_factory=dict)
    layers: dict[LayerId, LayerConfig] = field(default_factory=dict)
    layer_order: list[LayerId] = field(default_factory=list)

data class-attribute instance-attribute 

data: dict[LayerId, ParquetBytes] = field(
    default_factory=dict
)

layers class-attribute instance-attribute 

layers: dict[LayerId, LayerConfig] = field(
    default_factory=dict
)

layer_order class-attribute instance-attribute 

layer_order: list[LayerId] = field(default_factory=list)

__init__ 

__init__(
    data: dict[LayerId, ParquetBytes] = dict(),
    layers: dict[LayerId, LayerConfig] = dict(),
    layer_order: list[LayerId] = list(),
) -> None

ArrowStreamExportable

Bases: Protocol

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
79
80
81
@runtime_checkable
class ArrowStreamExportable(Protocol):
    def __arrow_c_stream__(self, requested_schema: Any = None) -> Any: ...

__arrow_c_stream__ 

__arrow_c_stream__(requested_schema: Any = None) -> Any
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
81
def __arrow_c_stream__(self, requested_schema: Any = None) -> Any: ...

ExploreLayer

Bases: Protocol

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
84
85
86
87
88
89
90
91
92
@runtime_checkable
class ExploreLayer(Protocol):
    data: ArrowStreamExportable
    """Any data structure that implements the
    [Arrow C data interface](https://arrow.apache.org/docs/format/CDataInterface.html),
    such as polars DataFrames or pyarrow Tables."""
    label: str | None
    """Unique name for the layer (optional).
    If not provided, a random 8-character ID will be used."""

data instance-attribute 

data: ArrowStreamExportable

Any data structure that implements the Arrow C data interface, such as polars DataFrames or pyarrow Tables.

label instance-attribute 

label: str | None

Unique name for the layer (optional). If not provided, a random 8-character ID will be used.

ScatterLayer dataclass

Bases: ExploreLayer

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass(frozen=True, slots=True)
class ScatterLayer(ExploreLayer):
    data: ArrowStreamExportable
    _: KW_ONLY
    radius_scale: float = 50.0
    radius_min_pixels: float = 2.0
    radius_max_pixels: float = 5.0
    line_width_min_pixels: float = 1.0
    fill_color: str | list[int] = "#027ec7"
    line_color: str | list[int] = "#000000"
    opacity: float = 0.8
    stroked: bool = False
    filled: bool = True
    pickable: bool = True
    label: str | None = None
    kind: Literal["scatter"] = field(default="scatter", init=False)

data instance-attribute 

data: ArrowStreamExportable

_ instance-attribute 

_: KW_ONLY

radius_scale class-attribute instance-attribute 

radius_scale: float = 50.0

radius_min_pixels class-attribute instance-attribute 

radius_min_pixels: float = 2.0

radius_max_pixels class-attribute instance-attribute 

radius_max_pixels: float = 5.0

line_width_min_pixels class-attribute instance-attribute 

line_width_min_pixels: float = 1.0

fill_color class-attribute instance-attribute 

fill_color: str | list[int] = '#027ec7'

line_color class-attribute instance-attribute 

line_color: str | list[int] = '#000000'

opacity class-attribute instance-attribute 

opacity: float = 0.8

stroked class-attribute instance-attribute 

stroked: bool = False

filled class-attribute instance-attribute 

filled: bool = True

pickable class-attribute instance-attribute 

pickable: bool = True

label class-attribute instance-attribute 

label: str | None = None

kind class-attribute instance-attribute 

kind: Literal["scatter"] = field(
    default="scatter", init=False
)

__init__ 

__init__(
    data: ArrowStreamExportable,
    *,
    radius_scale: float = 50.0,
    radius_min_pixels: float = 2.0,
    radius_max_pixels: float = 5.0,
    line_width_min_pixels: float = 1.0,
    fill_color: str | list[int] = "#027ec7",
    line_color: str | list[int] = "#000000",
    opacity: float = 0.8,
    stroked: bool = False,
    filled: bool = True,
    pickable: bool = True,
    label: str | None = None,
) -> None

Layer dataclass

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
126
127
128
129
130
131
132
@dataclass(frozen=True, slots=True)
class Layer:
    id: str
    _session: Session

    async def remove(self) -> None:
        await self._session.remove(self.id)

id instance-attribute 

id: str

__init__ 

__init__(id: str, _session: Session) -> None

remove async 

remove() -> None
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
131
132
async def remove(self) -> None:
    await self._session.remove(self.id)

Session dataclass

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@dataclass(frozen=True, slots=True)
class Session:
    state: BackendState

    def __post_init__(self) -> None:
        if not hasattr(self.state, "explore_state"):
            raise RuntimeError(
                "The 'tangram_explore' plugin is not active. "
                "Ensure it is added to the [core.plugins] list in your config."
            )

    async def _broadcast(self, op: str, **kwargs: Any) -> None:
        payload = {"op": op, **kwargs}
        topic = f"to:{EXPLORE_CHANNEL}:{EXPLORE_EVENT}"
        await self.state.redis_client.publish(topic, orjson.dumps(payload))

    async def push(self, layer: ExploreLayer) -> Layer:
        if not is_dataclass(layer):  # required for fields()
            raise TypeError("layer must be a dataclass")
        if not isinstance(layer, ExploreLayer):
            raise TypeError("layer must implement ExploreLayer protocol")

        data_id = str(uuid.uuid4())
        parquet_bytes = _to_parquet_bytes(layer.data)

        explore_state = get_explore_state(self.state)
        explore_state.data[data_id] = parquet_bytes

        style = {
            f.name: getattr(layer, f.name)
            for f in fields(layer)
            if f.name not in ("data", "label")
        }
        label = getattr(layer, "label", None)

        layer_def = {
            "id": data_id,
            "label": label or data_id[:8],
            "url": f"/explore/data/{data_id}",
            "style": style,
        }
        explore_state.layers[data_id] = layer_def
        explore_state.layer_order.append(data_id)

        await self._broadcast("add", layer=layer_def)

        return Layer(id=data_id, _session=self)

    async def remove(self, data_id: str) -> None:
        explore_state = get_explore_state(self.state)
        if data_id in explore_state.data:
            del explore_state.data[data_id]
        if data_id in explore_state.layers:
            del explore_state.layers[data_id]
        if data_id in explore_state.layer_order:
            explore_state.layer_order.remove(data_id)
        await self._broadcast("remove", id=data_id)

    async def clear(self) -> None:
        explore_state = get_explore_state(self.state)
        explore_state.data.clear()
        explore_state.layers.clear()
        explore_state.layer_order.clear()
        await self._broadcast("clear")

state instance-attribute 

state: BackendState

__init__ 

__init__(state: BackendState) -> None

__post_init__ 

__post_init__() -> None
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
139
140
141
142
143
144
def __post_init__(self) -> None:
    if not hasattr(self.state, "explore_state"):
        raise RuntimeError(
            "The 'tangram_explore' plugin is not active. "
            "Ensure it is added to the [core.plugins] list in your config."
        )

push async 

push(layer: ExploreLayer) -> Layer
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
async def push(self, layer: ExploreLayer) -> Layer:
    if not is_dataclass(layer):  # required for fields()
        raise TypeError("layer must be a dataclass")
    if not isinstance(layer, ExploreLayer):
        raise TypeError("layer must implement ExploreLayer protocol")

    data_id = str(uuid.uuid4())
    parquet_bytes = _to_parquet_bytes(layer.data)

    explore_state = get_explore_state(self.state)
    explore_state.data[data_id] = parquet_bytes

    style = {
        f.name: getattr(layer, f.name)
        for f in fields(layer)
        if f.name not in ("data", "label")
    }
    label = getattr(layer, "label", None)

    layer_def = {
        "id": data_id,
        "label": label or data_id[:8],
        "url": f"/explore/data/{data_id}",
        "style": style,
    }
    explore_state.layers[data_id] = layer_def
    explore_state.layer_order.append(data_id)

    await self._broadcast("add", layer=layer_def)

    return Layer(id=data_id, _session=self)

remove async 

remove(data_id: str) -> None
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
183
184
185
186
187
188
189
190
191
async def remove(self, data_id: str) -> None:
    explore_state = get_explore_state(self.state)
    if data_id in explore_state.data:
        del explore_state.data[data_id]
    if data_id in explore_state.layers:
        del explore_state.layers[data_id]
    if data_id in explore_state.layer_order:
        explore_state.layer_order.remove(data_id)
    await self._broadcast("remove", id=data_id)

clear async 

clear() -> None
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
193
194
195
196
197
198
async def clear(self) -> None:
    explore_state = get_explore_state(self.state)
    explore_state.data.clear()
    explore_state.layers.clear()
    explore_state.layer_order.clear()
    await self._broadcast("clear")

ExploreConfig dataclass

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
201
202
203
@dataclass
class ExploreConfig:
    enable_3d: bool = False

enable_3d class-attribute instance-attribute 

enable_3d: bool = False

__init__ 

__init__(enable_3d: bool = False) -> None

ExploreFrontendConfig dataclass

Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
206
207
208
209
@dataclass
class ExploreFrontendConfig:
    enable_3d: Annotated[bool, FrontendMutable()]
    """Whether to render scatter points in 3D"""

enable_3d instance-attribute 

enable_3d: Annotated[bool, FrontendMutable()]

Whether to render scatter points in 3D

__init__ 

__init__(
    enable_3d: Annotated[bool, FrontendMutable()],
) -> None

lifespan async 

lifespan(state: BackendState) -> AsyncGenerator[None, None]
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
51
52
53
54
55
@asynccontextmanager
async def lifespan(state: BackendState) -> AsyncGenerator[None, None]:
    setattr(state, "explore_state", ExploreState())
    yield
    delattr(state, "explore_state")

get_explore_state 

get_explore_state(
    state: InjectBackendState,
) -> ExploreState
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
58
59
def get_explore_state(state: tangram_core.InjectBackendState) -> ExploreState:
    return cast(ExploreState, getattr(state, "explore_state"))

get_explore_data async 

get_explore_data(
    data_id: str, state: InjectBackendState
) -> Response
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
62
63
64
65
66
67
68
69
70
@router.get("/data/{data_id}")
async def get_explore_data(
    data_id: str, state: tangram_core.InjectBackendState
) -> Response:
    explore_state = get_explore_state(state)
    data = explore_state.data.get(data_id)
    if data is None:
        return Response(status_code=404)  # shouldn't occur
    return Response(content=data, media_type="application/vnd.apache.parquet")

get_layers async 

get_layers(
    state: InjectBackendState,
) -> list[dict[str, Any]]
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
73
74
75
76
@router.get("/layers")
async def get_layers(state: tangram_core.InjectBackendState) -> list[dict[str, Any]]:
    explore_state = get_explore_state(state)
    return [explore_state.layers[uid] for uid in explore_state.layer_order]

into_frontend 

into_frontend(
    config: ExploreConfig,
) -> ExploreFrontendConfig
Source code in packages/tangram_explore/src/tangram_explore/__init__.py 
212
213
def into_frontend(config: ExploreConfig) -> ExploreFrontendConfig:
    return ExploreFrontendConfig(enable_3d=config.enable_3d)