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 collections.abc import Mapping, Sequence from copy import deepcopy as _deepcopy from typing import TYPE_CHECKING, Any, Literal, TypeVar, Union, overload import jsonschema import narwhals.stable.v1 as nw from altair import theme, utils from altair.expr import core as _expr_core from altair.utils import Optional, SchemaBase, 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.data import DataType from altair.utils.data import is_data_type as _is_data_type from altair.utils.schemapi import ConditionLike, _TypeMap 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, PrimitiveValue_T, SingleDefUnitChannel_T, Temporal if sys.version_info >= (3, 14): from typing import TypedDict else: from typing_extensions import TypedDict if sys.version_info >= (3, 12): from typing import Protocol, TypeAliasType, runtime_checkable else: from typing_extensions import ( # noqa: F401 Protocol, TypeAliasType, runtime_checkable, ) if sys.version_info >= (3, 11): from typing import LiteralString else: from typing_extensions import LiteralString if sys.version_info >= (3, 10): from typing import TypeAlias else: from typing_extensions import TypeAlias if TYPE_CHECKING: from collections.abc import Iterable, Iterator from pathlib import Path from typing import IO 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._config import BrushConfigKwds, DerivedStreamKwds, MergedStreamKwds 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, SingleTimeUnit_T, StackOffset_T, ) from .schema.channels import Column, Facet, Row from .schema.core import ( AggregatedFieldDef, AggregateOp, AnyMark, BindCheckbox, Binding, BindInput, BindRadioSelect, BindRange, BinParams, BrushConfig, DateTime, Expr, ExprRef, FacetedEncoding, FacetFieldDef, FieldName, GraticuleGenerator, ImputeMethod, ImputeSequence, InlineData, InlineDataset, IntervalSelectionConfig, JoinAggregateFieldDef, LayerRepeatMapping, LookupSelection, NamedData, ParameterName, PointSelectionConfig, PredicateComposition, ProjectionType, RepeatMapping, RepeatRef, 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=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, pass_through=True)) # 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): """ FacetMapping schema wrapper. Parameters ---------- column : str, :class:`FacetFieldDef`, :class:`Column` A field definition for the horizontal facet of trellis plots. row : str, :class:`FacetFieldDef`, :class:`Row` A field definition for the vertical facet of trellis plots. """ _class_is_valid_at_instantiation = False def __init__( self, column: Optional[str | FacetFieldDef | Column] = Undefined, row: Optional[str | FacetFieldDef | Row] = Undefined, **kwargs: Any, ) -> None: super().__init__(column=column, row=row, **kwargs) # type: ignore[arg-type] 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.""" _schema: t.ClassVar[_TypeMap[Literal["object"]]] = {"type": "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) -> PredicateComposition | Any: if self.param_type == "selection": return core.PredicateComposition({"not": {"param": self.name}}) else: return _expr_core.OperatorMixin.__invert__(self) def __and__(self, other: Any) -> PredicateComposition | Any: if self.param_type == "selection": if isinstance(other, Parameter): other = {"param": other.name} return core.PredicateComposition({"and": [{"param": self.name}, other]}) else: return _expr_core.OperatorMixin.__and__(self, other) def __or__(self, other: Any) -> PredicateComposition | Any: if self.param_type == "selection": if isinstance(other, Parameter): other = {"param": other.name} return core.PredicateComposition({"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. SelectionPredicateComposition = core.PredicateComposition class ParameterExpression(_expr_core.OperatorMixin): _schema: t.ClassVar[_TypeMap[Literal["object"]]] = {"type": "object"} 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): _schema: t.ClassVar[_TypeMap[Literal["object"]]] = {"type": "object"} 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, "_ConditionExtra", _TestPredicateType, _expr_core.OperatorMixin, ] """Permitted types for `predicate`.""" _ComposablePredicateType: TypeAlias = Union[ _expr_core.OperatorMixin, core.PredicateComposition ] """Permitted types for `&` reduced predicates.""" _StatementType: TypeAlias = Union[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 ``` """ _FieldEqualType: TypeAlias = Union["IntoExpression", Parameter, SchemaBase] """ Permitted types for equality checks on field values. Applies to the following context(s): import altair as alt alt.datum.field == ... alt.FieldEqualPredicate(field="field", equal=...) alt.when(field=...) alt.when().then().when(field=...) alt.Chart.transform_filter(field=...) """ def _is_test_predicate(obj: Any) -> TypeIs[_TestPredicateType]: return isinstance(obj, (str, _expr_core.Expression, core.PredicateComposition)) def _get_predicate_expr(p: Parameter) -> Optional[_TestPredicateType]: # 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 ) -> _Condition: condition: _Condition 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: _Condition, if_true: _StatementType, if_false: _StatementType, **kwargs: Any, ) -> SchemaBase | _Conditional[_Condition]: selection: SchemaBase | _Conditional[_Condition] if isinstance(if_true, 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) cond_mutable: Any = dict(condition) cond_mutable.update(if_true) if isinstance(if_false, SchemaBase): # For the selection, the channel definitions all allow selections # already. So use this SchemaBase wrapper if possible. selection = if_false.copy() selection.condition = cond_mutable elif isinstance(if_false, (str, dict)): if isinstance(if_false, str): if_false = utils.parse_shorthand(if_false) if_false.update(kwargs) selection = _Conditional(condition=cond_mutable, **if_false) # type: ignore[typeddict-item] else: raise TypeError(if_false) return selection 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[PrimitiveValue_T] _Condition: TypeAlias = _ConditionExtra """ A singular, *possibly* non-chainable condition produced by ``.when()``. The default **permissive** representation. Allows arbitrary additional keys that *may* be present in a `Conditional Field`_ but not a `Conditional Value`_. .. _Conditional Field: https://vega.github.io/vega-lite/docs/condition.html#field .. _Conditional Value: https://vega.github.io/vega-lite/docs/condition.html#value """ 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 _Conditions: TypeAlias = list[_ConditionClosed] """ Chainable conditions produced by ``.when()`` and ``Then.when()``. All must be a `Conditional Value`_. .. _Conditional Value: https://vega.github.io/vega-lite/docs/condition.html#value """ _C = TypeVar("_C", _Conditions, _Condition) class _Conditional(TypedDict, t.Generic[_C], total=False): """ A dictionary representation of a conditional encoding or property. Parameters ---------- condition One or more (predicate, statement) pairs which each form a condition. value An optional default value, used when no predicates were met. """ condition: Required[_C] value: Any IntoCondition: TypeAlias = Union[ConditionLike, _Conditional[Any]] """ Anything that can be converted into a conditional encoding or property. Notes ----- Represents all outputs from `when-then-otherwise` conditions, which are not ``SchemaBase`` types. """ 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, (SchemaBase, t.Mapping)): item = el.to_dict(validate=False) if isinstance(el, 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 _is_condition_closed(obj: Map) -> TypeIs[_ConditionClosed]: """Return `True` if ``obj`` can be used in a chained condition.""" return {"empty", "param", "test", "value"} >= obj.keys() 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, core.PredicateComposition)): 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, ) -> _Condition: 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, 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, 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: _Condition 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: _Condition, /) -> 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` .. _shorthand: https://altair-viz.github.io/user_guide/encodings/index.html#encoding-shorthands 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(ConditionLike, 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 `__ """ def __init__(self, conditions: _Conditional[_C], /) -> None: self.condition: _C = conditions["condition"] @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``. .. _shorthand: https://altair-viz.github.io/user_guide/encodings/index.html#encoding-shorthands 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. .. _Field Equal Predicate: https://vega.github.io/vega-lite/docs/predicate.html#equal-predicate 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]: return _Conditional(condition=self.condition.copy()) def __deepcopy__(self, memo: Any) -> Self: return type(self)(_Conditional(condition=_deepcopy(self.condition, memo))) def __repr__(self) -> str: name = type(self).__name__ COND = "condition: " LB, RB = "{", "}" if len(self.condition) == 1: args = f"{COND}{self.condition!r}".replace("\n", "\n ") else: conds = "\n ".join(f"{c!r}" for c in self.condition) args = f"{COND}[\n " f"{conds}\n ]" return f"{name}({LB}\n {args}\n{RB})" 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: _Condition, 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` .. _shorthand: https://altair-viz.github.io/user_guide/encodings/index.html#encoding-shorthands 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) if _is_condition_closed(condition): conditions = self._conditions.copy() conditions["condition"].append(condition) return Then(conditions) else: cond = _reveal_parsed_shorthand(condition) msg = ( f"Chained conditions cannot be mixed with field conditions.\n" f"Shorthand {statement!r} expanded to {cond!r}\n\n" f"Use `alt.value({statement!r})` if this is not a shorthand string." ) raise TypeError(msg) 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`_. .. _Field Equal Predicate: https://vega.github.io/vega-lite/docs/predicate.html#equal-predicate .. _polars.when: https://docs.pola.rs/py-polars/html/reference/expressions/api/polars.when.html 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) _SelectionPointValue: TypeAlias = "PrimitiveValue_T | Temporal | DateTime | Sequence[Mapping[SingleDefUnitChannel_T | LiteralString, PrimitiveValue_T | Temporal | DateTime]]" """ Point selections can be initialized with a single primitive value: import altair as alt alt.selection_point(fields=["year"], value=1980) You can also provide a sequence of mappings between ``encodings`` or ``fields`` to **values**: alt.selection_point( fields=["cylinders", "year"], value=[{"cylinders": 4, "year": 1981}, {"cylinders": 8, "year": 1972}], ) """ _SelectionIntervalValueMap: TypeAlias = Mapping[ SingleDefUnitChannel_T, Union[ tuple[bool, bool], tuple[float, float], tuple[str, str], tuple["Temporal | DateTime", "Temporal | DateTime"], Sequence[bool], Sequence[float], Sequence[str], Sequence["Temporal | DateTime"], ], ] """ Interval selections are initialized with a mapping between ``encodings`` to **values**: import altair as alt alt.selection_interval( encodings=["longitude"], empty=False, value={"longitude": [-50, -110]}, ) The values specify the **start** and **end** of the interval selection. You can use a ``tuple`` for type-checking each sequence has **two** elements: alt.selection_interval(value={"x": (55, 160), "y": (13, 37)}) .. note:: Unlike :func:`.selection_point()`, the use of ``None`` is not permitted. """ def selection_interval( name: str | None = None, value: Optional[_SelectionIntervalValueMap] = Undefined, bind: Optional[Binding | str] = Undefined, empty: Optional[bool] = Undefined, expr: Optional[str | Expr | Expression] = Undefined, encodings: Optional[Sequence[SingleDefUnitChannel_T]] = Undefined, on: Optional[str | MergedStreamKwds | DerivedStreamKwds] = Undefined, clear: Optional[str | bool | MergedStreamKwds | DerivedStreamKwds] = Undefined, resolve: Optional[SelectionResolution_T] = Undefined, mark: Optional[BrushConfig | BrushConfigKwds] = 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 : str (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 : bool (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 : Sequence[str] (optional) A list of encoding channels. The corresponding data field values must match for a data tuple to fall within the selection. on : str (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 : str, bool (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 : Literal['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:`BrushConfig` (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 : str, bool (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 : str, bool (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 : Any 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[_SelectionPointValue] = Undefined, bind: Optional[Binding | str] = Undefined, empty: Optional[bool] = Undefined, expr: Optional[str | Expr | Expression] = Undefined, encodings: Optional[Sequence[SingleDefUnitChannel_T]] = Undefined, fields: Optional[Sequence[str]] = Undefined, on: Optional[str | MergedStreamKwds | DerivedStreamKwds] = Undefined, clear: Optional[str | bool | MergedStreamKwds | DerivedStreamKwds] = 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 : str (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 : bool (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 : Sequence[str] (optional) A list of encoding channels. The corresponding data field values must match for a data tuple to fall within the selection. fields : Sequence[str] (optional) A list of field names whose values must match for a data tuple to fall within the selection. on : str (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 : str, bool (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 : Literal['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 : str, bool (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 : bool (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 : Any 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) def binding( input: str, *, autocomplete: Optional[str] = Undefined, debounce: Optional[float] = Undefined, element: Optional[str] = Undefined, name: Optional[str] = Undefined, placeholder: Optional[str] = Undefined, ) -> BindInput: """ A generic binding. Parameters ---------- input : str The type of input element to use. The valid values are ``"checkbox"``, ``"radio"``, ``"range"``, ``"select"``, and any other legal `HTML form input type `__. autocomplete : str A hint for form autofill. See the `HTML autocomplete attribute `__ for additional information. debounce : float If defined, delays event handling until the specified milliseconds have elapsed since the last event was fired. element : str An optional CSS selector string indicating the parent element to which the input element should be added. By default, all input elements are added within the parent container of the Vega view. name : str By default, the signal name is used to label input elements. This ``name`` property can be used instead to specify a custom label for the bound signal. placeholder : str Text that appears in the form control when it has no value set. """ return core.BindInput( autocomplete=autocomplete, debounce=debounce, element=element, input=input, name=name, placeholder=placeholder, ) @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, ) -> _Conditional[_Condition]: ... @overload def condition( predicate: _PredicateType, if_true: Map | str, if_false: Map, *, empty: Optional[bool] = ..., **kwargs: Any, ) -> _Conditional[_Condition]: ... @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 | _Conditional[_Condition]: """ 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 methods, called from `TopLevelMixin` that are defined in `SchemaBase`. Notes ----- - The `super` sub-branch is not statically checked *here*. - It would widen the inferred intersection to: - `( | super)` - Both dunder attributes are not in the `super` type stubs - Requiring 2x *# type: ignore[attr-defined]* - However it is required at runtime for any cases that use `super(..., copy)`. - The inferred type **is** used statically **outside** of this function. """ if (isinstance(obj, SchemaBase) and isinstance(obj, TopLevelMixin)) or ( not TYPE_CHECKING and ( isinstance(obj, super) and issubclass(obj.__self_class__, SchemaBase) and obj.__thisclass__ is TopLevelMixin ) ): return obj else: msg = f"{type(obj).__name__!r} does not derive from {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: Literal["vega-lite", "vega"] = "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 result against the schema. format : {"vega-lite", "vega"}, optional The chart specification format. 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. context : dict[str, Any], optional A context dictionary. Raises ------ SchemaValidationError : If ``validate`` and the result does not conform to the schema. Notes ----- - ``ignore``, ``context`` are usually not needed to be specified as a user. - *Technical*: ``ignore`` will **not** be passed to child :meth:`.to_dict()`. """ # 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 = nw.from_native(original_data, eager_or_interchange_only=True) 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 vegalite_spec: Any = _top_schema_base(super(TopLevelMixin, copy)).to_dict( 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 if func := theme.get(): vegalite_spec = utils.update_nested(func(), 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: Literal["vega-lite", "vega"] = "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 result 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 : {"vega-lite", "vega"}, optional The chart specification format. 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. context : dict[str, Any], optional A context dictionary. **kwargs Additional keyword arguments are passed to ``json.dumps()`` Raises ------ SchemaValidationError : If ``validate`` and the result does not conform to the schema. Notes ----- - ``ignore``, ``context`` are usually not needed to be specified as a user. - *Technical*: ``ignore`` will **not** be passed to child :meth:`.to_dict()`. """ 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