from __future__ import annotations import functools import hashlib import io import itertools import json import operator import sys import typing as t import warnings from copy import deepcopy as _deepcopy from typing import ( TYPE_CHECKING, Any, Literal, Protocol, Sequence, TypeVar, Union, overload, ) from typing_extensions import TypeAlias import jsonschema import narwhals.stable.v1 as nw from altair import utils from altair.expr import core as _expr_core from altair.utils import Optional, Undefined from altair.utils._vegafusion_data import ( compile_with_vegafusion as _compile_with_vegafusion, ) from altair.utils._vegafusion_data import using_vegafusion as _using_vegafusion from altair.utils.core import ( to_eager_narwhals_dataframe as _to_eager_narwhals_dataframe, ) from altair.utils.data import DataType from altair.utils.data import is_data_type as _is_data_type from .compiler import vegalite_compilers from .data import data_transformers from .display import VEGA_VERSION, VEGAEMBED_VERSION, VEGALITE_VERSION, renderers from .schema import SCHEMA_URL, channels, core, mixins from .schema._typing import Map from .theme import themes if sys.version_info >= (3, 13): from typing import TypedDict else: from typing_extensions import TypedDict if TYPE_CHECKING: from pathlib import Path from typing import IO, Iterable, Iterator from altair.utils.core import DataFrameLike if sys.version_info >= (3, 13): from typing import Required, TypeIs else: from typing_extensions import Required, TypeIs if sys.version_info >= (3, 11): from typing import Never, Self else: from typing_extensions import Never, Self from altair.expr.core import ( BinaryExpression, Expression, GetAttrExpression, GetItemExpression, IntoExpression, ) from altair.utils.display import MimeBundleType from .schema._typing import ( AggregateOp_T, AutosizeType_T, ColorName_T, CompositeMark_T, ImputeMethod_T, LayoutAlign_T, Mark_T, MultiTimeUnit_T, OneOrSeq, ProjectionType_T, ResolveMode_T, SelectionResolution_T, SelectionType_T, SingleDefUnitChannel_T, SingleTimeUnit_T, StackOffset_T, ) from .schema.channels import Column, Facet, Row from .schema.core import ( AggregatedFieldDef, AggregateOp, AnyMark, BindCheckbox, Binding, BindRadioSelect, BindRange, BinParams, Expr, ExprRef, FacetedEncoding, FacetFieldDef, FieldName, GraticuleGenerator, ImputeMethod, ImputeSequence, InlineData, InlineDataset, IntervalSelectionConfig, JoinAggregateFieldDef, LayerRepeatMapping, LookupSelection, Mark, NamedData, ParameterName, PointSelectionConfig, Predicate, PredicateComposition, ProjectionType, RepeatMapping, RepeatRef, SchemaBase, SelectionParameter, SequenceGenerator, SortField, SphereGenerator, Step, TimeUnit, TopLevelSelectionParameter, Transform, UrlData, VariableParameter, Vector2number, Vector2Vector2number, Vector3number, WindowFieldDef, ) __all__ = [ "TOPLEVEL_ONLY_KEYS", "Bin", "ChainedWhen", "Chart", "ChartDataType", "ConcatChart", "DataType", "FacetChart", "FacetMapping", "HConcatChart", "Impute", "LayerChart", "LookupData", "Parameter", "ParameterExpression", "RepeatChart", "SelectionExpression", "SelectionPredicateComposition", "Then", "Title", "TopLevelMixin", "VConcatChart", "When", "binding", "binding_checkbox", "binding_radio", "binding_range", "binding_select", "check_fields_and_encodings", "concat", "condition", "graticule", "hconcat", "layer", "mixins", "param", "repeat", "selection", "selection_interval", "selection_multi", "selection_point", "selection_single", "sequence", "sphere", "topo_feature", "value", "vconcat", "when", ] ChartDataType: TypeAlias = Optional[Union[DataType, core.Data, str, core.Generator]] _TSchemaBase = TypeVar("_TSchemaBase", bound=core.SchemaBase) # ------------------------------------------------------------------------ # Data Utilities def _dataset_name(values: dict[str, Any] | list | InlineDataset) -> str: """ Generate a unique hash of the data. Parameters ---------- values : list, dict, core.InlineDataset A representation of data values. Returns ------- name : string A unique name generated from the hash of the values. """ if isinstance(values, core.InlineDataset): values = values.to_dict() if values == [{}]: return "empty" values_json = json.dumps(values, sort_keys=True, default=str) hsh = hashlib.sha256(values_json.encode()).hexdigest()[:32] return "data-" + hsh def _consolidate_data( data: ChartDataType | UrlData, context: dict[str, Any] ) -> ChartDataType | NamedData | InlineData | UrlData: """ If data is specified inline, then move it to context['datasets']. This function will modify context in-place, and return a new version of data """ values: Any = Undefined kwds = {} if isinstance(data, core.InlineData): if utils.is_undefined(data.name) and not utils.is_undefined(data.values): if isinstance(data.values, core.InlineDataset): values = data.to_dict()["values"] else: values = data.values kwds = {"format": data.format} elif isinstance(data, dict) and "name" not in data and "values" in data: values = data["values"] kwds = {k: v for k, v in data.items() if k != "values"} if not utils.is_undefined(values): name = _dataset_name(values) data = core.NamedData(name=name, **kwds) context.setdefault("datasets", {})[name] = values return data def _prepare_data( data: ChartDataType, context: dict[str, Any] | None = None ) -> ChartDataType | NamedData | InlineData | UrlData | Any: """ Convert input data to data for use within schema. Parameters ---------- data : The input dataset in the form of a DataFrame, dictionary, altair data object, or other type that is recognized by the data transformers. context : dict (optional) The to_dict context in which the data is being prepared. This is used to keep track of information that needs to be passed up and down the recursive serialization routine, such as global named datasets. """ if data is Undefined: return data # convert dataframes or objects with __geo_interface__ to dict elif not isinstance(data, dict) and _is_data_type(data): if func := data_transformers.get(): data = func(nw.to_native(data, strict=False)) # convert string input to a URLData elif isinstance(data, str): data = core.UrlData(data) # consolidate inline data to top-level datasets if context is not None and data_transformers.consolidate_datasets: data = _consolidate_data(data, context) # if data is still not a recognized type, then return if not isinstance(data, (dict, core.Data)): warnings.warn(f"data of type {type(data)} not recognized", stacklevel=1) return data # ------------------------------------------------------------------------ # Aliases & specializations Bin = core.BinParams Impute = core.ImputeParams Title = core.TitleParams class LookupData(core.LookupData): @utils.use_signature(core.LookupData) def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) def to_dict(self, *args: Any, **kwargs: Any) -> dict[str, Any]: """Convert the chart to a dictionary suitable for JSON export.""" copy = self.copy(deep=False) copy.data = _prepare_data(copy.data, kwargs.get("context")) return super(LookupData, copy).to_dict(*args, **kwargs) class FacetMapping(core.FacetMapping): _class_is_valid_at_instantiation = False @utils.use_signature(core.FacetMapping) def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) def to_dict(self, *args: Any, **kwargs: Any) -> dict[str, Any]: copy = self.copy(deep=False) context = kwargs.get("context", {}) data = context.get("data", None) if isinstance(self.row, str): copy.row = core.FacetFieldDef(**utils.parse_shorthand(self.row, data)) if isinstance(self.column, str): copy.column = core.FacetFieldDef(**utils.parse_shorthand(self.column, data)) return super(FacetMapping, copy).to_dict(*args, **kwargs) # ------------------------------------------------------------------------ # Encoding will contain channel objects that aren't valid at instantiation core.FacetedEncoding._class_is_valid_at_instantiation = False # ------------------------------------------------------------------------ # These are parameters that are valid at the top level, but are not valid # for specs that are within a composite chart # (layer, hconcat, vconcat, facet, repeat) TOPLEVEL_ONLY_KEYS = {"background", "config", "autosize", "padding", "$schema"} # ------------------------------------------------------------------------- # Tools for working with parameters class Parameter(_expr_core.OperatorMixin): """A Parameter object.""" _counter: int = 0 @classmethod def _get_name(cls) -> str: cls._counter += 1 return f"param_{cls._counter}" def __init__( self, name: str | None = None, empty: Optional[bool] = Undefined, param: Optional[ VariableParameter | TopLevelSelectionParameter | SelectionParameter ] = Undefined, param_type: Optional[Literal["variable", "selection"]] = Undefined, ) -> None: if name is None: name = self._get_name() self.name = name self.empty = empty self.param = param self.param_type = param_type @utils.deprecated( version="5.0.0", alternative="to_dict", message="No need to call '.ref()' anymore.", ) def ref(self) -> dict[str, Any]: """'ref' is deprecated. No need to call '.ref()' anymore.""" return self.to_dict() def to_dict(self) -> dict[str, str | dict[str, Any]]: if self.param_type == "variable": return {"expr": self.name} elif self.param_type == "selection": nm: Any = self.name return {"param": nm.to_dict() if hasattr(nm, "to_dict") else nm} else: msg = f"Unrecognized parameter type: {self.param_type}" raise ValueError(msg) def __invert__(self) -> SelectionPredicateComposition | Any: if self.param_type == "selection": return SelectionPredicateComposition({"not": {"param": self.name}}) else: return _expr_core.OperatorMixin.__invert__(self) def __and__(self, other: Any) -> SelectionPredicateComposition | Any: if self.param_type == "selection": if isinstance(other, Parameter): other = {"param": other.name} return SelectionPredicateComposition({"and": [{"param": self.name}, other]}) else: return _expr_core.OperatorMixin.__and__(self, other) def __or__(self, other: Any) -> SelectionPredicateComposition | Any: if self.param_type == "selection": if isinstance(other, Parameter): other = {"param": other.name} return SelectionPredicateComposition({"or": [{"param": self.name}, other]}) else: return _expr_core.OperatorMixin.__or__(self, other) def __repr__(self) -> str: return f"Parameter({self.name!r}, {self.param})" def _to_expr(self) -> str: return self.name def _from_expr(self, expr: IntoExpression) -> ParameterExpression: return ParameterExpression(expr=expr) def __getattr__(self, field_name: str) -> GetAttrExpression | SelectionExpression: if field_name.startswith("__") and field_name.endswith("__"): raise AttributeError(field_name) _attrexpr = _expr_core.GetAttrExpression(self.name, field_name) # If self is a SelectionParameter and field_name is in its # fields or encodings list, then we want to return an expression. if check_fields_and_encodings(self, field_name): return SelectionExpression(_attrexpr) return _expr_core.GetAttrExpression(self.name, field_name) # TODO: Are there any special cases to consider for __getitem__? # This was copied from v4. def __getitem__(self, field_name: str) -> GetItemExpression: return _expr_core.GetItemExpression(self.name, field_name) # Enables use of ~, &, | with compositions of selection objects. class SelectionPredicateComposition(core.PredicateComposition): def __invert__(self) -> SelectionPredicateComposition: return SelectionPredicateComposition({"not": self.to_dict()}) def __and__(self, other: SchemaBase) -> SelectionPredicateComposition: return SelectionPredicateComposition({"and": [self.to_dict(), other.to_dict()]}) def __or__(self, other: SchemaBase) -> SelectionPredicateComposition: return SelectionPredicateComposition({"or": [self.to_dict(), other.to_dict()]}) class ParameterExpression(_expr_core.OperatorMixin): def __init__(self, expr: IntoExpression) -> None: self.expr = expr def to_dict(self) -> dict[str, str]: return {"expr": repr(self.expr)} def _to_expr(self) -> str: return repr(self.expr) def _from_expr(self, expr: IntoExpression) -> ParameterExpression: return ParameterExpression(expr=expr) class SelectionExpression(_expr_core.OperatorMixin): def __init__(self, expr: IntoExpression) -> None: self.expr = expr def to_dict(self) -> dict[str, str]: return {"expr": repr(self.expr)} def _to_expr(self) -> str: return repr(self.expr) def _from_expr(self, expr: IntoExpression) -> SelectionExpression: return SelectionExpression(expr=expr) def check_fields_and_encodings(parameter: Parameter, field_name: str) -> bool: param = parameter.param if utils.is_undefined(param) or isinstance(param, core.VariableParameter): return False for prop in ["fields", "encodings"]: try: if field_name in getattr(param.select, prop): return True except (AttributeError, TypeError): pass return False # ------------------------------------------------------------------------- # Tools for working with conditions _TestPredicateType: TypeAlias = Union[ str, _expr_core.Expression, core.PredicateComposition ] """https://vega.github.io/vega-lite/docs/predicate.html""" _PredicateType: TypeAlias = Union[ Parameter, core.Expr, Map, _TestPredicateType, _expr_core.OperatorMixin, ] """Permitted types for `predicate`.""" _ComposablePredicateType: TypeAlias = Union[ _expr_core.OperatorMixin, SelectionPredicateComposition ] """Permitted types for `&` reduced predicates.""" _StatementType: TypeAlias = Union[core.SchemaBase, Map, str] """Permitted types for `if_true`/`if_false`. In python terms: ```py if _PredicateType: return _StatementType elif _PredicateType: return _StatementType else: return _StatementType ``` """ _ConditionType: TypeAlias = t.Dict[str, Union[_TestPredicateType, Any]] """Intermediate type representing a converted `_PredicateType`. Prior to parsing any `_StatementType`. """ _LiteralValue: TypeAlias = Union[str, bool, float, int] """Primitive python value types.""" _FieldEqualType: TypeAlias = Union[_LiteralValue, Map, Parameter, core.SchemaBase] """Permitted types for equality checks on field values: - `datum.field == ...` - `FieldEqualPredicate(equal=...)` - `when(**constraints=...)` """ def _is_test_predicate(obj: Any) -> TypeIs[_TestPredicateType]: return isinstance(obj, (str, _expr_core.Expression, core.PredicateComposition)) def _get_predicate_expr(p: Parameter) -> Optional[str | SchemaBase]: # https://vega.github.io/vega-lite/docs/predicate.html return getattr(p.param, "expr", Undefined) def _predicate_to_condition( predicate: _PredicateType, *, empty: Optional[bool] = Undefined ) -> _ConditionType: condition: _ConditionType if isinstance(predicate, Parameter): predicate_expr = _get_predicate_expr(predicate) if predicate.param_type == "selection" or utils.is_undefined(predicate_expr): condition = {"param": predicate.name} if isinstance(empty, bool): condition["empty"] = empty elif isinstance(predicate.empty, bool): condition["empty"] = predicate.empty else: condition = {"test": predicate_expr} elif _is_test_predicate(predicate): condition = {"test": predicate} elif isinstance(predicate, dict): condition = predicate elif isinstance(predicate, _expr_core.OperatorMixin): condition = {"test": predicate._to_expr()} else: msg = ( f"Expected a predicate, but got: {type(predicate).__name__!r}\n\n" f"From `predicate={predicate!r}`." ) raise TypeError(msg) return condition def _condition_to_selection( condition: _ConditionType, if_true: _StatementType, if_false: _StatementType, **kwargs: Any, ) -> SchemaBase | dict[str, _ConditionType | Any]: selection: SchemaBase | dict[str, _ConditionType | Any] if isinstance(if_true, core.SchemaBase): if_true = if_true.to_dict() elif isinstance(if_true, str): if isinstance(if_false, str): msg = ( "A field cannot be used for both the `if_true` and `if_false` " "values of a condition. " "One of them has to specify a `value` or `datum` definition." ) raise ValueError(msg) else: if_true = utils.parse_shorthand(if_true) if_true.update(kwargs) condition.update(if_true) if isinstance(if_false, core.SchemaBase): # For the selection, the channel definitions all allow selections # already. So use this SchemaBase wrapper if possible. selection = if_false.copy() selection.condition = condition elif isinstance(if_false, (str, dict)): if isinstance(if_false, str): if_false = utils.parse_shorthand(if_false) if_false.update(kwargs) selection = dict(condition=condition, **if_false) else: raise TypeError(if_false) return selection class _ConditionClosed(TypedDict, closed=True, total=False): # type: ignore[call-arg] # https://peps.python.org/pep-0728/ # Parameter {"param", "value", "empty"} # Predicate {"test", "value"} empty: Optional[bool] param: Parameter | str test: _TestPredicateType value: Any class _ConditionExtra(TypedDict, closed=True, total=False): # type: ignore[call-arg] # https://peps.python.org/pep-0728/ # Likely a Field predicate empty: Optional[bool] param: Parameter | str test: _TestPredicateType value: Any __extra_items__: _StatementType | OneOrSeq[_LiteralValue] _Condition: TypeAlias = _ConditionExtra """A singular, non-chainable condition produced by ``.when()``.""" _Conditions: TypeAlias = t.List[_ConditionClosed] """Chainable conditions produced by ``.when()`` and ``Then.when()``.""" _C = TypeVar("_C", _Conditions, _Condition) class _Conditional(TypedDict, t.Generic[_C], total=False): condition: Required[_C] value: Any class _Value(TypedDict, closed=True, total=False): # type: ignore[call-arg] # https://peps.python.org/pep-0728/ value: Required[Any] __extra_items__: Any def _reveal_parsed_shorthand(obj: Map, /) -> dict[str, Any]: # Helper for producing error message on multiple field collision. return {k: v for k, v in obj.items() if k in utils.SHORTHAND_KEYS} def _is_extra(*objs: Any, kwds: Map) -> Iterator[bool]: for el in objs: if isinstance(el, (core.SchemaBase, t.Mapping)): item = el.to_dict(validate=False) if isinstance(el, core.SchemaBase) else el yield not (item.keys() - kwds.keys()).isdisjoint(utils.SHORTHAND_KEYS) else: continue def _is_condition_extra(obj: Any, *objs: Any, kwds: Map) -> TypeIs[_Condition]: # NOTE: Short circuits on the first conflict. # 1 - Originated from parse_shorthand # 2 - Used a wrapper or `dict` directly, including `extra_keys` return isinstance(obj, str) or any(_is_extra(obj, *objs, kwds=kwds)) def _parse_when_constraints( constraints: dict[str, _FieldEqualType], / ) -> Iterator[BinaryExpression]: """ Wrap kwargs with `alt.datum`. ```py # before alt.when(alt.datum.Origin == "Europe") # after alt.when(Origin="Europe") ``` """ for name, value in constraints.items(): yield _expr_core.GetAttrExpression("datum", name) == value def _validate_composables( predicates: Iterable[Any], / ) -> Iterator[_ComposablePredicateType]: for p in predicates: if isinstance(p, (_expr_core.OperatorMixin, SelectionPredicateComposition)): yield p else: msg = ( f"Predicate composition is not permitted for " f"{type(p).__name__!r}.\n" f"Try wrapping {p!r} in a `Parameter` first." ) raise TypeError(msg) def _parse_when_compose( predicates: tuple[Any, ...], constraints: dict[str, _FieldEqualType], /, ) -> BinaryExpression: """ Compose an `&` reduction predicate. Parameters ---------- predicates Collected positional arguments. constraints Collected keyword arguments. Raises ------ TypeError On the first non ``_ComposablePredicateType`` of `predicates` """ iters = [] if predicates: iters.append(_validate_composables(predicates)) if constraints: iters.append(_parse_when_constraints(constraints)) r = functools.reduce(operator.and_, itertools.chain.from_iterable(iters)) return t.cast(_expr_core.BinaryExpression, r) def _parse_when( predicate: Optional[_PredicateType], *more_predicates: _ComposablePredicateType, empty: Optional[bool], **constraints: _FieldEqualType, ) -> _ConditionType: composed: _PredicateType if utils.is_undefined(predicate): if more_predicates or constraints: composed = _parse_when_compose(more_predicates, constraints) else: msg = ( f"At least one predicate or constraint must be provided, " f"but got: {predicate=}" ) raise TypeError(msg) elif more_predicates or constraints: predicates = predicate, *more_predicates composed = _parse_when_compose(predicates, constraints) else: composed = predicate return _predicate_to_condition(composed, empty=empty) def _parse_literal(val: Any, /) -> dict[str, Any]: if isinstance(val, str): return utils.parse_shorthand(val) else: msg = ( f"Expected a shorthand `str`, but got: {type(val).__name__!r}\n\n" f"From `statement={val!r}`." ) raise TypeError(msg) def _parse_then(statement: _StatementType, kwds: dict[str, Any], /) -> dict[str, Any]: if isinstance(statement, core.SchemaBase): statement = statement.to_dict() elif not isinstance(statement, dict): statement = _parse_literal(statement) statement.update(kwds) return statement def _parse_otherwise( statement: _StatementType, conditions: _Conditional[Any], kwds: dict[str, Any], / ) -> SchemaBase | _Conditional[Any]: selection: SchemaBase | _Conditional[Any] if isinstance(statement, core.SchemaBase): selection = statement.copy() conditions.update(**kwds) # type: ignore[call-arg] selection.condition = conditions["condition"] else: if not isinstance(statement, t.Mapping): statement = _parse_literal(statement) selection = conditions selection.update(**statement, **kwds) # type: ignore[call-arg] return selection class _BaseWhen(Protocol): # NOTE: Temporary solution to non-SchemaBase copy _condition: _ConditionType def _when_then( self, statement: _StatementType, kwds: dict[str, Any], / ) -> _ConditionClosed | _Condition: condition: Any = _deepcopy(self._condition) then = _parse_then(statement, kwds) condition.update(then) return condition class When(_BaseWhen): """ Utility class for ``when-then-otherwise`` conditions. Represents the state after calling :func:`.when()`. This partial state requires calling :meth:`When.then()` to finish the condition. References ---------- `polars.when `__ """ def __init__(self, condition: _ConditionType, /) -> None: self._condition = condition def __repr__(self) -> str: return f"{type(self).__name__}({self._condition!r})" @overload def then(self, statement: str, /, **kwds: Any) -> Then[_Condition]: ... @overload def then(self, statement: _Value, /, **kwds: Any) -> Then[_Conditions]: ... @overload def then( self, statement: dict[str, Any] | SchemaBase, /, **kwds: Any ) -> Then[Any]: ... def then(self, statement: _StatementType, /, **kwds: Any) -> Then[Any]: """ Attach a statement to this predicate. Parameters ---------- statement A spec or value to use when the preceding :func:`.when()` clause is true. .. note:: ``str`` will be encoded as `shorthand`__. **kwds Additional keyword args are added to the resulting ``dict``. Returns ------- :class:`Then` Examples -------- Simple conditions may be expressed without defining a default:: import altair as alt from vega_datasets import data source = data.movies() predicate = (alt.datum.IMDB_Rating == None) | (alt.datum.Rotten_Tomatoes_Rating == None) alt.Chart(source).mark_point(invalid=None).encode( x="IMDB_Rating:Q", y="Rotten_Tomatoes_Rating:Q", color=alt.when(predicate).then(alt.value("grey")), ) """ condition = self._when_then(statement, kwds) if _is_condition_extra(condition, statement, kwds=kwds): return Then(_Conditional(condition=condition)) else: return Then(_Conditional(condition=[condition])) class Then(core.SchemaBase, t.Generic[_C]): """ Utility class for ``when-then-otherwise`` conditions. Represents the state after calling :func:`.when().then()`. This state is a valid condition on its own. It can be further specified, via multiple chained `when-then` calls, or finalized with :meth:`Then.otherwise()`. References ---------- `polars.when `__ """ _schema = {"type": "object"} def __init__(self, conditions: _Conditional[_C], /) -> None: super().__init__(**conditions) self.condition: _C @overload def otherwise(self, statement: _TSchemaBase, /, **kwds: Any) -> _TSchemaBase: ... @overload def otherwise(self, statement: str, /, **kwds: Any) -> _Conditional[_Condition]: ... @overload def otherwise( self, statement: _Value, /, **kwds: Any ) -> _Conditional[_Conditions]: ... @overload def otherwise( self, statement: dict[str, Any], /, **kwds: Any ) -> _Conditional[Any]: ... def otherwise( self, statement: _StatementType, /, **kwds: Any ) -> SchemaBase | _Conditional[Any]: """ Finalize the condition with a default value. Parameters ---------- statement A spec or value to use when no predicates were met. .. note:: Roughly equivalent to an ``else`` clause. .. note:: ``str`` will be encoded as `shorthand`__. **kwds Additional keyword args are added to the resulting ``dict``. Examples -------- Points outside of ``brush`` will not appear highlighted:: import altair as alt from vega_datasets import data source = data.cars() brush = alt.selection_interval() color = alt.when(brush).then("Origin:N").otherwise(alt.value("grey")) alt.Chart(source).mark_point().encode( x="Horsepower:Q", y="Miles_per_Gallon:Q", color=color, ).add_params(brush) """ conditions: _Conditional[Any] is_extra = functools.partial(_is_condition_extra, kwds=kwds) if is_extra(self.condition, statement): current = self.condition if isinstance(current, list) and len(current) == 1: # This case is guaranteed to have come from `When` and not `ChainedWhen` # The `list` isn't needed if we complete the condition here conditions = _Conditional(condition=current[0]) # pyright: ignore[reportArgumentType] elif isinstance(current, dict): if not is_extra(statement): conditions = self.to_dict() else: cond = _reveal_parsed_shorthand(current) msg = ( f"Only one field may be used within a condition.\n" f"Shorthand {statement!r} would conflict with {cond!r}\n\n" f"Use `alt.value({statement!r})` if this is not a shorthand string." ) raise TypeError(msg) else: # Generic message to cover less trivial cases msg = ( f"Chained conditions cannot be mixed with field conditions.\n" f"{self!r}\n\n{statement!r}" ) raise TypeError(msg) else: conditions = self.to_dict() return _parse_otherwise(statement, conditions, kwds) def when( self, predicate: Optional[_PredicateType] = Undefined, *more_predicates: _ComposablePredicateType, empty: Optional[bool] = Undefined, **constraints: _FieldEqualType, ) -> ChainedWhen: """ Attach another predicate to the condition. The resulting predicate is an ``&`` reduction over ``predicate`` and optional ``*``, ``**``, arguments. Parameters ---------- predicate A selection or test predicate. ``str`` input will be treated as a test operand. .. note:: Accepts the same range of inputs as in :func:`.condition()`. *more_predicates Additional predicates, restricted to types supporting ``&``. empty For selection parameters, the predicate of empty selections returns ``True`` by default. Override this behavior, with ``empty=False``. .. note:: When ``predicate`` is a ``Parameter`` that is used more than once, ``alt.when().then().when(..., empty=...)`` provides granular control for each occurrence. **constraints Specify `Field Equal Predicate `__'s. Shortcut for ``alt.datum.field_name == value``, see examples for usage. Returns ------- :class:`ChainedWhen` A partial state which requires calling :meth:`ChainedWhen.then()` to finish the condition. Examples -------- Chain calls to express precise queries:: import altair as alt from vega_datasets import data source = data.cars() color = ( alt.when(alt.datum.Miles_per_Gallon >= 30, Origin="Europe") .then(alt.value("crimson")) .when(alt.datum.Horsepower > 150) .then(alt.value("goldenrod")) .otherwise(alt.value("grey")) ) alt.Chart(source).mark_point().encode(x="Horsepower", y="Miles_per_Gallon", color=color) """ condition = _parse_when(predicate, *more_predicates, empty=empty, **constraints) conditions = self.to_dict() current = conditions["condition"] if isinstance(current, list): conditions = t.cast(_Conditional[_Conditions], conditions) return ChainedWhen(condition, conditions) elif isinstance(current, dict): cond = _reveal_parsed_shorthand(current) msg = ( f"Chained conditions cannot be mixed with field conditions.\n" f"Additional conditions would conflict with {cond!r}\n\n" f"Must finalize by calling `.otherwise()`." ) raise TypeError(msg) else: msg = ( f"The internal structure has been modified.\n" f"{type(current).__name__!r} found, but only `dict | list` are valid." ) raise NotImplementedError(msg) def to_dict(self, *args: Any, **kwds: Any) -> _Conditional[_C]: # type: ignore[override] m = super().to_dict(*args, **kwds) return _Conditional(condition=m["condition"]) def __deepcopy__(self, memo: Any) -> Self: return type(self)(_Conditional(condition=_deepcopy(self.condition))) class ChainedWhen(_BaseWhen): """ Utility class for ``when-then-otherwise`` conditions. Represents the state after calling :func:`.when().then().when()`. This partial state requires calling :meth:`ChainedWhen.then()` to finish the condition. References ---------- `polars.when `__ """ def __init__( self, condition: _ConditionType, conditions: _Conditional[_Conditions], /, ) -> None: self._condition = condition self._conditions = conditions def __repr__(self) -> str: return ( f"{type(self).__name__}(\n" f" {self._conditions!r},\n {self._condition!r}\n" ")" ) def then(self, statement: _StatementType, /, **kwds: Any) -> Then[_Conditions]: """ Attach a statement to this predicate. Parameters ---------- statement A spec or value to use when the preceding :meth:`Then.when()` clause is true. .. note:: ``str`` will be encoded as `shorthand`__. **kwds Additional keyword args are added to the resulting ``dict``. Returns ------- :class:`Then` Examples -------- Multiple conditions with an implicit default:: import altair as alt from vega_datasets import data source = data.movies() predicate = (alt.datum.IMDB_Rating == None) | (alt.datum.Rotten_Tomatoes_Rating == None) color = ( alt.when(predicate) .then(alt.value("grey")) .when(alt.datum.IMDB_Votes < 5000) .then(alt.value("lightblue")) ) alt.Chart(source).mark_point(invalid=None).encode( x="IMDB_Rating:Q", y="Rotten_Tomatoes_Rating:Q", color=color ) """ condition = self._when_then(statement, kwds) conditions = self._conditions.copy() conditions["condition"].append(condition) return Then(conditions) def when( predicate: Optional[_PredicateType] = Undefined, *more_predicates: _ComposablePredicateType, empty: Optional[bool] = Undefined, **constraints: _FieldEqualType, ) -> When: """ Start a ``when-then-otherwise`` condition. The resulting predicate is an ``&`` reduction over ``predicate`` and optional ``*``, ``**``, arguments. Parameters ---------- predicate A selection or test predicate. ``str`` input will be treated as a test operand. .. note:: Accepts the same range of inputs as in :func:`.condition()`. *more_predicates Additional predicates, restricted to types supporting ``&``. empty For selection parameters, the predicate of empty selections returns ``True`` by default. Override this behavior, with ``empty=False``. .. note:: When ``predicate`` is a ``Parameter`` that is used more than once, ``alt.when(..., empty=...)`` provides granular control for each occurrence. **constraints Specify `Field Equal Predicate `__'s. Shortcut for ``alt.datum.field_name == value``, see examples for usage. Returns ------- :class:`When` A partial state which requires calling :meth:`When.then()` to finish the condition. Notes ----- - Directly inspired by the ``when-then-otherwise`` syntax used in ``polars.when``. References ---------- `polars.when `__ Examples -------- Setting up a common chart:: import altair as alt from vega_datasets import data source = data.cars() brush = alt.selection_interval() points = ( alt.Chart(source) .mark_point() .encode(x="Horsepower", y="Miles_per_Gallon") .add_params(brush) ) points Basic ``if-then-else`` conditions translate directly to ``when-then-otherwise``:: points.encode(color=alt.when(brush).then("Origin").otherwise(alt.value("lightgray"))) Omitting the ``.otherwise()`` clause will use the channel default instead:: points.encode(color=alt.when(brush).then("Origin")) Predicates passed as positional arguments will be reduced with ``&``:: points.encode( color=alt.when( brush, (alt.datum.Miles_per_Gallon >= 30) | (alt.datum.Horsepower >= 130) ) .then("Origin") .otherwise(alt.value("lightgray")) ) Using keyword-argument ``constraints`` can simplify compositions like:: verbose_composition = ( (alt.datum.Name == "Name_1") & (alt.datum.Color == "Green") & (alt.datum.Age == 25) & (alt.datum.StartDate == "2000-10-01") ) when_verbose = alt.when(verbose_composition) when_concise = alt.when(Name="Name_1", Color="Green", Age=25, StartDate="2000-10-01") """ condition = _parse_when(predicate, *more_predicates, empty=empty, **constraints) return When(condition) # ------------------------------------------------------------------------ # Top-Level Functions def value(value: Any, **kwargs: Any) -> _Value: """Specify a value for use in an encoding.""" return _Value(value=value, **kwargs) # type: ignore[typeddict-item] def param( name: str | None = None, value: Optional[Any] = Undefined, bind: Optional[Binding] = Undefined, empty: Optional[bool] = Undefined, expr: Optional[str | Expr | Expression] = Undefined, **kwds: Any, ) -> Parameter: """ Create a named parameter, see https://altair-viz.github.io/user_guide/interactions.html for examples. Although both variable parameters and selection parameters can be created using this 'param' function, to create a selection parameter, it is recommended to use either 'selection_point' or 'selection_interval' instead. Parameters ---------- name : string (optional) The name of the parameter. If not specified, a unique name will be created. value : any (optional) The default value of the parameter. If not specified, the parameter will be created without a default value. bind : :class:`Binding` (optional) Binds the parameter to an external input element such as a slider, selection list or radio button group. empty : boolean (optional) For selection parameters, the predicate of empty selections returns True by default. Override this behavior, by setting this property 'empty=False'. expr : str, Expression (optional) An expression for the value of the parameter. This expression may include other parameters, in which case the parameter will automatically update in response to upstream parameter changes. **kwds : additional keywords will be used to construct a parameter. If 'select' is among the keywords, then a selection parameter will be created. Otherwise, a variable parameter will be created. Returns ------- parameter: Parameter The parameter object that can be used in chart creation. """ warn_msg = "The value of `empty` should be True or False." empty_remap = {"none": False, "all": True} parameter = Parameter(name) if not utils.is_undefined(empty): if isinstance(empty, bool) and not isinstance(empty, str): parameter.empty = empty elif empty in empty_remap: utils.deprecated_warn(warn_msg, version="5.0.0") parameter.empty = empty_remap[t.cast(str, empty)] else: raise ValueError(warn_msg) if _init := kwds.pop("init", None): utils.deprecated_warn("Use `value` instead of `init`.", version="5.0.0") # If both 'value' and 'init' are set, we ignore 'init'. if value is Undefined: kwds["value"] = _init # ignore[arg-type] comment is needed because we can also pass _expr_core.Expression if "select" not in kwds: parameter.param = core.VariableParameter( name=parameter.name, bind=bind, value=value, expr=expr, **kwds, ) parameter.param_type = "variable" elif "views" in kwds: parameter.param = core.TopLevelSelectionParameter( name=parameter.name, bind=bind, value=value, expr=expr, **kwds ) parameter.param_type = "selection" else: parameter.param = core.SelectionParameter( name=parameter.name, bind=bind, value=value, expr=expr, **kwds ) parameter.param_type = "selection" return parameter def _selection(type: Optional[SelectionType_T] = Undefined, **kwds: Any) -> Parameter: # We separate out the parameter keywords from the selection keywords select_kwds = {"name", "bind", "value", "empty", "init", "views"} param_kwds = {key: kwds.pop(key) for key in select_kwds & kwds.keys()} select: IntervalSelectionConfig | PointSelectionConfig if type == "interval": select = core.IntervalSelectionConfig(type=type, **kwds) elif type == "point": select = core.PointSelectionConfig(type=type, **kwds) elif type in {"single", "multi"}: select = core.PointSelectionConfig(type="point", **kwds) utils.deprecated_warn( "The types `single` and `multi` are now combined.", version="5.0.0", alternative="selection_point()", ) else: msg = """'type' must be 'point' or 'interval'""" raise ValueError(msg) return param(select=select, **param_kwds) @utils.deprecated( version="5.0.0", alternative="'selection_point()' or 'selection_interval()'", message="These functions also include more helpful docstrings.", ) def selection(type: Optional[SelectionType_T] = Undefined, **kwds: Any) -> Parameter: """'selection' is deprecated use 'selection_point' or 'selection_interval' instead, depending on the type of parameter you want to create.""" return _selection(type=type, **kwds) def selection_interval( name: str | None = None, value: Optional[Any] = Undefined, bind: Optional[Binding | str] = Undefined, empty: Optional[bool] = Undefined, expr: Optional[str | Expr | Expression] = Undefined, encodings: Optional[list[SingleDefUnitChannel_T]] = Undefined, on: Optional[str] = Undefined, clear: Optional[str | bool] = Undefined, resolve: Optional[SelectionResolution_T] = Undefined, mark: Optional[Mark] = Undefined, translate: Optional[str | bool] = Undefined, zoom: Optional[str | bool] = Undefined, **kwds: Any, ) -> Parameter: """ Create an interval selection parameter. Selection parameters define data queries that are driven by direct manipulation from user input (e.g., mouse clicks or drags). Interval selection parameters are used to select a continuous range of data values on drag, whereas point selection parameters (`selection_point`) are used to select multiple discrete data values.). Parameters ---------- name : string (optional) The name of the parameter. If not specified, a unique name will be created. value : any (optional) The default value of the parameter. If not specified, the parameter will be created without a default value. bind : :class:`Binding`, str (optional) Binds the parameter to an external input element such as a slider, selection list or radio button group. empty : boolean (optional) For selection parameters, the predicate of empty selections returns True by default. Override this behavior, by setting this property 'empty=False'. expr : :class:`Expr` (optional) An expression for the value of the parameter. This expression may include other parameters, in which case the parameter will automatically update in response to upstream parameter changes. encodings : List[str] (optional) A list of encoding channels. The corresponding data field values must match for a data tuple to fall within the selection. on : string (optional) A Vega event stream (object or selector) that triggers the selection. For interval selections, the event stream must specify a start and end. clear : string or boolean (optional) Clears the selection, emptying it of all values. This property can be an Event Stream or False to disable clear. Default is 'dblclick'. resolve : enum('global', 'union', 'intersect') (optional) With layered and multi-view displays, a strategy that determines how selections' data queries are resolved when applied in a filter transform, conditional encoding rule, or scale domain. One of: * 'global': only one brush exists for the entire SPLOM. When the user begins to drag, any previous brushes are cleared, and a new one is constructed. * 'union': each cell contains its own brush, and points are highlighted if they lie within any of these individual brushes. * 'intersect': each cell contains its own brush, and points are highlighted only if they fall within all of these individual brushes. The default is 'global'. mark : :class:`Mark` (optional) An interval selection also adds a rectangle mark to depict the extents of the interval. The mark property can be used to customize the appearance of the mark. translate : string or boolean (optional) When truthy, allows a user to interactively move an interval selection back-and-forth. Can be True, False (to disable panning), or a Vega event stream definition which must include a start and end event to trigger continuous panning. Discrete panning (e.g., pressing the left/right arrow keys) will be supported in future versions. The default value is True, which corresponds to [pointerdown, window:pointerup] > window:pointermove! This default allows users to click and drag within an interval selection to reposition it. zoom : string or boolean (optional) When truthy, allows a user to interactively resize an interval selection. Can be True, False (to disable zooming), or a Vega event stream definition. Currently, only wheel events are supported, but custom event streams can still be used to specify filters, debouncing, and throttling. Future versions will expand the set of events that can trigger this transformation. The default value is True, which corresponds to wheel!. This default allows users to use the mouse wheel to resize an interval selection. **kwds : Additional keywords to control the selection. Returns ------- parameter: Parameter The parameter object that can be used in chart creation. """ return _selection( type="interval", name=name, value=value, bind=bind, empty=empty, expr=expr, encodings=encodings, on=on, clear=clear, resolve=resolve, mark=mark, translate=translate, zoom=zoom, **kwds, ) def selection_point( name: str | None = None, value: Optional[Any] = Undefined, bind: Optional[Binding | str] = Undefined, empty: Optional[bool] = Undefined, expr: Optional[Expr] = Undefined, encodings: Optional[list[SingleDefUnitChannel_T]] = Undefined, fields: Optional[list[str]] = Undefined, on: Optional[str] = Undefined, clear: Optional[str | bool] = Undefined, resolve: Optional[SelectionResolution_T] = Undefined, toggle: Optional[str | bool] = Undefined, nearest: Optional[bool] = Undefined, **kwds: Any, ) -> Parameter: """ Create a point selection parameter. Selection parameters define data queries that are driven by direct manipulation from user input (e.g., mouse clicks or drags). Point selection parameters are used to select multiple discrete data values; the first value is selected on click and additional values toggled on shift-click. To select a continuous range of data values on drag interval selection parameters (`selection_interval`) can be used instead. Parameters ---------- name : string (optional) The name of the parameter. If not specified, a unique name will be created. value : any (optional) The default value of the parameter. If not specified, the parameter will be created without a default value. bind : :class:`Binding`, str (optional) Binds the parameter to an external input element such as a slider, selection list or radio button group. empty : boolean (optional) For selection parameters, the predicate of empty selections returns True by default. Override this behavior, by setting this property 'empty=False'. expr : :class:`Expr` (optional) An expression for the value of the parameter. This expression may include other parameters, in which case the parameter will automatically update in response to upstream parameter changes. encodings : List[str] (optional) A list of encoding channels. The corresponding data field values must match for a data tuple to fall within the selection. fields : List[str] (optional) A list of field names whose values must match for a data tuple to fall within the selection. on : string (optional) A Vega event stream (object or selector) that triggers the selection. For interval selections, the event stream must specify a start and end. clear : string or boolean (optional) Clears the selection, emptying it of all values. This property can be an Event Stream or False to disable clear. Default is 'dblclick'. resolve : enum('global', 'union', 'intersect') (optional) With layered and multi-view displays, a strategy that determines how selections' data queries are resolved when applied in a filter transform, conditional encoding rule, or scale domain. One of: * 'global': only one brush exists for the entire SPLOM. When the user begins to drag, any previous brushes are cleared, and a new one is constructed. * 'union': each cell contains its own brush, and points are highlighted if they lie within any of these individual brushes. * 'intersect': each cell contains its own brush, and points are highlighted only if they fall within all of these individual brushes. The default is 'global'. toggle : string or boolean (optional) Controls whether data values should be toggled (inserted or removed from a point selection) or only ever inserted into point selections. One of: * True (default): the toggle behavior, which corresponds to "event.shiftKey". As a result, data values are toggled when the user interacts with the shift-key pressed. * False: disables toggling behaviour; the selection will only ever contain a single data value corresponding to the most recent interaction. * A Vega expression which is re-evaluated as the user interacts. If the expression evaluates to True, the data value is toggled into or out of the point selection. If the expression evaluates to False, the point selection is first cleared, and the data value is then inserted. For example, setting the value to the Vega expression True will toggle data values without the user pressing the shift-key. nearest : boolean (optional) When true, an invisible voronoi diagram is computed to accelerate discrete selection. The data value nearest the mouse cursor is added to the selection. The default is False, which means that data values must be interacted with directly (e.g., clicked on) to be added to the selection. **kwds : Additional keywords to control the selection. Returns ------- parameter: Parameter The parameter object that can be used in chart creation. """ return _selection( type="point", name=name, value=value, bind=bind, empty=empty, expr=expr, encodings=encodings, fields=fields, on=on, clear=clear, resolve=resolve, toggle=toggle, nearest=nearest, **kwds, ) @utils.deprecated(version="5.0.0", alternative="selection_point") def selection_multi(**kwargs: Any) -> Parameter: """'selection_multi' is deprecated. Use 'selection_point'.""" return _selection(type="point", **kwargs) @utils.deprecated(version="5.0.0", alternative="selection_point") def selection_single(**kwargs: Any) -> Parameter: """'selection_single' is deprecated. Use 'selection_point'.""" return _selection(type="point", **kwargs) @utils.use_signature(core.Binding) def binding(input: Any, **kwargs: Any) -> Binding: """A generic binding.""" return core.Binding(input=input, **kwargs) @utils.use_signature(core.BindCheckbox) def binding_checkbox(**kwargs: Any) -> BindCheckbox: """A checkbox binding.""" return core.BindCheckbox(input="checkbox", **kwargs) @utils.use_signature(core.BindRadioSelect) def binding_radio(**kwargs: Any) -> BindRadioSelect: """A radio button binding.""" return core.BindRadioSelect(input="radio", **kwargs) @utils.use_signature(core.BindRadioSelect) def binding_select(**kwargs: Any) -> BindRadioSelect: """A select binding.""" return core.BindRadioSelect(input="select", **kwargs) @utils.use_signature(core.BindRange) def binding_range(**kwargs: Any) -> BindRange: """A range binding.""" return core.BindRange(input="range", **kwargs) @overload def condition( predicate: _PredicateType, if_true: _StatementType, if_false: _TSchemaBase, *, empty: Optional[bool] = ..., **kwargs: Any, ) -> _TSchemaBase: ... @overload def condition( predicate: _PredicateType, if_true: Map | SchemaBase, if_false: Map | str, *, empty: Optional[bool] = ..., **kwargs: Any, ) -> dict[str, _ConditionType | Any]: ... @overload def condition( predicate: _PredicateType, if_true: Map | str, if_false: Map, *, empty: Optional[bool] = ..., **kwargs: Any, ) -> dict[str, _ConditionType | Any]: ... @overload def condition( predicate: _PredicateType, if_true: str, if_false: str, **kwargs: Any ) -> Never: ... # TODO: update the docstring def condition( predicate: _PredicateType, if_true: _StatementType, if_false: _StatementType, *, empty: Optional[bool] = Undefined, **kwargs: Any, ) -> SchemaBase | dict[str, _ConditionType | Any]: """ A conditional attribute or encoding. Parameters ---------- predicate: Parameter, PredicateComposition, expr.Expression, dict, or string the selection predicate or test predicate for the condition. if a string is passed, it will be treated as a test operand. if_true: the spec or object to use if the selection predicate is true if_false: the spec or object to use if the selection predicate is false empty For selection parameters, the predicate of empty selections returns ``True`` by default. Override this behavior, with ``empty=False``. .. note:: When ``predicate`` is a ``Parameter`` that is used more than once, ``alt.condition(..., empty=...)`` provides granular control for each :func:`.condition()`. **kwargs: additional keyword args are added to the resulting dict Returns ------- spec: dict or VegaLiteSchema the spec that describes the condition """ condition = _predicate_to_condition(predicate, empty=empty) return _condition_to_selection(condition, if_true, if_false, **kwargs) # -------------------------------------------------------------------- # Top-level objects def _top_schema_base( # noqa: ANN202 obj: Any, / ): # -> """ Enforces an intersection type w/ `SchemaBase` & `TopLevelMixin` objects. Use for instance methods. """ if isinstance(obj, core.SchemaBase) and isinstance(obj, TopLevelMixin): return obj else: msg = f"{type(obj).__name__!r} does not derive from {type(core.SchemaBase).__name__!r}" raise TypeError(msg) class TopLevelMixin(mixins.ConfigMethodMixin): """Mixin for top-level chart objects such as Chart, LayeredChart, etc.""" _class_is_valid_at_instantiation: bool = False data: Any def to_dict( # noqa: C901 self, validate: bool = True, *, format: str = "vega-lite", ignore: list[str] | None = None, context: dict[str, Any] | None = None, ) -> dict[str, Any]: """ Convert the chart to a dictionary suitable for JSON export. Parameters ---------- validate : bool, optional If True (default), then validate the output dictionary against the schema. format : str, optional Chart specification format, one of "vega-lite" (default) or "vega" ignore : list[str], optional A list of keys to ignore. It is usually not needed to specify this argument as a user. context : dict[str, Any], optional A context dictionary. It is usually not needed to specify this argument as a user. Notes ----- Technical: The ignore parameter will *not* be passed to child to_dict function calls. Returns ------- dict The dictionary representation of this chart Raises ------ SchemaValidationError if validate=True and the dict does not conform to the schema """ # Validate format if format not in {"vega-lite", "vega"}: msg = f'The format argument must be either "vega-lite" or "vega". Received {format!r}' raise ValueError(msg) # We make use of three context markers: # - 'data' points to the data that should be referenced for column type # inference. # - 'top_level' is a boolean flag that is assumed to be true; if it's # true then a "$schema" arg is added to the dict. # - 'datasets' is a dict of named datasets that should be inserted # in the top-level object # - 'pre_transform' whether data transformations should be pre-evaluated # if the current data transformer supports it (currently only used when # the "vegafusion" transformer is enabled) # note: not a deep copy because we want datasets and data arguments to # be passed by reference context = context.copy() if context else {} context.setdefault("datasets", {}) is_top_level = context.get("top_level", True) copy = _top_schema_base(self).copy(deep=False) original_data = getattr(copy, "data", Undefined) if not utils.is_undefined(original_data): try: data = _to_eager_narwhals_dataframe(original_data) except TypeError: # Non-narwhalifiable type supported by Altair, such as dict data = original_data copy.data = _prepare_data(data, context) context["data"] = data # remaining to_dict calls are not at top level context["top_level"] = False # TopLevelMixin instance does not necessarily have to_dict defined # but due to how Altair is set up this should hold. # Too complex to type hint right now vegalite_spec: Any = super(TopLevelMixin, copy).to_dict( # type: ignore[misc] validate=validate, ignore=ignore, context=dict(context, pre_transform=False) ) # TODO: following entries are added after validation. Should they be validated? if is_top_level: # since this is top-level we add $schema if it's missing if "$schema" not in vegalite_spec: vegalite_spec["$schema"] = SCHEMA_URL # apply theme from theme registry if theme := themes.get(): vegalite_spec = utils.update_nested(theme(), vegalite_spec, copy=True) else: msg = ( f"Expected a theme to be set but got {None!r}.\n" f"Call `themes.enable('default')` to reset the `ThemeRegistry`." ) raise TypeError(msg) # update datasets if context["datasets"]: vegalite_spec.setdefault("datasets", {}).update(context["datasets"]) if context.get("pre_transform", True) and _using_vegafusion(): if format == "vega-lite": msg = ( 'When the "vegafusion" data transformer is enabled, the \n' "to_dict() and to_json() chart methods must be called with " 'format="vega". \n' "For example: \n" ' >>> chart.to_dict(format="vega")\n' ' >>> chart.to_json(format="vega")' ) raise ValueError(msg) else: return _compile_with_vegafusion(vegalite_spec) elif format == "vega": plugin = vegalite_compilers.get() if plugin is None: msg = "No active vega-lite compiler plugin found" raise ValueError(msg) return plugin(vegalite_spec) else: return vegalite_spec def to_json( self, validate: bool = True, indent: int | str | None = 2, sort_keys: bool = True, *, format: str = "vega-lite", ignore: list[str] | None = None, context: dict[str, Any] | None = None, **kwargs: Any, ) -> str: """ Convert a chart to a JSON string. Parameters ---------- validate : bool, optional If True (default), then validate the output dictionary against the schema. indent : int, optional The number of spaces of indentation to use. The default is 2. sort_keys : bool, optional If True (default), sort keys in the output. format : str, optional The chart specification format. One of "vega-lite" (default) or "vega". The "vega" format relies on the active Vega-Lite compiler plugin, which by default requires the vl-convert-python package. ignore : list[str], optional A list of keys to ignore. It is usually not needed to specify this argument as a user. context : dict[str, Any], optional A context dictionary. It is usually not needed to specify this argument as a user. **kwargs Additional keyword arguments are passed to ``json.dumps()`` """ if ignore is None: ignore = [] if context is None: context = {} spec = self.to_dict( validate=validate, format=format, ignore=ignore, context=context ) return json.dumps(spec, indent=indent, sort_keys=sort_keys, **kwargs) def to_html( self, base_url: str = "https://cdn.jsdelivr.net/npm", output_div: str = "vis", embed_options: dict | None = None, json_kwds: dict | None = None, fullhtml: bool = True, requirejs: bool = False, inline: bool = False, **kwargs: Any, ) -> str: """ Embed a Vega/Vega-Lite spec into an HTML page. Parameters ---------- base_url : string (optional) The base url from which to load the javascript libraries. output_div : string (optional) The id of the div element where the plot will be shown. embed_options : dict (optional) Dictionary of options to pass to the vega-embed script. Default entry is {'mode': mode}. json_kwds : dict (optional) Dictionary of keywords to pass to json.dumps(). fullhtml : boolean (optional) If True (default) then return a full html page. If False, then return an HTML snippet that can be embedded into an HTML page. requirejs : boolean (optional) If False (default) then load libraries from base_url using