import warnings import hashlib import io import json import jsonschema import pandas as pd from toolz.curried import pipe as _pipe import itertools import sys from typing import cast, List, Optional, Any, Iterable, Union, Literal, IO # Have to rename it here as else it overlaps with schema.core.Type and schema.core.Dict from typing import Type as TypingType from typing import Dict as TypingDict from .schema import core, channels, mixins, Undefined, UndefinedType, SCHEMA_URL from .data import data_transformers from ... import utils, expr from ...expr import core as _expr_core from .display import renderers, VEGALITE_VERSION, VEGAEMBED_VERSION, VEGA_VERSION from .theme import themes from .compiler import vegalite_compilers from ...utils._vegafusion_data import ( using_vegafusion as _using_vegafusion, compile_with_vegafusion as _compile_with_vegafusion, ) from ...utils.core import DataFrameLike from ...utils.data import DataType if sys.version_info >= (3, 11): from typing import Self else: from typing_extensions import Self ChartDataType = Union[DataType, core.Data, str, core.Generator, UndefinedType] # ------------------------------------------------------------------------ # Data Utilities def _dataset_name(values: Union[dict, list, core.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, context): """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 = Undefined kwds = {} if isinstance(data, core.InlineData): if data.name is Undefined and data.values is not Undefined: if isinstance(data.values, core.InlineDataset): values = data.to_dict()["values"] else: values = data.values kwds = {"format": data.format} elif isinstance(data, dict): if "name" not in data and "values" in data: values = data["values"] kwds = {k: v for k, v in data.items() if k != "values"} if values is not Undefined: name = _dataset_name(values) data = core.NamedData(name=name, **kwds) context.setdefault("datasets", {})[name] = values return data def _prepare_data(data, context=None): """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 isinstance(data, pd.DataFrame) or hasattr(data, "__geo_interface__"): data = _pipe(data, data_transformers.get()) # convert string input to a URLData elif isinstance(data, str): data = core.UrlData(data) elif isinstance(data, DataFrameLike): data = _pipe(data, data_transformers.get()) # 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("data of type {} not recognized".format(type(data)), 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, **kwargs): super().__init__(*args, **kwargs) def to_dict(self, *args, **kwargs) -> dict: """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, **kwargs): super().__init__(*args, **kwargs) def to_dict(self, *args, **kwargs) -> dict: 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"} def _get_channels_mapping() -> TypingDict[TypingType[core.SchemaBase], str]: mapping: TypingDict[TypingType[core.SchemaBase], str] = {} for attr in dir(channels): cls = getattr(channels, attr) if isinstance(cls, type) and issubclass(cls, core.SchemaBase): mapping[cls] = attr.replace("Value", "").lower() return mapping # ------------------------------------------------------------------------- # Tools for working with parameters class Parameter(_expr_core.OperatorMixin): """A Parameter object""" # NOTE: If you change this class, make sure that the protocol in # altair/vegalite/v5/schema/core.py is updated accordingly if needed. _counter: int = 0 @classmethod def _get_name(cls) -> str: cls._counter += 1 return f"param_{cls._counter}" def __init__( self, name: Optional[str] = None, empty: Union[bool, UndefinedType] = Undefined, param: Union[ core.VariableParameter, core.TopLevelSelectionParameter, core.SelectionParameter, UndefinedType, ] = Undefined, param_type: Union[Literal["variable", "selection"], UndefinedType] = 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.deprecation.deprecated( message="'ref' is deprecated. No need to call '.ref()' anymore." ) def ref(self) -> dict: "'ref' is deprecated. No need to call '.ref()' anymore." return self.to_dict() def to_dict(self) -> TypingDict[str, Union[str, dict]]: if self.param_type == "variable": return {"expr": self.name} elif self.param_type == "selection": return { "param": self.name.to_dict() if hasattr(self.name, "to_dict") else self.name } else: raise ValueError(f"Unrecognized parameter type: {self.param_type}") def __invert__(self): if self.param_type == "selection": return SelectionPredicateComposition({"not": {"param": self.name}}) else: return _expr_core.OperatorMixin.__invert__(self) def __and__(self, other): 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): 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 "Parameter({0!r}, {1})".format(self.name, self.param) def _to_expr(self) -> str: return self.name def _from_expr(self, expr) -> "ParameterExpression": return ParameterExpression(expr=expr) def __getattr__( self, field_name: str ) -> Union[_expr_core.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) -> _expr_core.GetItemExpression: return _expr_core.GetItemExpression(self.name, field_name) # Enables use of ~, &, | with compositions of selection objects. class SelectionPredicateComposition(core.PredicateComposition): def __invert__(self): return SelectionPredicateComposition({"not": self.to_dict()}) def __and__(self, other): return SelectionPredicateComposition({"and": [self.to_dict(), other.to_dict()]}) def __or__(self, other): return SelectionPredicateComposition({"or": [self.to_dict(), other.to_dict()]}) class ParameterExpression(_expr_core.OperatorMixin, object): def __init__(self, expr) -> None: self.expr = expr def to_dict(self) -> TypingDict[str, str]: return {"expr": repr(self.expr)} def _to_expr(self) -> str: return repr(self.expr) def _from_expr(self, expr) -> "ParameterExpression": return ParameterExpression(expr=expr) class SelectionExpression(_expr_core.OperatorMixin, object): def __init__(self, expr) -> None: self.expr = expr def to_dict(self) -> TypingDict[str, str]: return {"expr": repr(self.expr)} def _to_expr(self) -> str: return repr(self.expr) def _from_expr(self, expr) -> "SelectionExpression": return SelectionExpression(expr=expr) def check_fields_and_encodings(parameter: Parameter, field_name: str) -> bool: for prop in ["fields", "encodings"]: try: if field_name in getattr(parameter.param.select, prop): # type: ignore[union-attr] return True except (AttributeError, TypeError): pass return False # ------------------------------------------------------------------------ # Top-Level Functions def value(value, **kwargs) -> dict: """Specify a value for use in an encoding""" return dict(value=value, **kwargs) def param( name: Optional[str] = None, value: Union[Any, UndefinedType] = Undefined, bind: Union[core.Binding, UndefinedType] = Undefined, empty: Union[bool, UndefinedType] = Undefined, expr: Union[str, core.Expr, _expr_core.Expression, UndefinedType] = Undefined, **kwds, ) -> 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. """ parameter = Parameter(name) if empty is not Undefined: parameter.empty = empty if parameter.empty == "none": warnings.warn( """The value of 'empty' should be True or False.""", utils.AltairDeprecationWarning, stacklevel=1, ) parameter.empty = False elif parameter.empty == "all": warnings.warn( """The value of 'empty' should be True or False.""", utils.AltairDeprecationWarning, stacklevel=1, ) parameter.empty = True elif (parameter.empty is False) or (parameter.empty is True): pass else: raise ValueError("The value of 'empty' should be True or False.") if "init" in kwds: warnings.warn( """Use 'value' instead of 'init'.""", utils.AltairDeprecationWarning, stacklevel=1, ) if value is Undefined: kwds["value"] = kwds.pop("init") else: # If both 'value' and 'init' are set, we ignore 'init'. kwds.pop("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: Union[Literal["interval", "point"], UndefinedType] = Undefined, **kwds ) -> Parameter: # We separate out the parameter keywords from the selection keywords param_kwds = {} for kwd in {"name", "bind", "value", "empty", "init", "views"}: if kwd in kwds: param_kwds[kwd] = kwds.pop(kwd) select: Union[core.IntervalSelectionConfig, core.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) warnings.warn( """The types 'single' and 'multi' are now combined and should be specified using "selection_point()".""", utils.AltairDeprecationWarning, stacklevel=1, ) else: raise ValueError("""'type' must be 'point' or 'interval'""") return param(select=select, **param_kwds) @utils.deprecation.deprecated( message="""'selection' is deprecated. Use 'selection_point()' or 'selection_interval()' instead; these functions also include more helpful docstrings.""" ) def selection( type: Union[Literal["interval", "point"], UndefinedType] = Undefined, **kwds ) -> Parameter: """ Users are recommended to use either 'selection_point' or 'selection_interval' instead, depending on the type of parameter they want to create. Create a selection parameter. Parameters ---------- type : enum('point', 'interval') (required) Determines the default event processing and data query for the selection. Vega-Lite currently supports two selection types: * "point" - to select multiple discrete data values; the first value is selected on click and additional values toggled on shift-click. * "interval" - to select a continuous range of data values on drag. **kwds : additional keywords to control the selection. """ return _selection(type=type, **kwds) def selection_interval( name: Optional[str] = None, value: Union[Any, UndefinedType] = Undefined, bind: Union[core.Binding, str, UndefinedType] = Undefined, empty: Union[bool, UndefinedType] = Undefined, expr: Union[str, core.Expr, _expr_core.Expression, UndefinedType] = Undefined, encodings: Union[List[str], UndefinedType] = Undefined, on: Union[str, UndefinedType] = Undefined, clear: Union[str, bool, UndefinedType] = Undefined, resolve: Union[Literal["global", "union", "intersect"], UndefinedType] = Undefined, mark: Union[core.Mark, UndefinedType] = Undefined, translate: Union[str, bool, UndefinedType] = Undefined, zoom: Union[str, bool, UndefinedType] = Undefined, **kwds, ) -> 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: Optional[str] = None, value: Union[Any, UndefinedType] = Undefined, bind: Union[core.Binding, str, UndefinedType] = Undefined, empty: Union[bool, UndefinedType] = Undefined, expr: Union[core.Expr, UndefinedType] = Undefined, encodings: Union[List[str], UndefinedType] = Undefined, fields: Union[List[str], UndefinedType] = Undefined, on: Union[str, UndefinedType] = Undefined, clear: Union[str, bool, UndefinedType] = Undefined, resolve: Union[Literal["global", "union", "intersect"], UndefinedType] = Undefined, toggle: Union[str, bool, UndefinedType] = Undefined, nearest: Union[bool, UndefinedType] = Undefined, **kwds, ) -> 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.deprecation.deprecated( message="'selection_multi' is deprecated. Use 'selection_point'" ) @utils.use_signature(core.PointSelectionConfig) def selection_multi(**kwargs): """'selection_multi' is deprecated. Use 'selection_point'""" return _selection(type="point", **kwargs) @utils.deprecation.deprecated( message="'selection_single' is deprecated. Use 'selection_point'" ) @utils.use_signature(core.PointSelectionConfig) def selection_single(**kwargs): """'selection_single' is deprecated. Use 'selection_point'""" return _selection(type="point", **kwargs) @utils.use_signature(core.Binding) def binding(input, **kwargs): """A generic binding""" return core.Binding(input=input, **kwargs) @utils.use_signature(core.BindCheckbox) def binding_checkbox(**kwargs): """A checkbox binding""" return core.BindCheckbox(input="checkbox", **kwargs) @utils.use_signature(core.BindRadioSelect) def binding_radio(**kwargs): """A radio button binding""" return core.BindRadioSelect(input="radio", **kwargs) @utils.use_signature(core.BindRadioSelect) def binding_select(**kwargs): """A select binding""" return core.BindRadioSelect(input="select", **kwargs) @utils.use_signature(core.BindRange) def binding_range(**kwargs): """A range binding""" return core.BindRange(input="range", **kwargs) # TODO: update the docstring def condition( predicate: Union[ Parameter, str, expr.Expression, core.Expr, core.PredicateComposition, dict ], # Types of these depends on where the condition is used so we probably # can't be more specific here. if_true: Any, if_false: Any, **kwargs, ) -> Union[dict, core.SchemaBase]: """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 **kwargs: additional keyword args are added to the resulting dict Returns ------- spec: dict or VegaLiteSchema the spec that describes the condition """ test_predicates = (str, expr.Expression, core.PredicateComposition) condition: TypingDict[ str, Union[ bool, str, _expr_core.Expression, core.PredicateComposition, UndefinedType ], ] if isinstance(predicate, Parameter): if ( predicate.param_type == "selection" or getattr(predicate.param, "expr", Undefined) is Undefined ): condition = {"param": predicate.name} if "empty" in kwargs: condition["empty"] = kwargs.pop("empty") elif isinstance(predicate.empty, bool): condition["empty"] = predicate.empty else: condition = {"test": getattr(predicate.param, "expr", Undefined)} elif isinstance(predicate, test_predicates): condition = {"test": predicate} elif isinstance(predicate, dict): condition = predicate else: raise NotImplementedError( "condition predicate of type {}" "".format(type(predicate)) ) if isinstance(if_true, core.SchemaBase): # convert to dict for now; the from_dict call below will wrap this # dict in the appropriate schema if_true = if_true.to_dict() elif isinstance(if_true, str): if isinstance(if_false, str): raise ValueError( "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." ) else: if_true = utils.parse_shorthand(if_true) if_true.update(kwargs) condition.update(if_true) selection: Union[dict, core.SchemaBase] 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): selection = {"condition": condition, "shorthand": if_false} selection.update(kwargs) else: selection = dict(condition=condition, **if_false) return selection # -------------------------------------------------------------------- # Top-level objects class TopLevelMixin(mixins.ConfigMethodMixin): """Mixin for top-level chart objects such as Chart, LayeredChart, etc.""" _class_is_valid_at_instantiation: bool = False def to_dict( self, validate: bool = True, *, format: str = "vega-lite", ignore: Optional[List[str]] = None, context: Optional[TypingDict[str, Any]] = None, ) -> dict: """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"): raise ValueError( f'The format argument must be either "vega-lite" or "vega". Received {repr(format)}' ) # 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) # TopLevelMixin instance does not necessarily have copy defined but due to how # Altair is set up this should hold. Too complex to type hint right now copy = self.copy(deep=False) # type: ignore[attr-defined] original_data = getattr(copy, "data", Undefined) copy.data = _prepare_data(original_data, context) if original_data is not Undefined: context["data"] = original_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 = 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 the_theme = themes.get() # Use assert to tell type checkers that it is not None. Holds true # as there is always a default theme set when importing Altair assert the_theme is not None vegalite_spec = utils.update_nested(the_theme(), vegalite_spec, copy=True) # 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": raise ValueError( '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")' ) else: return _compile_with_vegafusion(vegalite_spec) else: if format == "vega": plugin = vegalite_compilers.get() if plugin is None: raise ValueError("No active vega-lite compiler plugin found") return plugin(vegalite_spec) else: return vegalite_spec def to_json( self, validate: bool = True, indent: Optional[Union[int, str]] = 2, sort_keys: bool = True, *, format: str = "vega-lite", ignore: Optional[List[str]] = None, context: Optional[TypingDict[str, Any]] = None, **kwargs, ) -> 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: Optional[dict] = None, json_kwds: Optional[dict] = None, fullhtml: bool = True, requirejs: bool = False, inline: bool = False, **kwargs, ) -> 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