Word Clouds

Static Word Clouds

WordCloud pydantic-model

Bases: BaseModel

A Pydantic model for WordCloud options.

Config:

• arbitrary_types_allowed: True
• json_schema_extra: DocJSONSchema.model_json_schema()

Fields:

• data (single_doc_types | multi_doc_types | DataFrame)
• docs (Optional[int | str | list[int] | list[str]])
• limit (Optional[int])
• title (Optional[str])
• height (int)
• width (int)
• opts (dict[str, Any])
• figure_opts (dict[str, Any])
• round (Optional[int])
• counts (dict[str, int])
• cloud (WordCloud | None)
• fig (Optional[Figure])

Source code in lexos/visualization/cloud.py

class WordCloud(BaseModel):
    """A Pydantic model for WordCloud options."""

    data: single_doc_types | multi_doc_types | pd.DataFrame = Field(
        ...,
        description="The data to generate the word cloud from. Accepts data from a string, list of lists or tuples, a dict with terms as keys and counts/frequencies as values, or a dataframe.",
    )
    docs: Optional[int | str | list[int] | list[str]] = Field(
        None, description="A list of documents to be selected from the DTM."
    )
    limit: Optional[int] = Field(
        None, description="The maximum number of terms to plot."
    )
    title: Optional[str] = Field(None, description="The title of the plot.")
    height: int = Field(
        200, gt=50, description="The height of the word cloud in pixels."
    )
    width: int = Field(200, gt=50, description="The width of the word cloud in pixels.")
    opts: dict[str, Any] = Field(
        default_factory=lambda: {
            "background_color": "white",
            "max_words": 2000,
            "contour_width": 0,
            "contour_color": "steelblue",
        },
        description="The WordCloud() options.",
    )
    figure_opts: dict[str, Any] = Field(
        default_factory=dict,
        description="A dict of matplotlib figure options.",
    )
    round: Optional[int] = Field(
        0,
        description="An integer to apply a mask that rounds the word cloud. It is best to use 100 or higher for a circular mask, but it will depend on the height and width of the word cloud.",
    )
    counts: dict[str, int] = Field(None, description="A dictionary of term counts.")
    cloud: PythonWordCloud | None = Field(
        None, description="The generated WordCloud object."
    )
    fig: Optional[plt.Figure] = Field(
        None, description="The matplotlib figure object for the word cloud."
    )

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_schema_extra=DocJSONSchema.model_json_schema(),
    )

    def __init__(self, **data: Any) -> None:
        """Initialize the WordCloud model."""
        super().__init__(**data)

        # Set the figure dimensions
        self.opts["height"] = self.height
        self.opts["width"] = self.width

        # Set the mask, if using
        if self.round > 0:
            x, y = np.ogrid[:300, :300]
            mask = (x - 150) ** 2 + (y - 150) ** 2 > self.round**2
            mask = 255 * mask.astype(int)
            self.opts["mask"] = mask

        # Process the data into a consistent format
        self.counts = processors.process_data(self.data, self.docs, self.limit)

        # Generate the word cloud
        self.cloud = PythonWordCloud(**self.opts).generate_from_frequencies(self.counts)

    @validate_call
    def save(self, path: Path | str, **kwargs: Any) -> None:
        """Save the WordCloud to a file.

        Args:
            path (Path | str): The file path to save the WordCloud image.
            **kwargs (Any): Additional keyword arguments for `plt.savefig`.
        """
        if self.cloud is None:
            raise LexosException("No WordCloud object to save.")
        self.fig = plt.figure(**self.figure_opts)
        ax = self.fig.add_subplot(111)
        if self.title:
            self.fig.suptitle(self.title)
        ax.axis("off")
        ax.imshow(self.cloud.to_array(), interpolation="bilinear")
        self.fig.savefig(path, **kwargs)
        plt.close(self.fig)

    def show(self) -> None:
        """Show the figure if it is hidden.

        This is a helper method. It will generally display in a
        Jupyter notebook.
        """
        self.fig = plt.figure(**self.figure_opts)
        if self.title:
            self.fig.suptitle(self.title)
        plt.axis("off")
        plt.imshow(self.cloud.to_array(), interpolation="bilinear")

data: single_doc_types | multi_doc_types | pd.DataFrame pydantic-field

The data to generate the word cloud from. Accepts data from a string, list of lists or tuples, a dict with terms as keys and counts/frequencies as values, or a dataframe.

docs: Optional[int | str | list[int] | list[str]] = None pydantic-field

A list of documents to be selected from the DTM.

fig: Optional[plt.Figure] = None pydantic-field

The matplotlib figure object for the word cloud.

figure_opts: dict[str, Any] pydantic-field

A dict of matplotlib figure options.

height: int = 200 pydantic-field

The height of the word cloud in pixels.

limit: Optional[int] = None pydantic-field

The maximum number of terms to plot.

opts: dict[str, Any] pydantic-field

The WordCloud() options.

round: Optional[int] = 0 pydantic-field

An integer to apply a mask that rounds the word cloud. It is best to use 100 or higher for a circular mask, but it will depend on the height and width of the word cloud.

title: Optional[str] = None pydantic-field

The title of the plot.

width: int = 200 pydantic-field

The width of the word cloud in pixels.

__init__(**data: Any) -> None

Initialize the WordCloud model.

Source code in lexos/visualization/cloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the WordCloud model."""
    super().__init__(**data)

    # Set the figure dimensions
    self.opts["height"] = self.height
    self.opts["width"] = self.width

    # Set the mask, if using
    if self.round > 0:
        x, y = np.ogrid[:300, :300]
        mask = (x - 150) ** 2 + (y - 150) ** 2 > self.round**2
        mask = 255 * mask.astype(int)
        self.opts["mask"] = mask

    # Process the data into a consistent format
    self.counts = processors.process_data(self.data, self.docs, self.limit)

    # Generate the word cloud
    self.cloud = PythonWordCloud(**self.opts).generate_from_frequencies(self.counts)

save(path: Path | str, **kwargs: Any) -> None

Save the WordCloud to a file.

Parameters:

Name	Type	Description	Default
path	Path | str	The file path to save the WordCloud image.	required
**kwargs	Any	Additional keyword arguments for plt.savefig.	{}

Source code in lexos/visualization/cloud.py

@validate_call
def save(self, path: Path | str, **kwargs: Any) -> None:
    """Save the WordCloud to a file.

    Args:
        path (Path | str): The file path to save the WordCloud image.
        **kwargs (Any): Additional keyword arguments for `plt.savefig`.
    """
    if self.cloud is None:
        raise LexosException("No WordCloud object to save.")
    self.fig = plt.figure(**self.figure_opts)
    ax = self.fig.add_subplot(111)
    if self.title:
        self.fig.suptitle(self.title)
    ax.axis("off")
    ax.imshow(self.cloud.to_array(), interpolation="bilinear")
    self.fig.savefig(path, **kwargs)
    plt.close(self.fig)

show() -> None

Show the figure if it is hidden.

This is a helper method. It will generally display in a Jupyter notebook.

Source code in lexos/visualization/cloud.py

def show(self) -> None:
    """Show the figure if it is hidden.

    This is a helper method. It will generally display in a
    Jupyter notebook.
    """
    self.fig = plt.figure(**self.figure_opts)
    if self.title:
        self.fig.suptitle(self.title)
    plt.axis("off")
    plt.imshow(self.cloud.to_array(), interpolation="bilinear")

data: single_doc_types | multi_doc_types | pd.DataFrame pydantic-field

The data to generate the word cloud from. Accepts data from a string, list of lists or tuples, a dict with terms as keys and counts/frequencies as values, or a dataframe.

docs: Optional[int | str | list[int] | list[str]] = None pydantic-field

A list of documents to be selected from the DTM.

limit: Optional[int] = None pydantic-field

The maximum number of terms to plot.

title: Optional[str] = None pydantic-field

The title of the plot.

height: int = 200 pydantic-field

The height of the word cloud in pixels.

width: int = 200 pydantic-field

The width of the word cloud in pixels.

opts: dict[str, Any] pydantic-field

The WordCloud() options.

figure_opts: dict[str, Any] pydantic-field

A dict of matplotlib figure options.

round: Optional[int] = 0 pydantic-field

An integer to apply a mask that rounds the word cloud. It is best to use 100 or higher for a circular mask, but it will depend on the height and width of the word cloud.

counts: dict[str, int] pydantic-field

cloud: PythonWordCloud | None pydantic-field

fig: Optional[plt.Figure] = None pydantic-field

The matplotlib figure object for the word cloud.

model_config = ConfigDict(arbitrary_types_allowed=True, json_schema_extra=(DocJSONSchema.model_json_schema())) class-attribute instance-attribute

__init__(**data: Any) -> None

Initialize the WordCloud model.

Source code in lexos/visualization/cloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the WordCloud model."""
    super().__init__(**data)

    # Set the figure dimensions
    self.opts["height"] = self.height
    self.opts["width"] = self.width

    # Set the mask, if using
    if self.round > 0:
        x, y = np.ogrid[:300, :300]
        mask = (x - 150) ** 2 + (y - 150) ** 2 > self.round**2
        mask = 255 * mask.astype(int)
        self.opts["mask"] = mask

    # Process the data into a consistent format
    self.counts = processors.process_data(self.data, self.docs, self.limit)

    # Generate the word cloud
    self.cloud = PythonWordCloud(**self.opts).generate_from_frequencies(self.counts)

save(path: Path | str, **kwargs: Any) -> None

Save the WordCloud to a file.

Parameters:

Name	Type	Description	Default
path	Path | str	The file path to save the WordCloud image.	required
**kwargs	Any	Additional keyword arguments for plt.savefig.	{}

Source code in lexos/visualization/cloud.py

@validate_call
def save(self, path: Path | str, **kwargs: Any) -> None:
    """Save the WordCloud to a file.

    Args:
        path (Path | str): The file path to save the WordCloud image.
        **kwargs (Any): Additional keyword arguments for `plt.savefig`.
    """
    if self.cloud is None:
        raise LexosException("No WordCloud object to save.")
    self.fig = plt.figure(**self.figure_opts)
    ax = self.fig.add_subplot(111)
    if self.title:
        self.fig.suptitle(self.title)
    ax.axis("off")
    ax.imshow(self.cloud.to_array(), interpolation="bilinear")
    self.fig.savefig(path, **kwargs)
    plt.close(self.fig)

show() -> None

Show the figure if it is hidden.

This is a helper method. It will generally display in a Jupyter notebook.

Source code in lexos/visualization/cloud.py

def show(self) -> None:
    """Show the figure if it is hidden.

    This is a helper method. It will generally display in a
    Jupyter notebook.
    """
    self.fig = plt.figure(**self.figure_opts)
    if self.title:
        self.fig.suptitle(self.title)
    plt.axis("off")
    plt.imshow(self.cloud.to_array(), interpolation="bilinear")

MultiCloud pydantic-model

Bases: BaseModel

A Pydantic model for creating multiple WordClouds arranged in a grid using the topic_clouds approach.

Config:

• arbitrary_types_allowed: True
• json_schema_extra: DocJSONSchema.model_json_schema()

Fields:

• data (list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | DataFrame)
• docs (Optional[int | str | list[int] | list[str]])
• limit (Optional[int])
• figsize (tuple[int, int])
• layout (Optional[str | tuple[int, int]])
• opts (dict[str, Any])
• round (Optional[int])
• title (Optional[str])
• labels (Optional[list[str]])
• doc_data (Optional[list[dict[str, int | float]]])
• fig (Optional[Figure])
• wordcloud (Optional[WordCloud])

Source code in lexos/visualization/cloud.py

class MultiCloud(BaseModel):
    """A Pydantic model for creating multiple WordClouds arranged in a grid using the topic_clouds approach."""

    data: list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | pd.DataFrame = (
        Field(
            ...,
            description="The data to generate word clouds from. Accepts list of documents, DTM, or DataFrame.",
        )
    )
    docs: Optional[int | str | list[int] | list[str]] = Field(
        None, description="A list of documents to be selected from the DTM/DataFrame."
    )
    limit: Optional[int] = Field(
        None, description="The maximum number of terms to plot per cloud."
    )
    figsize: tuple[int, int] = Field(
        (10, 10), description="The size of the overall figure."
    )
    layout: Optional[str | tuple[int, int]] = Field(
        "auto",
        description="The number of rows and columns in the figure. Default is 'auto'.",
    )
    opts: dict[str, Any] = Field(
        default_factory=lambda: {
            "background_color": "white",
            "max_words": 2000,
            "contour_width": 0,
            "contour_color": "steelblue",
        },
        description="The WordCloud() options applied to each word cloud.",
    )
    round: Optional[int] = Field(
        0,
        description="An integer to apply a mask that rounds each word cloud. It is best to use 100 or higher for a circular mask.",
    )
    title: Optional[str] = Field(None, description="Overall title for the figure.")
    labels: Optional[list[str]] = Field(
        None, description="Labels for each subplot/word cloud."
    )
    doc_data: Optional[list[dict[str, int | float]]] = Field(
        None, description="Processed document data for each word cloud."
    )
    fig: Optional[plt.Figure] = Field(
        None, description="The matplotlib figure object for the multi-cloud plot."
    )
    wordcloud: Optional[PythonWordCloud] = Field(
        None, description="The WordCloud object used for generating clouds."
    )

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_schema_extra=DocJSONSchema.model_json_schema(),
    )

    def __init__(self, **data: Any) -> None:
        """Initialize the MultiCloud model."""
        super().__init__(**data)

        # Process different data types to get individual document data
        self.doc_data = self._process_data()

        # Setup the WordCloud object
        self.wordcloud = self._setup_wordcloud()

        # Render the figure
        self._render()

    def _process_data(self) -> list[dict[str, int | float]]:
        """Process the input data into individual document dictionaries."""
        if isinstance(self.data, DTM):
            # Make sure there is data
            if (
                self.data.doc_term_matrix is None
                or self.data.doc_term_matrix.shape[0] == 0
            ):
                raise LexosException("Empty DTM provided.")
            # Extract documents from DTM
            doc_data = []
            selected_docs = (
                self.docs
                if self.docs is not None
                else range(self.data.doc_term_matrix.shape[0])
            )
            if isinstance(selected_docs, (int, str)):
                selected_docs = [selected_docs]

            for doc_idx in selected_docs:
                # Get term frequencies for this document
                if isinstance(doc_idx, str):
                    doc_idx = self.data.labels.index(doc_idx)
                doc_counts = {}

                # Get the row as a 1D array and convert to list/scalar values
                doc_row = self.data.doc_term_matrix[doc_idx]
                if hasattr(doc_row, "toarray"):  # Sparse matrix
                    doc_row = doc_row.toarray().flatten()

                for term_idx, count in enumerate(doc_row):
                    # Convert to scalar value before comparison
                    count_value = (
                        float(count.item()) if hasattr(count, "item") else float(count)
                    )
                    if count_value > 0:
                        doc_counts[self.data.vectorizer.terms_list[term_idx]] = (
                            count_value
                        )
                doc_data.append(doc_counts)

        elif isinstance(self.data, pd.DataFrame):
            # Make sure there is data
            if self.data.empty:
                raise LexosException("Empty DataFrame provided.")
            # Process DataFrame - assume it's a document-term matrix
            doc_data = []
            selected_docs = (
                self.docs if self.docs is not None else range(len(self.data))
            )
            if isinstance(selected_docs, (int, str)):
                selected_docs = [selected_docs]

            for doc_idx in selected_docs:
                if isinstance(doc_idx, str):
                    doc_idx = self.data.index.get_loc(doc_idx)
                doc_counts = self.data.iloc[doc_idx].to_dict()
                # Filter out zero counts and convert to float
                doc_counts = {
                    k: float(v.item() if hasattr(v, "item") else v)
                    for k, v in doc_counts.items()
                    if (float(v.item()) if hasattr(v, "item") else float(v)) > 0
                }
                doc_data.append(doc_counts)

        elif isinstance(self.data, list):
            # Make sure the data is not empty
            if not self.data or len(self.data) == 0:
                raise LexosException("No valid data provided for MultiCloud.")
            # Process list of documents using the processors module
            doc_data = [
                processors.process_data(doc, None, self.limit) for doc in self.data
            ]

        else:
            raise LexosException("Unsupported data type for MultiCloud.")

        return doc_data

    def _setup_wordcloud(self) -> PythonWordCloud:
        """Configure a single WordCloud object to be reused."""
        # Set the mask if using round
        if self.round > 0:
            x, y = np.ogrid[:300, :300]
            mask = (x - 150) ** 2 + (y - 150) ** 2 > self.round**2
            mask = 255 * mask.astype(int)
            self.opts["mask"] = mask

        # Set max_words if limit is specified
        if self.limit:
            self.opts["max_words"] = self.limit

        return PythonWordCloud(**self.opts)

    def _render(self) -> None:
        """Generate and display the multi-cloud figure."""
        # Create a local figure without mutating global matplotlib style
        self.fig = plt.figure(figsize=self.figsize)

        # Calculate layout
        n = len(self.doc_data)
        if self.layout == "auto":
            columns = math.floor(math.sqrt(n))
            rows = math.ceil(n / columns)
        elif isinstance(self.layout, tuple):
            rows, columns = self.layout
        else:
            raise LexosException("Invalid layout specification.")

        # Add overall title
        if self.title:
            self.fig.suptitle(self.title, fontsize=16)

        # Generate the word clouds
        for i, doc_counts in enumerate(self.doc_data):
            self.wordcloud.generate_from_frequencies(doc_counts)
            ax = self.fig.add_subplot(rows, columns, i + 1)
            ax.imshow(self.wordcloud.to_array(), interpolation="bilinear")
            ax.axis("off")

            # Add label if provided
            if self.labels and i < len(self.labels):
                ax.set_title(self.labels[i])
            else:
                ax.set_title(f"Doc {i}")

        # Get the figure and close to prevent automatic display
        self.fig = plt.gcf()
        plt.close()

    @validate_call
    def save(self, path: Path | str, **kwargs: Any) -> None:
        """Save the MultiCloud figure to a file.

        Args:
            path (Path | str): The file path to save the MultiCloud image.
            **kwargs (Any): Additional keyword arguments for `plt.savefig`.
        """
        if self.fig is None:
            raise LexosException("No figure to save.")
        self.fig.savefig(path, **kwargs)

    def show(self) -> None:
        """Display the multi-cloud figure."""
        if self.fig is None:
            raise LexosException("No figure to show.")
        # Use IPython display for Jupyter notebooks
        try:
            from IPython.display import display

            display(self.fig)
        except ImportError:
            # Fallback for non-Jupyter environments
            plt.figure(self.fig.number)
            plt.show()

data: list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | pd.DataFrame pydantic-field

The data to generate word clouds from. Accepts list of documents, DTM, or DataFrame.

docs: Optional[int | str | list[int] | list[str]] = None pydantic-field

A list of documents to be selected from the DTM/DataFrame.

fig: Optional[plt.Figure] = None pydantic-field

The matplotlib figure object for the multi-cloud plot.

figsize: tuple[int, int] = (10, 10) pydantic-field

The size of the overall figure.

labels: Optional[list[str]] = None pydantic-field

Labels for each subplot/word cloud.

layout: Optional[str | tuple[int, int]] = 'auto' pydantic-field

The number of rows and columns in the figure. Default is 'auto'.

limit: Optional[int] = None pydantic-field

The maximum number of terms to plot per cloud.

opts: dict[str, Any] pydantic-field

The WordCloud() options applied to each word cloud.

round: Optional[int] = 0 pydantic-field

An integer to apply a mask that rounds each word cloud. It is best to use 100 or higher for a circular mask.

title: Optional[str] = None pydantic-field

Overall title for the figure.

__init__(**data: Any) -> None

Initialize the MultiCloud model.

Source code in lexos/visualization/cloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the MultiCloud model."""
    super().__init__(**data)

    # Process different data types to get individual document data
    self.doc_data = self._process_data()

    # Setup the WordCloud object
    self.wordcloud = self._setup_wordcloud()

    # Render the figure
    self._render()

save(path: Path | str, **kwargs: Any) -> None

Save the MultiCloud figure to a file.

Parameters:

Name	Type	Description	Default
path	Path | str	The file path to save the MultiCloud image.	required
**kwargs	Any	Additional keyword arguments for plt.savefig.	{}

Source code in lexos/visualization/cloud.py

@validate_call
def save(self, path: Path | str, **kwargs: Any) -> None:
    """Save the MultiCloud figure to a file.

    Args:
        path (Path | str): The file path to save the MultiCloud image.
        **kwargs (Any): Additional keyword arguments for `plt.savefig`.
    """
    if self.fig is None:
        raise LexosException("No figure to save.")
    self.fig.savefig(path, **kwargs)

show() -> None

Display the multi-cloud figure.

Source code in lexos/visualization/cloud.py

def show(self) -> None:
    """Display the multi-cloud figure."""
    if self.fig is None:
        raise LexosException("No figure to show.")
    # Use IPython display for Jupyter notebooks
    try:
        from IPython.display import display

        display(self.fig)
    except ImportError:
        # Fallback for non-Jupyter environments
        plt.figure(self.fig.number)
        plt.show()

data: list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | pd.DataFrame pydantic-field

The data to generate word clouds from. Accepts list of documents, DTM, or DataFrame.

docs: Optional[int | str | list[int] | list[str]] = None pydantic-field

A list of documents to be selected from the DTM/DataFrame.

limit: Optional[int] = None pydantic-field

The maximum number of terms to plot per cloud.

figsize: tuple[int, int] = (10, 10) pydantic-field

The size of the overall figure.

layout: Optional[str | tuple[int, int]] = 'auto' pydantic-field

The number of rows and columns in the figure. Default is 'auto'.

opts: dict[str, Any] pydantic-field

The WordCloud() options applied to each word cloud.

round: Optional[int] = 0 pydantic-field

An integer to apply a mask that rounds each word cloud. It is best to use 100 or higher for a circular mask.

title: Optional[str] = None pydantic-field

Overall title for the figure.

labels: Optional[list[str]] = None pydantic-field

Labels for each subplot/word cloud.

doc_data: Optional[list[dict[str, int | float]]] pydantic-field

fig: Optional[plt.Figure] = None pydantic-field

The matplotlib figure object for the multi-cloud plot.

wordcloud: Optional[PythonWordCloud] pydantic-field

model_config = ConfigDict(arbitrary_types_allowed=True, json_schema_extra=(DocJSONSchema.model_json_schema())) class-attribute instance-attribute

__init__(**data: Any) -> None

Initialize the MultiCloud model.

Source code in lexos/visualization/cloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the MultiCloud model."""
    super().__init__(**data)

    # Process different data types to get individual document data
    self.doc_data = self._process_data()

    # Setup the WordCloud object
    self.wordcloud = self._setup_wordcloud()

    # Render the figure
    self._render()

_process_data() -> list[dict[str, int | float]]

Process the input data into individual document dictionaries.

Source code in lexos/visualization/cloud.py

def _process_data(self) -> list[dict[str, int | float]]:
    """Process the input data into individual document dictionaries."""
    if isinstance(self.data, DTM):
        # Make sure there is data
        if (
            self.data.doc_term_matrix is None
            or self.data.doc_term_matrix.shape[0] == 0
        ):
            raise LexosException("Empty DTM provided.")
        # Extract documents from DTM
        doc_data = []
        selected_docs = (
            self.docs
            if self.docs is not None
            else range(self.data.doc_term_matrix.shape[0])
        )
        if isinstance(selected_docs, (int, str)):
            selected_docs = [selected_docs]

        for doc_idx in selected_docs:
            # Get term frequencies for this document
            if isinstance(doc_idx, str):
                doc_idx = self.data.labels.index(doc_idx)
            doc_counts = {}

            # Get the row as a 1D array and convert to list/scalar values
            doc_row = self.data.doc_term_matrix[doc_idx]
            if hasattr(doc_row, "toarray"):  # Sparse matrix
                doc_row = doc_row.toarray().flatten()

            for term_idx, count in enumerate(doc_row):
                # Convert to scalar value before comparison
                count_value = (
                    float(count.item()) if hasattr(count, "item") else float(count)
                )
                if count_value > 0:
                    doc_counts[self.data.vectorizer.terms_list[term_idx]] = (
                        count_value
                    )
            doc_data.append(doc_counts)

    elif isinstance(self.data, pd.DataFrame):
        # Make sure there is data
        if self.data.empty:
            raise LexosException("Empty DataFrame provided.")
        # Process DataFrame - assume it's a document-term matrix
        doc_data = []
        selected_docs = (
            self.docs if self.docs is not None else range(len(self.data))
        )
        if isinstance(selected_docs, (int, str)):
            selected_docs = [selected_docs]

        for doc_idx in selected_docs:
            if isinstance(doc_idx, str):
                doc_idx = self.data.index.get_loc(doc_idx)
            doc_counts = self.data.iloc[doc_idx].to_dict()
            # Filter out zero counts and convert to float
            doc_counts = {
                k: float(v.item() if hasattr(v, "item") else v)
                for k, v in doc_counts.items()
                if (float(v.item()) if hasattr(v, "item") else float(v)) > 0
            }
            doc_data.append(doc_counts)

    elif isinstance(self.data, list):
        # Make sure the data is not empty
        if not self.data or len(self.data) == 0:
            raise LexosException("No valid data provided for MultiCloud.")
        # Process list of documents using the processors module
        doc_data = [
            processors.process_data(doc, None, self.limit) for doc in self.data
        ]

    else:
        raise LexosException("Unsupported data type for MultiCloud.")

    return doc_data

_setup_wordcloud() -> PythonWordCloud

Configure a single WordCloud object to be reused.

Source code in lexos/visualization/cloud.py

def _setup_wordcloud(self) -> PythonWordCloud:
    """Configure a single WordCloud object to be reused."""
    # Set the mask if using round
    if self.round > 0:
        x, y = np.ogrid[:300, :300]
        mask = (x - 150) ** 2 + (y - 150) ** 2 > self.round**2
        mask = 255 * mask.astype(int)
        self.opts["mask"] = mask

    # Set max_words if limit is specified
    if self.limit:
        self.opts["max_words"] = self.limit

    return PythonWordCloud(**self.opts)

MultiCloudOld pydantic-model

Bases: BaseModel

A Pydantic model for creating multiple WordClouds arranged in a grid.

NOTE: This Class is deprecated.

Config:

• arbitrary_types_allowed: True
• json_schema_extra: DocJSONSchema.model_json_schema()

Fields:

• data (list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | DataFrame)
• docs (Optional[int | str | list[int] | list[str]])
• limit (Optional[int])
• ncols (int)
• height (int)
• width (int)
• opts (Optional[dict[str, Any]])
• figure_opts (Optional[dict[str, Any]])
• round (Optional[int])
• title (Optional[str])
• labels (Optional[list[str]])
• padding (float)
• clouds (list[WordCloud])
• fig (Optional[Figure])

Source code in lexos/visualization/cloud.py

class MultiCloudOld(BaseModel):
    """A Pydantic model for creating multiple WordClouds arranged in a grid.

    # NOTE: This Class is deprecated.
    """

    data: list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | pd.DataFrame = (
        Field(
            ...,
            description="The data to generate word clouds from. Accepts list of documents, DTM, or DataFrame.",
        )
    )
    docs: Optional[int | str | list[int] | list[str]] = Field(
        None, description="A list of documents to be selected from the DTM/DataFrame."
    )
    limit: Optional[int] = Field(
        None, description="The maximum number of terms to plot."
    )
    ncols: int = Field(3, gt=0, description="Number of columns in the grid layout.")
    height: int = Field(
        200, gt=50, description="The height of each word cloud in pixels."
    )
    width: int = Field(
        200, gt=50, description="The width of each word cloud in pixels."
    )
    opts: Optional[dict[str, Any]] = Field(
        {
            "background_color": "white",
            "max_words": 2000,
            "contour_width": 0,
            "contour_color": "steelblue",
        },
        description="The WordCloud() options applied to each word cloud.",
    )
    figure_opts: Optional[dict[str, Any]] = Field(
        {}, description="A dict of matplotlib figure options."
    )
    round: Optional[int] = Field(
        0,
        description="An integer to apply a mask that rounds each word cloud. It is best to use 100 or higher for a circular mask.",
    )
    title: Optional[str] = Field(None, description="Overall title for the figure.")
    labels: Optional[list[str]] = Field(
        None, description="Labels for each subplot/word cloud."
    )
    padding: float = Field(
        0.3,
        ge=0.0,
        le=1.0,
        description="Amount of padding between subplots (0.0 to 1.0).",
    )
    clouds: list[WordCloud] = Field(
        default_factory=list, description="List of generated WordCloud objects."
    )
    fig: Optional[plt.Figure] = Field(
        None, description="The matplotlib figure object for the multi-cloud plot."
    )

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_schema_extra=DocJSONSchema.model_json_schema(),
    )

    def __init__(self, **data: Any) -> None:
        """Initialize the MultiCloud model."""
        super().__init__(**data)

        # Process different data types to get individual document data
        doc_data = self._process_data()

        # Create individual WordCloud objects
        self.clouds = []
        for doc in doc_data:
            try:
                # Create a WordCloud instance for each document
                wc = WordCloud(
                    data=doc,
                    limit=self.limit,
                    opts=self.opts,
                    round=self.round,
                    width=self.width,
                    height=self.height,
                )
                self.clouds.append(wc)
            except Exception as e:
                raise LexosException(f"Failed to create word cloud: {e}")

        # Render the figure
        self._render()

    def _process_data(self) -> list:
        """Process the input data into individual documents."""
        if isinstance(self.data, DTM):
            # Make sure there is data
            if (
                self.data.doc_term_matrix is None
                or self.data.doc_term_matrix.shape[0] == 0
            ):
                raise LexosException("Empty DTM provided.")
            # Extract documents from DTM
            doc_data = []
            selected_docs = (
                self.docs
                if self.docs is not None
                else range(self.data.doc_term_matrix.shape[0])
            )
            if isinstance(selected_docs, (int, str)):
                selected_docs = [selected_docs]

            for doc_idx in selected_docs:
                # Get term frequencies for this document
                if isinstance(doc_idx, str):
                    doc_idx = self.data.labels.index(doc_idx)
                doc_counts = {}

                # Get the row as a 1D array and convert to list/scalar values
                doc_row = self.data.doc_term_matrix[doc_idx]
                if hasattr(doc_row, "toarray"):  # Sparse matrix
                    doc_row = doc_row.toarray().flatten()

                for term_idx, count in enumerate(doc_row):
                    # Convert to scalar value before comparison
                    count_value = (
                        float(count.item()) if hasattr(count, "item") else float(count)
                    )
                    if count_value > 0:
                        doc_counts[self.data.vectorizer.terms_list[term_idx]] = int(
                            count_value
                        )
                doc_data.append(doc_counts)

        elif isinstance(self.data, pd.DataFrame):
            # Make sure there is data
            if self.data.empty == True:
                raise LexosException("Empty DataFrame provided.")
            # Process DataFrame - assume it's a document-term matrix
            doc_data = []
            selected_docs = (
                self.docs if self.docs is not None else range(len(self.data))
            )
            if isinstance(selected_docs, (int, str)):
                selected_docs = [selected_docs]

            for doc_idx in selected_docs:
                if isinstance(doc_idx, str):
                    doc_idx = self.data.index.get_loc(doc_idx)
                doc_counts = self.data.iloc[doc_idx].to_dict()
                # Filter out zero counts
                doc_counts = {
                    k: v
                    for k, v in doc_counts.items()
                    if (float(v) if hasattr(v, "item") else v) > 0
                }
                doc_data.append(doc_counts)

        elif isinstance(self.data, list):
            # Make sure the data is not empty
            if not self.data or len(self.data) == 0:
                raise LexosException("No valid data provided for MultiCloud.")
            # Handle list of documents
            doc_data = self.data

        return doc_data

    def _render(self) -> None:
        """Generate and display the multi-cloud figure."""
        # Calculate layout
        num_clouds = len(self.clouds)
        nrows = int(np.ceil(num_clouds / self.ncols))

        # Set up figure with padding
        figure_opts = self.figure_opts.copy()
        figure_opts.setdefault("figsize", (self.ncols * 4, nrows * 3))

        # Remove constrained_layout if it exists since we're setting manual spacing
        figure_opts.pop("constrained_layout", None)

        self.fig, axes = plt.subplots(nrows, self.ncols, **figure_opts)

        # Add padding between subplots and adjust top margin for title
        if self.title:
            # More space below title when there's a suptitle
            self.fig.subplots_adjust(
                wspace=self.padding,
                hspace=self.padding,
                top=0.82,  # Leaves more space at the top for the title
            )
        else:
            # Normal spacing when no title
            self.fig.subplots_adjust(wspace=self.padding, hspace=self.padding)

        # Add padding between subplots
        self.fig.subplots_adjust(wspace=self.padding, hspace=self.padding)

        # Handle single row case
        if nrows == 1:
            axes = axes.reshape(1, -1) if self.ncols > 1 else np.array([[axes]])
        elif self.ncols == 1:
            axes = axes.reshape(-1, 1)

        # Add overall title
        if self.title:
            self.fig.suptitle(self.title, fontsize=16, y=0.90)  # Positioned lower

        # Plot each word cloud
        for i, cloud in enumerate(self.clouds):
            row = i // self.ncols
            col = i % self.ncols

            ax = axes[row, col]

            # Display the word cloud
            ax.imshow(cloud.cloud.to_array(), interpolation="bilinear")
            ax.axis("off")

            # Add label if provided
            if self.labels and i < len(self.labels):
                ax.set_title(self.labels[i])
            elif hasattr(cloud.data, "__len__"):
                ax.set_title(f"Doc {i + 1}", fontdict={"fontsize": 10})

        # Hide unused subplots
        for i in range(num_clouds, nrows * self.ncols):
            row = i // self.ncols
            col = i % self.ncols
            axes[row, col].axis("off")
            axes[row, col].set_visible(False)

        # Prevent automatic display
        self.fig = plt.gcf()
        plt.close()

    @validate_call
    def save(self, path: Path | str, **kwargs: Any) -> None:
        """Save the MultiCloud figure to a file.

        Args:
            path (Path | str): The file path to save the MultiCloud image.
            **kwargs (Any): Additional keyword arguments for `plt.savefig`.
        """
        if self.fig is None:
            raise LexosException("No figure to save.")
        self.fig.savefig(path, **kwargs)

    def get_clouds(self) -> list[WordCloud]:
        """Return the list of individual WordCloud objects."""
        return self.clouds

    def show(self) -> plt.Figure:
        """Display the multi-cloud figure."""
        if self.fig is None:
            raise LexosException("No figure to show.")
        return self.fig

data: list[str] | list[list[str]] | list[Doc] | list[Span] | DTM | pd.DataFrame pydantic-field

The data to generate word clouds from. Accepts list of documents, DTM, or DataFrame.

docs: Optional[int | str | list[int] | list[str]] = None pydantic-field

A list of documents to be selected from the DTM/DataFrame.

fig: Optional[plt.Figure] = None pydantic-field

The matplotlib figure object for the multi-cloud plot.

figure_opts: Optional[dict[str, Any]] = {} pydantic-field

A dict of matplotlib figure options.

height: int = 200 pydantic-field

The height of each word cloud in pixels.

labels: Optional[list[str]] = None pydantic-field

Labels for each subplot/word cloud.

limit: Optional[int] = None pydantic-field

The maximum number of terms to plot.

ncols: int = 3 pydantic-field

Number of columns in the grid layout.

opts: Optional[dict[str, Any]] = {'background_color': 'white', 'max_words': 2000, 'contour_width': 0, 'contour_color': 'steelblue'} pydantic-field

The WordCloud() options applied to each word cloud.

padding: float = 0.3 pydantic-field

Amount of padding between subplots (0.0 to 1.0).

round: Optional[int] = 0 pydantic-field

An integer to apply a mask that rounds each word cloud. It is best to use 100 or higher for a circular mask.

title: Optional[str] = None pydantic-field

Overall title for the figure.

width: int = 200 pydantic-field

The width of each word cloud in pixels.

__init__(**data: Any) -> None

Initialize the MultiCloud model.

Source code in lexos/visualization/cloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the MultiCloud model."""
    super().__init__(**data)

    # Process different data types to get individual document data
    doc_data = self._process_data()

    # Create individual WordCloud objects
    self.clouds = []
    for doc in doc_data:
        try:
            # Create a WordCloud instance for each document
            wc = WordCloud(
                data=doc,
                limit=self.limit,
                opts=self.opts,
                round=self.round,
                width=self.width,
                height=self.height,
            )
            self.clouds.append(wc)
        except Exception as e:
            raise LexosException(f"Failed to create word cloud: {e}")

    # Render the figure
    self._render()

get_clouds() -> list[WordCloud]

Return the list of individual WordCloud objects.

Source code in lexos/visualization/cloud.py

def get_clouds(self) -> list[WordCloud]:
    """Return the list of individual WordCloud objects."""
    return self.clouds

save(path: Path | str, **kwargs: Any) -> None

Save the MultiCloud figure to a file.

Parameters:

Name	Type	Description	Default
path	Path | str	The file path to save the MultiCloud image.	required
**kwargs	Any	Additional keyword arguments for plt.savefig.	{}

Source code in lexos/visualization/cloud.py

@validate_call
def save(self, path: Path | str, **kwargs: Any) -> None:
    """Save the MultiCloud figure to a file.

    Args:
        path (Path | str): The file path to save the MultiCloud image.
        **kwargs (Any): Additional keyword arguments for `plt.savefig`.
    """
    if self.fig is None:
        raise LexosException("No figure to save.")
    self.fig.savefig(path, **kwargs)

show() -> plt.Figure

Display the multi-cloud figure.

Source code in lexos/visualization/cloud.py

def show(self) -> plt.Figure:
    """Display the multi-cloud figure."""
    if self.fig is None:
        raise LexosException("No figure to show.")
    return self.fig

__init__(**data: Any) -> None

Initialize the MultiCloud model.

Source code in lexos/visualization/cloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the MultiCloud model."""
    super().__init__(**data)

    # Process different data types to get individual document data
    doc_data = self._process_data()

    # Create individual WordCloud objects
    self.clouds = []
    for doc in doc_data:
        try:
            # Create a WordCloud instance for each document
            wc = WordCloud(
                data=doc,
                limit=self.limit,
                opts=self.opts,
                round=self.round,
                width=self.width,
                height=self.height,
            )
            self.clouds.append(wc)
        except Exception as e:
            raise LexosException(f"Failed to create word cloud: {e}")

    # Render the figure
    self._render()

_process_data() -> list

Process the input data into individual documents.

Source code in lexos/visualization/cloud.py

def _process_data(self) -> list:
    """Process the input data into individual documents."""
    if isinstance(self.data, DTM):
        # Make sure there is data
        if (
            self.data.doc_term_matrix is None
            or self.data.doc_term_matrix.shape[0] == 0
        ):
            raise LexosException("Empty DTM provided.")
        # Extract documents from DTM
        doc_data = []
        selected_docs = (
            self.docs
            if self.docs is not None
            else range(self.data.doc_term_matrix.shape[0])
        )
        if isinstance(selected_docs, (int, str)):
            selected_docs = [selected_docs]

        for doc_idx in selected_docs:
            # Get term frequencies for this document
            if isinstance(doc_idx, str):
                doc_idx = self.data.labels.index(doc_idx)
            doc_counts = {}

            # Get the row as a 1D array and convert to list/scalar values
            doc_row = self.data.doc_term_matrix[doc_idx]
            if hasattr(doc_row, "toarray"):  # Sparse matrix
                doc_row = doc_row.toarray().flatten()

            for term_idx, count in enumerate(doc_row):
                # Convert to scalar value before comparison
                count_value = (
                    float(count.item()) if hasattr(count, "item") else float(count)
                )
                if count_value > 0:
                    doc_counts[self.data.vectorizer.terms_list[term_idx]] = int(
                        count_value
                    )
            doc_data.append(doc_counts)

    elif isinstance(self.data, pd.DataFrame):
        # Make sure there is data
        if self.data.empty == True:
            raise LexosException("Empty DataFrame provided.")
        # Process DataFrame - assume it's a document-term matrix
        doc_data = []
        selected_docs = (
            self.docs if self.docs is not None else range(len(self.data))
        )
        if isinstance(selected_docs, (int, str)):
            selected_docs = [selected_docs]

        for doc_idx in selected_docs:
            if isinstance(doc_idx, str):
                doc_idx = self.data.index.get_loc(doc_idx)
            doc_counts = self.data.iloc[doc_idx].to_dict()
            # Filter out zero counts
            doc_counts = {
                k: v
                for k, v in doc_counts.items()
                if (float(v) if hasattr(v, "item") else v) > 0
            }
            doc_data.append(doc_counts)

    elif isinstance(self.data, list):
        # Make sure the data is not empty
        if not self.data or len(self.data) == 0:
            raise LexosException("No valid data provided for MultiCloud.")
        # Handle list of documents
        doc_data = self.data

    return doc_data

_render() -> None

Generate and display the multi-cloud figure.

Source code in lexos/visualization/cloud.py

def _render(self) -> None:
    """Generate and display the multi-cloud figure."""
    # Calculate layout
    num_clouds = len(self.clouds)
    nrows = int(np.ceil(num_clouds / self.ncols))

    # Set up figure with padding
    figure_opts = self.figure_opts.copy()
    figure_opts.setdefault("figsize", (self.ncols * 4, nrows * 3))

    # Remove constrained_layout if it exists since we're setting manual spacing
    figure_opts.pop("constrained_layout", None)

    self.fig, axes = plt.subplots(nrows, self.ncols, **figure_opts)

    # Add padding between subplots and adjust top margin for title
    if self.title:
        # More space below title when there's a suptitle
        self.fig.subplots_adjust(
            wspace=self.padding,
            hspace=self.padding,
            top=0.82,  # Leaves more space at the top for the title
        )
    else:
        # Normal spacing when no title
        self.fig.subplots_adjust(wspace=self.padding, hspace=self.padding)

    # Add padding between subplots
    self.fig.subplots_adjust(wspace=self.padding, hspace=self.padding)

    # Handle single row case
    if nrows == 1:
        axes = axes.reshape(1, -1) if self.ncols > 1 else np.array([[axes]])
    elif self.ncols == 1:
        axes = axes.reshape(-1, 1)

    # Add overall title
    if self.title:
        self.fig.suptitle(self.title, fontsize=16, y=0.90)  # Positioned lower

    # Plot each word cloud
    for i, cloud in enumerate(self.clouds):
        row = i // self.ncols
        col = i % self.ncols

        ax = axes[row, col]

        # Display the word cloud
        ax.imshow(cloud.cloud.to_array(), interpolation="bilinear")
        ax.axis("off")

        # Add label if provided
        if self.labels and i < len(self.labels):
            ax.set_title(self.labels[i])
        elif hasattr(cloud.data, "__len__"):
            ax.set_title(f"Doc {i + 1}", fontdict={"fontsize": 10})

    # Hide unused subplots
    for i in range(num_clouds, nrows * self.ncols):
        row = i // self.ncols
        col = i % self.ncols
        axes[row, col].axis("off")
        axes[row, col].set_visible(False)

    # Prevent automatic display
    self.fig = plt.gcf()
    plt.close()

save(path: Path | str, **kwargs: Any) -> None

Save the MultiCloud figure to a file.

Parameters:

Name	Type	Description	Default
path	Path | str	The file path to save the MultiCloud image.	required
**kwargs	Any	Additional keyword arguments for plt.savefig.	{}

Source code in lexos/visualization/cloud.py

@validate_call
def save(self, path: Path | str, **kwargs: Any) -> None:
    """Save the MultiCloud figure to a file.

    Args:
        path (Path | str): The file path to save the MultiCloud image.
        **kwargs (Any): Additional keyword arguments for `plt.savefig`.
    """
    if self.fig is None:
        raise LexosException("No figure to save.")
    self.fig.savefig(path, **kwargs)

get_clouds() -> list[WordCloud]

Return the list of individual WordCloud objects.

Source code in lexos/visualization/cloud.py

def get_clouds(self) -> list[WordCloud]:
    """Return the list of individual WordCloud objects."""
    return self.clouds

show() -> plt.Figure

Display the multi-cloud figure.

Source code in lexos/visualization/cloud.py

def show(self) -> plt.Figure:
    """Display the multi-cloud figure."""
    if self.fig is None:
        raise LexosException("No figure to show.")
    return self.fig

D3 Word Clouds

D3WordCloud pydantic-model

Bases: BaseModel

A Pydantic model for D3 WordCloud options.

Config:

• arbitrary_types_allowed: True
• json_schema_extra: DocJSONSchema.model_json_schema()

Fields:

• data (single_doc_types | multi_doc_types | DataFrame)
• docs (Optional[int | str | list[int] | list[str]])
• layout (dict[str, Any])
• limit (int)
• font (str)
• spiral (str)
• scale (str)
• angle_count (int)
• angle_from (int)
• angle_to (int)
• width (int)
• height (int)
• title (str)
• background_color (str)
• colorscale (str)
• auto_open (bool)
• template (Path | str)
• include_d3js (bool | str | None)
• include_d3_cloud (bool | str)
• counts (dict[str, int])
• html (str)

Validators:

• validate_spiral → spiral
• validate_scale → scale
• validate_angles

Source code in lexos/visualization/d3_wordcloud.py

class D3WordCloud(BaseModel):
    """A Pydantic model for D3 WordCloud options."""

    data: single_doc_types | multi_doc_types | pd.DataFrame = Field(
        ...,
        description="The data to generate the word cloud from. Accepts data from a string, list of lists or tuples, a dict with terms as keys and counts/frequencies as values, or a dataframe.",
    )
    docs: Optional[int | str | list[int] | list[str]] = Field(
        None, description="A list of documents to be selected from the DTM."
    )
    layout: dict[str, Any] = Field(default_factory=dict)
    limit: int = Field(100, description="The maximum number of terms in the cloud.")
    font: str = Field("Impact", description="The font to use for the word cloud.")
    spiral: str = Field(
        "archimedean",
        description="The spiral type to use for the word cloud, 'archimedean' or 'rectangular'.",
    )
    scale: str = Field(
        "log",
        description="The scale type to use for the word cloud, 'log', 'sqrt', or 'linear'.",
    )
    angle_count: int = Field(
        5, description="The number of angles to use for the word cloud."
    )
    angle_from: int = Field(-60, description="The starting angle for the word cloud.")
    angle_to: int = Field(60, description="The ending angle for the word cloud.")
    width: int = Field(600, gt=50, description="The width of the word cloud in pixels.")
    height: int = Field(
        600, gt=50, description="The height of the word cloud in pixels."
    )
    title: str = Field(
        "Word Cloud Visualization", description="The title of the word cloud."
    )
    background_color: str = Field(
        "white", description="The background color of the word cloud."
    )
    colorscale: str = Field(
        "d3.scale.category20b",
        description="The name of a categorical d3 scale to use for the word cloud. See https://d3js.org/d3-scale.",
    )
    auto_open: bool = Field(
        True, description="Whether to open the chart in a web browser automatically."
    )
    template: Path | str = Field(
        "d3_cloud_template-1.0.html",
        description="The template file for the word cloud.",
    )
    include_d3js: bool | str | None = Field(
        True,
        description="Whether to include the D3.js library. Can be 'cdn', 'directory', or a custom path. If False, the D3.js library will not be included. The cloud bundle is always included unless the setting is 'directory' or False.",
    )
    include_d3_cloud: bool | str = Field(
        True,
        description="Whether to include the D3 cloud library. Can be a custom path to a JavaScript file or True to use the default bundled version.",
    )
    counts: dict[str, int] = Field({}, description="A dictionary of word counts.")
    html: str = Field("", description="The HTML representation of the word cloud.")

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_schema_extra=DocJSONSchema.model_json_schema(),
    )

    @field_validator("spiral")
    @classmethod
    def validate_spiral(cls, v):
        """Validate the spiral setting."""
        if v not in ["archimedean", "rectangular"]:
            raise ValueError('spiral must be "archimedean" or "rectangular"')
        return v

    @field_validator("scale")
    @classmethod
    def validate_scale(cls, v):
        """Validate the scale setting."""
        if v not in ["log", "sqrt", "linear"]:
            raise ValueError('scale must be "log", "sqrt", or "linear"')
        return v

    @model_validator(mode="after")
    def validate_angles(self):
        """Validate the angle settings."""
        if self.angle_from >= self.angle_to:
            raise ValueError("angle_from must be less than angle_to")
        return self

    def __init__(self, **data: Any) -> None:
        """Initialize with better error handling."""
        try:
            super().__init__(**data)
        except Exception as e:
            raise LexosException(f"Failed to initialize D3WordCloud: {e}") from e

        # Process the data into a consistent format
        self.counts = processors.process_data(self.data, self.docs, self.limit)
        self._render()
        self._include_d3()
        self._include_d3_cloud()

    def _render(self) -> None:
        """Render the word cloud as an HTML string."""
        template = Template(self._load_template())
        self.html = template.render(
            termCounts=json.dumps(self.counts),
            width=self.width,
            height=self.height,
            title=self.title,
            backgroundColor=self.background_color,
            colorscale=self.colorscale,
            font=self.font,
            spiral=self.spiral,
            scale=self.scale,
            angleCount=self.angle_count,
            angleFrom=self.angle_from,
            angleTo=self.angle_to,
        )

        # If auto_open is True, open the chart in a web browser
        if self.auto_open:
            self._open()

    def _get_asset_path(self, filename: str) -> Path:
        """Centralized asset path resolution."""
        return Path(__file__).parent / "d3_cloud_assets" / filename

    def _get_d3_js(self, path: str = "d3.min.js") -> str:
        """Retrieve the contents of the d3.js bundle.

        Args:
            path (str): The path to the d3.js file. Defaults to "d3.min.js".

        Returns:
            str: The HTML script tag containing or pointing to the d3.js script.
        """
        if path == "d3.min.js":
            path = self._get_asset_path("d3.min.js")
        path_obj = Path(path)
        try:
            return f'<script id="d3">\n{_load_local_asset(path_obj)}\n</script>'
        except FileNotFoundError:
            raise LexosException(f"Script file not found: {path}")

    def _include_d3(self) -> None:
        """Modify the template to include d3.js and d3 cloud scripts."""
        # Handle loading/initializing d3.js
        if isinstance(self.include_d3js, str) and self.include_d3js.lower() == "cdn":
            load_d3js = (
                f'<script charset="utf-8" src="https://d3js.org/d3.min.js"></script>'
            )
        elif (
            isinstance(self.include_d3js, str)
            and self.include_d3js.lower() == "directory"
        ):
            load_d3js = f'<script charset="utf-8" src="{self._get_asset_path("d3.min.js")}"></script>'
        elif isinstance(self.include_d3js, str) and self.include_d3js.endswith(".js"):
            load_d3js = self._get_d3_js(self.include_d3js)
        elif self.include_d3js is True:
            load_d3js = self._get_d3_js()
        elif self.include_d3js is False:
            load_d3js = None  # Don't include d3.js
        elif self.include_d3js is None:
            load_d3js = self._get_d3_js()
        if load_d3js:
            self.html = self.html.replace('<script id="d3"></script>', load_d3js)

    def _include_d3_cloud(self) -> None:
        """Modify the template to include d3 cloud scripts."""
        if not self.include_d3_cloud:
            return

        if self.include_d3_cloud is True:
            path = self._get_asset_path("d3cloud_bundle.min.js")
        elif isinstance(self.include_d3_cloud, str) and self.include_d3_cloud.endswith(
            ".js"
        ):
            path = Path(self.include_d3_cloud)
        else:
            return

        self.html = self.html.replace(
            '<script id="d3cloud"></script>',
            f'<script id="d3cloud">\n{_load_local_asset(path)}\n</script>',
        )

    def _load_template(self) -> str:
        """Load the HTML template for the word cloud."""
        template = self._get_asset_path(self.template)
        try:
            return _load_local_asset(template)
        except FileNotFoundError as exc:
            raise LexosException(f"Template file not found: {template}") from exc

    def _minify_html(self, html: str) -> str:
        """Basic HTML minification."""
        # Remove extra whitespace
        html = re.sub(r"\s+", " ", html)
        # Remove comments
        html = re.sub(r"<!--.*?-->", "", html, flags=re.DOTALL)
        return html.strip()

    def _open(self) -> None:
        """Open the HTML file in a web browser."""
        # Create a temporary file to store the HTML
        with tempfile.NamedTemporaryFile(
            "w", delete=False, suffix=".html", encoding="utf-8"
        ) as temp_file:
            temp_file.write(self.html)
            temp_file_path = temp_file.name

            # Open the temporary HTML file in the default web browser
            webbrowser.open(f"file:///{temp_file_path}")

    @validate_call
    def save(self, path: Path | str, minify: bool = False) -> None:
        """Save the word cloud HTML to a file with optional HTML minification."""
        html_content = self.html

        if minify:
            html_content = self._minify_html(html_content)

        with open(path, "w", encoding="utf-8") as f:
            f.write(html_content)

angle_count: int = 5 pydantic-field

The number of angles to use for the word cloud.

angle_from: int = -60 pydantic-field

The starting angle for the word cloud.

angle_to: int = 60 pydantic-field

The ending angle for the word cloud.

auto_open: bool = True pydantic-field

Whether to open the chart in a web browser automatically.

background_color: str = 'white' pydantic-field

The background color of the word cloud.

colorscale: str = 'd3.scale.category20b' pydantic-field

The name of a categorical d3 scale to use for the word cloud. See https://d3js.org/d3-scale.

data: single_doc_types | multi_doc_types | pd.DataFrame pydantic-field

The data to generate the word cloud from. Accepts data from a string, list of lists or tuples, a dict with terms as keys and counts/frequencies as values, or a dataframe.

docs: Optional[int | str | list[int] | list[str]] = None pydantic-field

A list of documents to be selected from the DTM.

font: str = 'Impact' pydantic-field

The font to use for the word cloud.

height: int = 600 pydantic-field

The height of the word cloud in pixels.

html: str = '' pydantic-field

The HTML representation of the word cloud.

include_d3_cloud: bool | str = True pydantic-field

Whether to include the D3 cloud library. Can be a custom path to a JavaScript file or True to use the default bundled version.

include_d3js: bool | str | None = True pydantic-field

Whether to include the D3.js library. Can be 'cdn', 'directory', or a custom path. If False, the D3.js library will not be included. The cloud bundle is always included unless the setting is 'directory' or False.

limit: int = 100 pydantic-field

The maximum number of terms in the cloud.

scale: str = 'log' pydantic-field

The scale type to use for the word cloud, 'log', 'sqrt', or 'linear'.

spiral: str = 'archimedean' pydantic-field

The spiral type to use for the word cloud, 'archimedean' or 'rectangular'.

template: Path | str = 'd3_cloud_template-1.0.html' pydantic-field

The template file for the word cloud.

title: str = 'Word Cloud Visualization' pydantic-field

The title of the word cloud.

width: int = 600 pydantic-field

The width of the word cloud in pixels.

__init__(**data: Any) -> None

Initialize with better error handling.

Source code in lexos/visualization/d3_wordcloud.py

def __init__(self, **data: Any) -> None:
    """Initialize with better error handling."""
    try:
        super().__init__(**data)
    except Exception as e:
        raise LexosException(f"Failed to initialize D3WordCloud: {e}") from e

    # Process the data into a consistent format
    self.counts = processors.process_data(self.data, self.docs, self.limit)
    self._render()
    self._include_d3()
    self._include_d3_cloud()

save(path: Path | str, minify: bool = False) -> None

Save the word cloud HTML to a file with optional HTML minification.

Source code in lexos/visualization/d3_wordcloud.py

@validate_call
def save(self, path: Path | str, minify: bool = False) -> None:
    """Save the word cloud HTML to a file with optional HTML minification."""
    html_content = self.html

    if minify:
        html_content = self._minify_html(html_content)

    with open(path, "w", encoding="utf-8") as f:
        f.write(html_content)

validate_angles() pydantic-validator

Validate the angle settings.

Source code in lexos/visualization/d3_wordcloud.py

@model_validator(mode="after")
def validate_angles(self):
    """Validate the angle settings."""
    if self.angle_from >= self.angle_to:
        raise ValueError("angle_from must be less than angle_to")
    return self

validate_scale(v) pydantic-validator

Validate the scale setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("scale")
@classmethod
def validate_scale(cls, v):
    """Validate the scale setting."""
    if v not in ["log", "sqrt", "linear"]:
        raise ValueError('scale must be "log", "sqrt", or "linear"')
    return v

validate_spiral(v) pydantic-validator

Validate the spiral setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("spiral")
@classmethod
def validate_spiral(cls, v):
    """Validate the spiral setting."""
    if v not in ["archimedean", "rectangular"]:
        raise ValueError('spiral must be "archimedean" or "rectangular"')
    return v

__init__(**data: Any) -> None

Initialize with better error handling.

Source code in lexos/visualization/d3_wordcloud.py

def __init__(self, **data: Any) -> None:
    """Initialize with better error handling."""
    try:
        super().__init__(**data)
    except Exception as e:
        raise LexosException(f"Failed to initialize D3WordCloud: {e}") from e

    # Process the data into a consistent format
    self.counts = processors.process_data(self.data, self.docs, self.limit)
    self._render()
    self._include_d3()
    self._include_d3_cloud()

validate_spiral(v) pydantic-validator

Validate the spiral setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("spiral")
@classmethod
def validate_spiral(cls, v):
    """Validate the spiral setting."""
    if v not in ["archimedean", "rectangular"]:
        raise ValueError('spiral must be "archimedean" or "rectangular"')
    return v

validate_scale(v) pydantic-validator

Validate the scale setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("scale")
@classmethod
def validate_scale(cls, v):
    """Validate the scale setting."""
    if v not in ["log", "sqrt", "linear"]:
        raise ValueError('scale must be "log", "sqrt", or "linear"')
    return v

validate_angles() pydantic-validator

Validate the angle settings.

Source code in lexos/visualization/d3_wordcloud.py

@model_validator(mode="after")
def validate_angles(self):
    """Validate the angle settings."""
    if self.angle_from >= self.angle_to:
        raise ValueError("angle_from must be less than angle_to")
    return self

_render() -> None

Render the word cloud as an HTML string.

Source code in lexos/visualization/d3_wordcloud.py

def _render(self) -> None:
    """Render the word cloud as an HTML string."""
    template = Template(self._load_template())
    self.html = template.render(
        termCounts=json.dumps(self.counts),
        width=self.width,
        height=self.height,
        title=self.title,
        backgroundColor=self.background_color,
        colorscale=self.colorscale,
        font=self.font,
        spiral=self.spiral,
        scale=self.scale,
        angleCount=self.angle_count,
        angleFrom=self.angle_from,
        angleTo=self.angle_to,
    )

    # If auto_open is True, open the chart in a web browser
    if self.auto_open:
        self._open()

_get_asset_path(filename: str) -> Path

Centralized asset path resolution.

Source code in lexos/visualization/d3_wordcloud.py

def _get_asset_path(self, filename: str) -> Path:
    """Centralized asset path resolution."""
    return Path(__file__).parent / "d3_cloud_assets" / filename

_get_d3_js(path: str = 'd3.min.js') -> str

Retrieve the contents of the d3.js bundle.

Parameters:

Name	Type	Description	Default
path	str	The path to the d3.js file. Defaults to "d3.min.js".	'd3.min.js'

Returns:

Name	Type	Description
str	str	The HTML script tag containing or pointing to the d3.js script.

Source code in lexos/visualization/d3_wordcloud.py

def _get_d3_js(self, path: str = "d3.min.js") -> str:
    """Retrieve the contents of the d3.js bundle.

    Args:
        path (str): The path to the d3.js file. Defaults to "d3.min.js".

    Returns:
        str: The HTML script tag containing or pointing to the d3.js script.
    """
    if path == "d3.min.js":
        path = self._get_asset_path("d3.min.js")
    path_obj = Path(path)
    try:
        return f'<script id="d3">\n{_load_local_asset(path_obj)}\n</script>'
    except FileNotFoundError:
        raise LexosException(f"Script file not found: {path}")

_include_d3() -> None

Modify the template to include d3.js and d3 cloud scripts.

Source code in lexos/visualization/d3_wordcloud.py

def _include_d3(self) -> None:
    """Modify the template to include d3.js and d3 cloud scripts."""
    # Handle loading/initializing d3.js
    if isinstance(self.include_d3js, str) and self.include_d3js.lower() == "cdn":
        load_d3js = (
            f'<script charset="utf-8" src="https://d3js.org/d3.min.js"></script>'
        )
    elif (
        isinstance(self.include_d3js, str)
        and self.include_d3js.lower() == "directory"
    ):
        load_d3js = f'<script charset="utf-8" src="{self._get_asset_path("d3.min.js")}"></script>'
    elif isinstance(self.include_d3js, str) and self.include_d3js.endswith(".js"):
        load_d3js = self._get_d3_js(self.include_d3js)
    elif self.include_d3js is True:
        load_d3js = self._get_d3_js()
    elif self.include_d3js is False:
        load_d3js = None  # Don't include d3.js
    elif self.include_d3js is None:
        load_d3js = self._get_d3_js()
    if load_d3js:
        self.html = self.html.replace('<script id="d3"></script>', load_d3js)

_include_d3_cloud() -> None

Modify the template to include d3 cloud scripts.

Source code in lexos/visualization/d3_wordcloud.py

def _include_d3_cloud(self) -> None:
    """Modify the template to include d3 cloud scripts."""
    if not self.include_d3_cloud:
        return

    if self.include_d3_cloud is True:
        path = self._get_asset_path("d3cloud_bundle.min.js")
    elif isinstance(self.include_d3_cloud, str) and self.include_d3_cloud.endswith(
        ".js"
    ):
        path = Path(self.include_d3_cloud)
    else:
        return

    self.html = self.html.replace(
        '<script id="d3cloud"></script>',
        f'<script id="d3cloud">\n{_load_local_asset(path)}\n</script>',
    )

_load_template() -> str

Load the HTML template for the word cloud.

Source code in lexos/visualization/d3_wordcloud.py

def _load_template(self) -> str:
    """Load the HTML template for the word cloud."""
    template = self._get_asset_path(self.template)
    try:
        return _load_local_asset(template)
    except FileNotFoundError as exc:
        raise LexosException(f"Template file not found: {template}") from exc

_minify_html(html: str) -> str

Basic HTML minification.

Source code in lexos/visualization/d3_wordcloud.py

def _minify_html(self, html: str) -> str:
    """Basic HTML minification."""
    # Remove extra whitespace
    html = re.sub(r"\s+", " ", html)
    # Remove comments
    html = re.sub(r"<!--.*?-->", "", html, flags=re.DOTALL)
    return html.strip()

_open() -> None

Open the HTML file in a web browser.

Source code in lexos/visualization/d3_wordcloud.py

def _open(self) -> None:
    """Open the HTML file in a web browser."""
    # Create a temporary file to store the HTML
    with tempfile.NamedTemporaryFile(
        "w", delete=False, suffix=".html", encoding="utf-8"
    ) as temp_file:
        temp_file.write(self.html)
        temp_file_path = temp_file.name

        # Open the temporary HTML file in the default web browser
        webbrowser.open(f"file:///{temp_file_path}")

save(path: Path | str, minify: bool = False) -> None

Save the word cloud HTML to a file with optional HTML minification.

Source code in lexos/visualization/d3_wordcloud.py

@validate_call
def save(self, path: Path | str, minify: bool = False) -> None:
    """Save the word cloud HTML to a file with optional HTML minification."""
    html_content = self.html

    if minify:
        html_content = self._minify_html(html_content)

    with open(path, "w", encoding="utf-8") as f:
        f.write(html_content)

D3MultiCloud pydantic-model

Bases: BaseModel

A Pydantic model for creating multiple D3 WordClouds in a grid layout.

Config:

• arbitrary_types_allowed: True

Fields:

• data_sources (list[multi_doc_types])
• labels (Optional[list[str]])
• cloud_width (int)
• cloud_height (int)
• columns (int)
• title (Optional[str])
• cloud_spacing (int)
• limit (int)
• font (str)
• spiral (str)
• scale (str)
• angle_count (int)
• angle_from (int)
• angle_to (int)
• background_color (str)
• colorscale (str)
• auto_open (bool)
• template (Path | str)
• include_d3js (bool | str | None)
• include_d3_cloud (bool | str)
• word_clouds (list[D3WordCloud])
• html (str)

Validators:

• validate_spiral → spiral
• validate_scale → scale
• validate_angles

Source code in lexos/visualization/d3_wordcloud.py

class D3MultiCloud(BaseModel):
    """A Pydantic model for creating multiple D3 WordClouds in a grid layout."""

    data_sources: list[multi_doc_types] = Field(
        ...,
        description="List of data sources to create individual word clouds from.",
    )
    labels: Optional[list[str]] = Field(
        None,
        description="List of titles for each word cloud. If None, will use 'Cloud 1', 'Cloud 2', etc.",
    )
    cloud_width: int = Field(
        300, gt=50, description="The width of each individual word cloud in pixels."
    )
    cloud_height: int = Field(
        300, gt=50, description="The height of each individual word cloud in pixels."
    )
    columns: int = Field(
        3, gt=0, description="The number of columns in the grid layout."
    )
    title: Optional[str] = Field(None, description="Overall title for the figure.")
    cloud_spacing: int = Field(
        20, ge=0, description="The spacing between clouds in pixels."
    )
    limit: int = Field(50, description="The maximum number of terms in each cloud.")
    font: str = Field("Impact", description="The font to use for all word clouds.")
    spiral: str = Field(
        "archimedean",
        description="The spiral type to use for all word clouds.",
    )
    scale: str = Field(
        "log",
        description="The scale type to use for all word clouds.",
    )
    angle_count: int = Field(
        5, description="The number of angles to use for all word clouds."
    )
    angle_from: int = Field(-60, description="The starting angle for all word clouds.")
    angle_to: int = Field(60, description="The ending angle for all word clouds.")
    background_color: str = Field(
        "white", description="The background color of the overall visualization."
    )
    colorscale: str = Field(
        "d3.scale.category20b",
        description="The name of a categorical d3 scale to use for all word clouds.",
    )
    auto_open: bool = Field(
        True, description="Whether to open the chart in a web browser automatically."
    )
    template: Path | str = Field(
        "d3_multicloud_template-1.0.html",
        description="The template file for the multi-cloud visualization.",
    )
    include_d3js: bool | str | None = Field(
        True,
        description="Whether to include the D3.js library.",
    )
    include_d3_cloud: bool | str = Field(
        True,
        description="Whether to include the D3 cloud library.",
    )

    # Generated fields
    word_clouds: list[D3WordCloud] = Field(
        [], description="List of generated D3WordCloud objects."
    )
    html: str = Field("", description="The HTML representation of the multi-cloud.")

    model_config = ConfigDict(arbitrary_types_allowed=True)

    @field_validator("spiral")
    @classmethod
    def validate_spiral(cls, v):
        """Validate the spiral setting."""
        if v not in ["archimedean", "rectangular"]:
            raise LexosException('spiral must be "archimedean" or "rectangular"')
        return v

    @field_validator("scale")
    @classmethod
    def validate_scale(cls, v):
        """Validate the scale setting."""
        if v not in ["log", "sqrt", "linear"]:
            raise LexosException('scale must be "log", "sqrt", or "linear"')
        return v

    @model_validator(mode="after")
    def validate_angles(self):
        """Validate the angle settings."""
        if self.angle_from >= self.angle_to:
            raise ValueError("angle_from must be less than angle_to")
        return self

    def __init__(self, **data: Any) -> None:
        """Initialize the multi-cloud visualization."""
        try:
            super().__init__(**data)
        except Exception as e:
            raise LexosException(f"Failed to initialize D3MultiCloud: {e}") from e

        # Generate labels if not provided
        if self.labels is None:
            self.labels = [f"Doc {i + 1}" for i in range(len(self.data_sources))]
        elif len(self.labels) != len(self.data_sources):
            raise LexosException(
                "Number of labels must match number of data sources or be None"
            )

        # Generate individual word clouds
        self._generate_word_clouds()

        # Generate the combined HTML
        self._render()

    def _generate_word_clouds(self) -> None:
        """Generate individual D3WordCloud objects for each data source."""
        self.word_clouds = []

        for i, (data_source, label) in enumerate(zip(self.data_sources, self.labels)):
            cloud = D3WordCloud(
                data=data_source,
                width=self.cloud_width,
                height=self.cloud_height,
                title=label,
                limit=self.limit,
                font=self.font,
                spiral=self.spiral,
                scale=self.scale,
                angle_count=self.angle_count,
                angle_from=self.angle_from,
                angle_to=self.angle_to,
                background_color=self.background_color,
                colorscale=self.colorscale,
                auto_open=False,
                include_d3js=False,  # We'll include D3 once in the master template
                include_d3_cloud=False,  # We'll include cloud lib once in the master template
            )
            self.word_clouds.append(cloud)

    def _get_asset_path(self, filename: str) -> Path:
        """Centralized asset path resolution."""
        return Path(__file__).parent / "d3_cloud_assets" / filename

    def _get_cloud(self, index: int) -> D3WordCloud:
        """Get a specific word cloud by index."""
        if 0 <= index < len(self.word_clouds):
            return self.word_clouds[index]
        raise IndexError(f"Cloud index {index} out of range")

    def _get_d3_js(self, path: str = "d3.min.js") -> str:
        """Retrieve the contents of the d3.js bundle."""
        if path == "d3.min.js":
            path = self._get_asset_path("d3.min.js")
        try:
            with open(path) as f:
                return f'<script id="d3">\n{f.read()}\n</script>'
        except FileNotFoundError:
            raise LexosException(f"Script file not found: {path}")

    def _include_d3(self) -> None:
        """Modify the template to include d3.js."""
        if isinstance(self.include_d3js, str) and self.include_d3js.lower() == "cdn":
            load_d3js = (
                '<script charset="utf-8" src="https://d3js.org/d3.v3.min.js"></script>'
            )
        elif (
            isinstance(self.include_d3js, str)
            and self.include_d3js.lower() == "directory"
        ):
            load_d3js = f'<script charset="utf-8" src="{self._get_asset_path("d3.min.js")}"></script>'
        elif isinstance(self.include_d3js, str) and self.include_d3js.endswith(".js"):
            load_d3js = self._get_d3_js(self.include_d3js)
        elif self.include_d3js is True:
            load_d3js = self._get_d3_js()
        elif self.include_d3js is False:
            load_d3js = ""
        else:
            load_d3js = self._get_d3_js()

        self.html = self.html.replace('<script id="d3"></script>', load_d3js)

    def _include_d3_cloud(self) -> None:
        """Modify the template to include d3 cloud scripts."""
        if self.include_d3_cloud is True:
            path = "d3_cloud_assets/d3cloud_bundle.min.js"
        elif isinstance(self.include_d3_cloud, str) and self.include_d3_cloud.endswith(
            ".js"
        ):
            path = self.include_d3_cloud
        else:
            path = "d3_cloud_assets/d3cloud_bundle.min.js"

        if self.include_d3_cloud:
            try:
                with open(path) as f:
                    self.html = self.html.replace(
                        '<script id="d3cloud"></script>',
                        f'<script id="d3cloud">\n{f.read()}\n</script>',
                    )
            except FileNotFoundError:
                # Fallback to CDN
                self.html = self.html.replace(
                    '<script id="d3cloud"></script>',
                    '<script src="https://cdn.jsdelivr.net/gh/jasondavies/d3-cloud/build/d3.layout.cloud.js"></script>',
                )

    def _load_template(self) -> str:
        """Load the HTML template for the multi-cloud visualization."""
        template = self._get_asset_path(self.template)
        with open(template) as f:
            return f.read()

    def _open(self) -> None:
        """Open the HTML file in a web browser."""
        # Create a temporary file to store the HTML
        with tempfile.NamedTemporaryFile(
            "w", delete=False, suffix=".html", encoding="utf-8"
        ) as temp_file:
            temp_file.write(self.html)
            temp_file_path = temp_file.name

            # Open the temporary HTML file in the default web browser
            webbrowser.open(f"file:///{temp_file_path}")

    def _render(self) -> None:
        """Generate the combined HTML for all word clouds."""
        template = Template(self._load_template())

        # Calculate grid dimensions
        rows = (len(self.word_clouds) + self.columns - 1) // self.columns
        total_width = (self.cloud_width * self.columns) + (
            self.cloud_spacing * (self.columns - 1)
        )
        total_height = (self.cloud_height * rows) + (self.cloud_spacing * (rows - 1))

        # Prepare cloud data for template
        cloud_data = []
        for i, cloud in enumerate(self.word_clouds):
            row = i // self.columns
            col = i % self.columns
            x_pos = col * (self.cloud_width + self.cloud_spacing)
            y_pos = row * (self.cloud_height + self.cloud_spacing)

            cloud_data.append(
                {
                    "id": f"cloud_{i}",
                    "title": cloud.title,
                    "termCounts": cloud.counts,
                    "x": x_pos,
                    "y": y_pos,
                    "width": self.cloud_width,
                    "height": self.cloud_height,
                }
            )

        self.html = template.render(
            title=self.title,
            total_width=total_width,
            total_height=total_height + 100,  # Extra space for title
            cloud_data=json.dumps(cloud_data),
            font=self.font,
            spiral=self.spiral,
            scale=self.scale,
            angleCount=self.angle_count,
            angleFrom=self.angle_from,
            angleTo=self.angle_to,
            backgroundColor=self.background_color,
            colorscale=self.colorscale,
        )

        # Include D3 libraries
        self._include_d3()
        self._include_d3_cloud()

        # If auto_open is True, open the chart in a web browser
        if self.auto_open:
            self._open()

    @validate_call
    def get_cloud_counts(self, index: int) -> dict[str, int]:
        """Get word counts for a specific cloud by index."""
        return self._get_cloud(index).counts

    @validate_call
    def save(self, path: Path | str, minify: bool = False) -> None:
        """Save the multi-cloud HTML to a file."""
        html_content = self.html

        if minify:
            import re

            html_content = re.sub(r"\s+", " ", html_content)
            html_content = re.sub(r"<!--.*?-->", "", html_content, flags=re.DOTALL)
            html_content = html_content.strip()

        with open(path, "w", encoding="utf-8") as f:
            f.write(html_content)

angle_count: int = 5 pydantic-field

The number of angles to use for all word clouds.

angle_from: int = -60 pydantic-field

The starting angle for all word clouds.

angle_to: int = 60 pydantic-field

The ending angle for all word clouds.

auto_open: bool = True pydantic-field

Whether to open the chart in a web browser automatically.

background_color: str = 'white' pydantic-field

The background color of the overall visualization.

cloud_height: int = 300 pydantic-field

The height of each individual word cloud in pixels.

cloud_spacing: int = 20 pydantic-field

The spacing between clouds in pixels.

cloud_width: int = 300 pydantic-field

The width of each individual word cloud in pixels.

colorscale: str = 'd3.scale.category20b' pydantic-field

The name of a categorical d3 scale to use for all word clouds.

columns: int = 3 pydantic-field

The number of columns in the grid layout.

data_sources: list[multi_doc_types] pydantic-field

List of data sources to create individual word clouds from.

font: str = 'Impact' pydantic-field

The font to use for all word clouds.

html: str = '' pydantic-field

The HTML representation of the multi-cloud.

include_d3_cloud: bool | str = True pydantic-field

Whether to include the D3 cloud library.

include_d3js: bool | str | None = True pydantic-field

Whether to include the D3.js library.

labels: Optional[list[str]] = None pydantic-field

List of titles for each word cloud. If None, will use 'Cloud 1', 'Cloud 2', etc.

limit: int = 50 pydantic-field

The maximum number of terms in each cloud.

scale: str = 'log' pydantic-field

The scale type to use for all word clouds.

spiral: str = 'archimedean' pydantic-field

The spiral type to use for all word clouds.

template: Path | str = 'd3_multicloud_template-1.0.html' pydantic-field

The template file for the multi-cloud visualization.

title: Optional[str] = None pydantic-field

Overall title for the figure.

word_clouds: list[D3WordCloud] = [] pydantic-field

List of generated D3WordCloud objects.

__init__(**data: Any) -> None

Initialize the multi-cloud visualization.

Source code in lexos/visualization/d3_wordcloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the multi-cloud visualization."""
    try:
        super().__init__(**data)
    except Exception as e:
        raise LexosException(f"Failed to initialize D3MultiCloud: {e}") from e

    # Generate labels if not provided
    if self.labels is None:
        self.labels = [f"Doc {i + 1}" for i in range(len(self.data_sources))]
    elif len(self.labels) != len(self.data_sources):
        raise LexosException(
            "Number of labels must match number of data sources or be None"
        )

    # Generate individual word clouds
    self._generate_word_clouds()

    # Generate the combined HTML
    self._render()

get_cloud_counts(index: int) -> dict[str, int]

Get word counts for a specific cloud by index.

Source code in lexos/visualization/d3_wordcloud.py

@validate_call
def get_cloud_counts(self, index: int) -> dict[str, int]:
    """Get word counts for a specific cloud by index."""
    return self._get_cloud(index).counts

save(path: Path | str, minify: bool = False) -> None

Save the multi-cloud HTML to a file.

Source code in lexos/visualization/d3_wordcloud.py

@validate_call
def save(self, path: Path | str, minify: bool = False) -> None:
    """Save the multi-cloud HTML to a file."""
    html_content = self.html

    if minify:
        import re

        html_content = re.sub(r"\s+", " ", html_content)
        html_content = re.sub(r"<!--.*?-->", "", html_content, flags=re.DOTALL)
        html_content = html_content.strip()

    with open(path, "w", encoding="utf-8") as f:
        f.write(html_content)

validate_angles() pydantic-validator

Validate the angle settings.

Source code in lexos/visualization/d3_wordcloud.py

@model_validator(mode="after")
def validate_angles(self):
    """Validate the angle settings."""
    if self.angle_from >= self.angle_to:
        raise ValueError("angle_from must be less than angle_to")
    return self

validate_scale(v) pydantic-validator

Validate the scale setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("scale")
@classmethod
def validate_scale(cls, v):
    """Validate the scale setting."""
    if v not in ["log", "sqrt", "linear"]:
        raise LexosException('scale must be "log", "sqrt", or "linear"')
    return v

validate_spiral(v) pydantic-validator

Validate the spiral setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("spiral")
@classmethod
def validate_spiral(cls, v):
    """Validate the spiral setting."""
    if v not in ["archimedean", "rectangular"]:
        raise LexosException('spiral must be "archimedean" or "rectangular"')
    return v

__init__(**data: Any) -> None

Initialize the multi-cloud visualization.

Source code in lexos/visualization/d3_wordcloud.py

def __init__(self, **data: Any) -> None:
    """Initialize the multi-cloud visualization."""
    try:
        super().__init__(**data)
    except Exception as e:
        raise LexosException(f"Failed to initialize D3MultiCloud: {e}") from e

    # Generate labels if not provided
    if self.labels is None:
        self.labels = [f"Doc {i + 1}" for i in range(len(self.data_sources))]
    elif len(self.labels) != len(self.data_sources):
        raise LexosException(
            "Number of labels must match number of data sources or be None"
        )

    # Generate individual word clouds
    self._generate_word_clouds()

    # Generate the combined HTML
    self._render()

validate_spiral(v) pydantic-validator

Validate the spiral setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("spiral")
@classmethod
def validate_spiral(cls, v):
    """Validate the spiral setting."""
    if v not in ["archimedean", "rectangular"]:
        raise LexosException('spiral must be "archimedean" or "rectangular"')
    return v

validate_scale(v) pydantic-validator

Validate the scale setting.

Source code in lexos/visualization/d3_wordcloud.py

@field_validator("scale")
@classmethod
def validate_scale(cls, v):
    """Validate the scale setting."""
    if v not in ["log", "sqrt", "linear"]:
        raise LexosException('scale must be "log", "sqrt", or "linear"')
    return v

validate_angles() pydantic-validator

Validate the angle settings.

Source code in lexos/visualization/d3_wordcloud.py

@model_validator(mode="after")
def validate_angles(self):
    """Validate the angle settings."""
    if self.angle_from >= self.angle_to:
        raise ValueError("angle_from must be less than angle_to")
    return self

_generate_word_clouds() -> None

Generate individual D3WordCloud objects for each data source.

Source code in lexos/visualization/d3_wordcloud.py

def _generate_word_clouds(self) -> None:
    """Generate individual D3WordCloud objects for each data source."""
    self.word_clouds = []

    for i, (data_source, label) in enumerate(zip(self.data_sources, self.labels)):
        cloud = D3WordCloud(
            data=data_source,
            width=self.cloud_width,
            height=self.cloud_height,
            title=label,
            limit=self.limit,
            font=self.font,
            spiral=self.spiral,
            scale=self.scale,
            angle_count=self.angle_count,
            angle_from=self.angle_from,
            angle_to=self.angle_to,
            background_color=self.background_color,
            colorscale=self.colorscale,
            auto_open=False,
            include_d3js=False,  # We'll include D3 once in the master template
            include_d3_cloud=False,  # We'll include cloud lib once in the master template
        )
        self.word_clouds.append(cloud)

_get_asset_path(filename: str) -> Path

Centralized asset path resolution.

Source code in lexos/visualization/d3_wordcloud.py

def _get_asset_path(self, filename: str) -> Path:
    """Centralized asset path resolution."""
    return Path(__file__).parent / "d3_cloud_assets" / filename

_get_cloud(index: int) -> D3WordCloud

Get a specific word cloud by index.

Source code in lexos/visualization/d3_wordcloud.py

def _get_cloud(self, index: int) -> D3WordCloud:
    """Get a specific word cloud by index."""
    if 0 <= index < len(self.word_clouds):
        return self.word_clouds[index]
    raise IndexError(f"Cloud index {index} out of range")

_get_d3_js(path: str = 'd3.min.js') -> str

Retrieve the contents of the d3.js bundle.

Source code in lexos/visualization/d3_wordcloud.py

def _get_d3_js(self, path: str = "d3.min.js") -> str:
    """Retrieve the contents of the d3.js bundle."""
    if path == "d3.min.js":
        path = self._get_asset_path("d3.min.js")
    try:
        with open(path) as f:
            return f'<script id="d3">\n{f.read()}\n</script>'
    except FileNotFoundError:
        raise LexosException(f"Script file not found: {path}")

_include_d3() -> None

Modify the template to include d3.js.

Source code in lexos/visualization/d3_wordcloud.py

def _include_d3(self) -> None:
    """Modify the template to include d3.js."""
    if isinstance(self.include_d3js, str) and self.include_d3js.lower() == "cdn":
        load_d3js = (
            '<script charset="utf-8" src="https://d3js.org/d3.v3.min.js"></script>'
        )
    elif (
        isinstance(self.include_d3js, str)
        and self.include_d3js.lower() == "directory"
    ):
        load_d3js = f'<script charset="utf-8" src="{self._get_asset_path("d3.min.js")}"></script>'
    elif isinstance(self.include_d3js, str) and self.include_d3js.endswith(".js"):
        load_d3js = self._get_d3_js(self.include_d3js)
    elif self.include_d3js is True:
        load_d3js = self._get_d3_js()
    elif self.include_d3js is False:
        load_d3js = ""
    else:
        load_d3js = self._get_d3_js()

    self.html = self.html.replace('<script id="d3"></script>', load_d3js)

_include_d3_cloud() -> None

Modify the template to include d3 cloud scripts.

Source code in lexos/visualization/d3_wordcloud.py

def _include_d3_cloud(self) -> None:
    """Modify the template to include d3 cloud scripts."""
    if self.include_d3_cloud is True:
        path = "d3_cloud_assets/d3cloud_bundle.min.js"
    elif isinstance(self.include_d3_cloud, str) and self.include_d3_cloud.endswith(
        ".js"
    ):
        path = self.include_d3_cloud
    else:
        path = "d3_cloud_assets/d3cloud_bundle.min.js"

    if self.include_d3_cloud:
        try:
            with open(path) as f:
                self.html = self.html.replace(
                    '<script id="d3cloud"></script>',
                    f'<script id="d3cloud">\n{f.read()}\n</script>',
                )
        except FileNotFoundError:
            # Fallback to CDN
            self.html = self.html.replace(
                '<script id="d3cloud"></script>',
                '<script src="https://cdn.jsdelivr.net/gh/jasondavies/d3-cloud/build/d3.layout.cloud.js"></script>',
            )

_load_template() -> str

Load the HTML template for the multi-cloud visualization.

Source code in lexos/visualization/d3_wordcloud.py

def _load_template(self) -> str:
    """Load the HTML template for the multi-cloud visualization."""
    template = self._get_asset_path(self.template)
    with open(template) as f:
        return f.read()

_open() -> None

Open the HTML file in a web browser.

Source code in lexos/visualization/d3_wordcloud.py

def _open(self) -> None:
    """Open the HTML file in a web browser."""
    # Create a temporary file to store the HTML
    with tempfile.NamedTemporaryFile(
        "w", delete=False, suffix=".html", encoding="utf-8"
    ) as temp_file:
        temp_file.write(self.html)
        temp_file_path = temp_file.name

        # Open the temporary HTML file in the default web browser
        webbrowser.open(f"file:///{temp_file_path}")

_render() -> None

Generate the combined HTML for all word clouds.

Source code in lexos/visualization/d3_wordcloud.py

def _render(self) -> None:
    """Generate the combined HTML for all word clouds."""
    template = Template(self._load_template())

    # Calculate grid dimensions
    rows = (len(self.word_clouds) + self.columns - 1) // self.columns
    total_width = (self.cloud_width * self.columns) + (
        self.cloud_spacing * (self.columns - 1)
    )
    total_height = (self.cloud_height * rows) + (self.cloud_spacing * (rows - 1))

    # Prepare cloud data for template
    cloud_data = []
    for i, cloud in enumerate(self.word_clouds):
        row = i // self.columns
        col = i % self.columns
        x_pos = col * (self.cloud_width + self.cloud_spacing)
        y_pos = row * (self.cloud_height + self.cloud_spacing)

        cloud_data.append(
            {
                "id": f"cloud_{i}",
                "title": cloud.title,
                "termCounts": cloud.counts,
                "x": x_pos,
                "y": y_pos,
                "width": self.cloud_width,
                "height": self.cloud_height,
            }
        )

    self.html = template.render(
        title=self.title,
        total_width=total_width,
        total_height=total_height + 100,  # Extra space for title
        cloud_data=json.dumps(cloud_data),
        font=self.font,
        spiral=self.spiral,
        scale=self.scale,
        angleCount=self.angle_count,
        angleFrom=self.angle_from,
        angleTo=self.angle_to,
        backgroundColor=self.background_color,
        colorscale=self.colorscale,
    )

    # Include D3 libraries
    self._include_d3()
    self._include_d3_cloud()

    # If auto_open is True, open the chart in a web browser
    if self.auto_open:
        self._open()

get_cloud_counts(index: int) -> dict[str, int]

Get word counts for a specific cloud by index.

Source code in lexos/visualization/d3_wordcloud.py

@validate_call
def get_cloud_counts(self, index: int) -> dict[str, int]:
    """Get word counts for a specific cloud by index."""
    return self._get_cloud(index).counts

save(path: Path | str, minify: bool = False) -> None

Save the multi-cloud HTML to a file.

Source code in lexos/visualization/d3_wordcloud.py

@validate_call
def save(self, path: Path | str, minify: bool = False) -> None:
    """Save the multi-cloud HTML to a file."""
    html_content = self.html

    if minify:
        import re

        html_content = re.sub(r"\s+", " ", html_content)
        html_content = re.sub(r"<!--.*?-->", "", html_content, flags=re.DOTALL)
        html_content = html_content.strip()

    with open(path, "w", encoding="utf-8") as f:
        f.write(html_content)