Skip to content

Pigeon

zenml.integrations.pigeon special

Initialization of the Pigeon integration.

PigeonIntegration (Integration)

Definition of Pigeon integration for ZenML.

Source code in zenml/integrations/pigeon/__init__.py
class PigeonIntegration(Integration):
    """Definition of Pigeon integration for ZenML."""

    NAME = PIGEON
    REQUIREMENTS = ["ipywidgets>=8.0.0"]

    @classmethod
    def flavors(cls) -> List[Type[Flavor]]:
        """Declare the stack component flavors for the Pigeon integration.

        Returns:
            List of stack component flavors for this integration.
        """
        from zenml.integrations.pigeon.flavors import (
            PigeonAnnotatorFlavor,
        )

        return [PigeonAnnotatorFlavor]

flavors() classmethod

Declare the stack component flavors for the Pigeon integration.

Returns:

Type Description
List[Type[zenml.stack.flavor.Flavor]]

List of stack component flavors for this integration.

Source code in zenml/integrations/pigeon/__init__.py
@classmethod
def flavors(cls) -> List[Type[Flavor]]:
    """Declare the stack component flavors for the Pigeon integration.

    Returns:
        List of stack component flavors for this integration.
    """
    from zenml.integrations.pigeon.flavors import (
        PigeonAnnotatorFlavor,
    )

    return [PigeonAnnotatorFlavor]

annotators special

Initialization of the Pigeon annotators submodule.

pigeon_annotator

Pigeon annotator.

Credit for the implementation of this code to @agermanidis in the Pigeon package and library. This code has been slightly modified to fit the ZenML framework. We use the modified code directly here because the original package (and code) is no longer compatible with more recent versions of ipywidgets.

https://github.com/agermanidis/pigeon

PigeonAnnotator (BaseAnnotator)

Annotator for using Pigeon in Jupyter notebooks.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
class PigeonAnnotator(BaseAnnotator):
    """Annotator for using Pigeon in Jupyter notebooks."""

    @property
    def config(self) -> PigeonAnnotatorConfig:
        """Get the Pigeon annotator config.

        Returns:
            The Pigeon annotator config.
        """
        return cast(PigeonAnnotatorConfig, self._config)

    def get_url(self) -> str:
        """Get the URL of the Pigeon annotator.

        Raises:
            NotImplementedError: Pigeon annotator does not have a URL.
        """
        raise NotImplementedError("Pigeon annotator does not have a URL.")

    def get_url_for_dataset(self, dataset_name: str) -> str:
        """Get the URL of the Pigeon annotator for a specific dataset.

        Args:
            dataset_name: Name of the dataset (annotation file).

        Raises:
            NotImplementedError: Pigeon annotator does not have a URL.
        """
        raise NotImplementedError("Pigeon annotator does not have a URL.")

    def get_datasets(self) -> List[str]:
        """Get a list of datasets (annotation files) in the output directory.

        Returns:
            A list of dataset names (annotation file names) (or empty list when no datasets are present).
        """
        output_dir = self.config.output_dir
        try:
            return [f for f in os.listdir(output_dir) if f.endswith(".txt")]
        except FileNotFoundError:
            return []

    def get_dataset_names(self) -> List[str]:
        """List dataset names (annotation file names) in the output directory.

        Returns:
            A list of dataset names (annotation file names).
        """
        return self.get_datasets()

    def get_dataset_stats(self, dataset_name: str) -> Tuple[int, int]:
        """List labeled and unlabeled examples in a dataset (annotation file).

        Args:
            dataset_name: Name of the dataset (annotation file).

        Returns:
            A tuple containing (num_labeled_examples, num_unlabeled_examples).
        """
        dataset_path = os.path.join(self.config.output_dir, dataset_name)
        num_labeled_examples = 0
        # Placeholder as logic to determine this is not implemented
        num_unlabeled_examples = 0

        try:
            with open(dataset_path, "r") as file:
                num_labeled_examples = sum(1 for _ in file)
        except FileNotFoundError:
            logger.error(f"File not found: {dataset_path}")

        return num_labeled_examples, num_unlabeled_examples

    def _annotate(
        self,
        data: List[Any],
        options: List[str],
        display_fn: Optional[Any] = None,
    ) -> List[Tuple[Any, Any]]:
        """Internal method to build an interactive widget for annotating.

        Args:
            data: List of examples to annotate.
            options: List of labels to choose from.
            display_fn: Optional function to display examples.

        Returns:
            A list of tuples containing (example, label) for each annotated example.
        """
        examples = list(data)
        annotations = []
        current_index = 0
        out = widgets.Output()

        def show_next() -> None:
            nonlocal current_index
            if current_index >= len(examples):
                with out:
                    clear_output(wait=True)
                    logger.info("Annotation done.")
                return
            with out:
                clear_output(wait=True)
                if display_fn:
                    display_fn(examples[current_index])
                else:
                    display(examples[current_index])

        def add_annotation(btn: widgets.Button) -> None:
            """Add an annotation to the list of annotations.

            Args:
                btn: The button that triggered the event.
            """
            nonlocal current_index
            annotation = btn.description
            annotations.append((examples[current_index], annotation))
            current_index += 1
            show_next()

        def submit_annotations(btn: widgets.Button) -> None:
            """Submit all annotations and save them to a file.

            Args:
                btn: The button that triggered the event.
            """
            self._save_annotations(annotations)
            with out:
                clear_output(wait=True)
                logger.info("Annotations saved.")

        count_label = widgets.Label()
        display(count_label)

        buttons = []
        for label in options:
            btn = widgets.Button(description=label)
            btn.on_click(add_annotation)
            buttons.append(btn)

        submit_btn = widgets.Button(
            description="Save labels", button_style="success"
        )
        submit_btn.on_click(submit_annotations)
        buttons.append(submit_btn)

        navigation_box = widgets.HBox(buttons)
        display(navigation_box)
        display(out)
        show_next()

        return annotations

    def launch(self, **kwargs: Any) -> None:
        """Launch the Pigeon annotator in the Jupyter notebook.

        Args:
            **kwargs: Additional keyword arguments to pass to the annotation client.

        Raises:
            NotImplementedError: Pigeon annotator does not support launching with a URL.
        """
        raise NotImplementedError(
            "Pigeon annotator does not support launching with a URL."
        )

    def annotate(
        self,
        data: List[Any],
        options: List[str],
        display_fn: Optional[Any] = None,
    ) -> List[Tuple[Any, Any]]:
        """Annotate with the Pigeon annotator in the Jupyter notebook.

        Args:
            data: List of examples to annotate.
            options: List of labels to choose from.
            display_fn: Optional function to display examples.

        Returns:
            A list of tuples containing (example, label) for each annotated example.
        """
        annotations = self._annotate(data, options, display_fn)
        return annotations

    def _save_annotations(self, annotations: List[Tuple[Any, Any]]) -> None:
        """Save annotations to a file with a unique date-time suffix.

        Args:
            annotations: List of tuples containing (example, label) for each annotated example.
        """
        output_dir = self.config.output_dir
        os.makedirs(output_dir, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = os.path.join(output_dir, f"annotations_{timestamp}.json")
        with open(output_file, "w") as f:
            json.dump(annotations, f)

    def add_dataset(self, **kwargs: Any) -> Any:
        """Add a dataset (annotation file) to the Pigeon annotator.

        Args:
            **kwargs: keyword arguments.

        Raises:
            NotImplementedError: Pigeon annotator does not support adding datasets.
        """
        raise NotImplementedError(
            "Pigeon annotator does not support adding datasets."
        )

    def delete_dataset(self, **kwargs: Any) -> None:
        """Delete a dataset (annotation file).

        Takes the `dataset_name` argument from the kwargs.

        Args:
            **kwargs: Keyword arguments containing the `dataset_name` to delete.

        Raises:
            ValueError: Dataset name is required to delete a dataset.
        """
        dataset_name = kwargs.get("dataset_name")
        if not dataset_name:
            raise ValueError(
                "Dataset name (`dataset_name`) is required to delete a dataset."
            )
        dataset_path = os.path.join(self.config.output_dir, dataset_name)
        os.remove(dataset_path)

    def get_dataset(self, **kwargs: Any) -> List[Tuple[Any, Any]]:
        """Get the annotated examples from a dataset (annotation file).

        Takes the `dataset_name` argument from the kwargs.

        Args:
            **kwargs: Keyword arguments containing the `dataset_name` to retrieve.

        Returns:
            A list of tuples containing (example, label) for each annotated
            example.

        Raises:
            ValueError: Dataset name is required to retrieve a dataset.
        """
        dataset_name = kwargs.get("dataset_name")
        if not dataset_name:
            raise ValueError(
                "Dataset name (`dataset_name`) is required to retrieve a dataset."
            )
        dataset_path = os.path.join(self.config.output_dir, dataset_name)
        with open(dataset_path, "r") as f:
            annotations = json.load(f)
        return cast(List[Tuple[Any, Any]], annotations)

    def get_labeled_data(self, **kwargs: Any) -> List[Tuple[Any, Any]]:
        """Get the labeled examples from a dataset (annotation file).

        Takes the `dataset_name` argument from the kwargs.

        Args:
            **kwargs: Keyword arguments containing the `dataset_name` to retrieve.

        Returns:
            A list of tuples containing (example, label) for each labeled
            example.

        Raises:
            ValueError: Dataset name is required to retrieve labeled data.
        """
        if dataset_name := kwargs.get("dataset_name"):
            return self.get_dataset(dataset_name=dataset_name)
        else:
            raise ValueError(
                "Dataset name (`dataset_name`) is required to retrieve labeled data."
            )

    def get_unlabeled_data(self, **kwargs: Any) -> Any:
        """Get the unlabeled examples from a dataset (annotation file).

        Args:
            **kwargs: keyword arguments.

        Raises:
            NotImplementedError: Pigeon annotator does not support retrieving unlabeled data.
        """
        raise NotImplementedError(
            "Pigeon annotator does not support retrieving unlabeled data."
        )
config: PigeonAnnotatorConfig property readonly

Get the Pigeon annotator config.

Returns:

Type Description
PigeonAnnotatorConfig

The Pigeon annotator config.

add_dataset(self, **kwargs)

Add a dataset (annotation file) to the Pigeon annotator.

Parameters:

Name Type Description Default
**kwargs Any

keyword arguments.

{}

Exceptions:

Type Description
NotImplementedError

Pigeon annotator does not support adding datasets.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def add_dataset(self, **kwargs: Any) -> Any:
    """Add a dataset (annotation file) to the Pigeon annotator.

    Args:
        **kwargs: keyword arguments.

    Raises:
        NotImplementedError: Pigeon annotator does not support adding datasets.
    """
    raise NotImplementedError(
        "Pigeon annotator does not support adding datasets."
    )
annotate(self, data, options, display_fn=None)

Annotate with the Pigeon annotator in the Jupyter notebook.

Parameters:

Name Type Description Default
data List[Any]

List of examples to annotate.

required
options List[str]

List of labels to choose from.

required
display_fn Optional[Any]

Optional function to display examples.

None

Returns:

Type Description
List[Tuple[Any, Any]]

A list of tuples containing (example, label) for each annotated example.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def annotate(
    self,
    data: List[Any],
    options: List[str],
    display_fn: Optional[Any] = None,
) -> List[Tuple[Any, Any]]:
    """Annotate with the Pigeon annotator in the Jupyter notebook.

    Args:
        data: List of examples to annotate.
        options: List of labels to choose from.
        display_fn: Optional function to display examples.

    Returns:
        A list of tuples containing (example, label) for each annotated example.
    """
    annotations = self._annotate(data, options, display_fn)
    return annotations
delete_dataset(self, **kwargs)

Delete a dataset (annotation file).

Takes the dataset_name argument from the kwargs.

Parameters:

Name Type Description Default
**kwargs Any

Keyword arguments containing the dataset_name to delete.

{}

Exceptions:

Type Description
ValueError

Dataset name is required to delete a dataset.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def delete_dataset(self, **kwargs: Any) -> None:
    """Delete a dataset (annotation file).

    Takes the `dataset_name` argument from the kwargs.

    Args:
        **kwargs: Keyword arguments containing the `dataset_name` to delete.

    Raises:
        ValueError: Dataset name is required to delete a dataset.
    """
    dataset_name = kwargs.get("dataset_name")
    if not dataset_name:
        raise ValueError(
            "Dataset name (`dataset_name`) is required to delete a dataset."
        )
    dataset_path = os.path.join(self.config.output_dir, dataset_name)
    os.remove(dataset_path)
get_dataset(self, **kwargs)

Get the annotated examples from a dataset (annotation file).

Takes the dataset_name argument from the kwargs.

Parameters:

Name Type Description Default
**kwargs Any

Keyword arguments containing the dataset_name to retrieve.

{}

Returns:

Type Description
List[Tuple[Any, Any]]

A list of tuples containing (example, label) for each annotated example.

Exceptions:

Type Description
ValueError

Dataset name is required to retrieve a dataset.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_dataset(self, **kwargs: Any) -> List[Tuple[Any, Any]]:
    """Get the annotated examples from a dataset (annotation file).

    Takes the `dataset_name` argument from the kwargs.

    Args:
        **kwargs: Keyword arguments containing the `dataset_name` to retrieve.

    Returns:
        A list of tuples containing (example, label) for each annotated
        example.

    Raises:
        ValueError: Dataset name is required to retrieve a dataset.
    """
    dataset_name = kwargs.get("dataset_name")
    if not dataset_name:
        raise ValueError(
            "Dataset name (`dataset_name`) is required to retrieve a dataset."
        )
    dataset_path = os.path.join(self.config.output_dir, dataset_name)
    with open(dataset_path, "r") as f:
        annotations = json.load(f)
    return cast(List[Tuple[Any, Any]], annotations)
get_dataset_names(self)

List dataset names (annotation file names) in the output directory.

Returns:

Type Description
List[str]

A list of dataset names (annotation file names).

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_dataset_names(self) -> List[str]:
    """List dataset names (annotation file names) in the output directory.

    Returns:
        A list of dataset names (annotation file names).
    """
    return self.get_datasets()
get_dataset_stats(self, dataset_name)

List labeled and unlabeled examples in a dataset (annotation file).

Parameters:

Name Type Description Default
dataset_name str

Name of the dataset (annotation file).

required

Returns:

Type Description
Tuple[int, int]

A tuple containing (num_labeled_examples, num_unlabeled_examples).

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_dataset_stats(self, dataset_name: str) -> Tuple[int, int]:
    """List labeled and unlabeled examples in a dataset (annotation file).

    Args:
        dataset_name: Name of the dataset (annotation file).

    Returns:
        A tuple containing (num_labeled_examples, num_unlabeled_examples).
    """
    dataset_path = os.path.join(self.config.output_dir, dataset_name)
    num_labeled_examples = 0
    # Placeholder as logic to determine this is not implemented
    num_unlabeled_examples = 0

    try:
        with open(dataset_path, "r") as file:
            num_labeled_examples = sum(1 for _ in file)
    except FileNotFoundError:
        logger.error(f"File not found: {dataset_path}")

    return num_labeled_examples, num_unlabeled_examples
get_datasets(self)

Get a list of datasets (annotation files) in the output directory.

Returns:

Type Description
List[str]

A list of dataset names (annotation file names) (or empty list when no datasets are present).

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_datasets(self) -> List[str]:
    """Get a list of datasets (annotation files) in the output directory.

    Returns:
        A list of dataset names (annotation file names) (or empty list when no datasets are present).
    """
    output_dir = self.config.output_dir
    try:
        return [f for f in os.listdir(output_dir) if f.endswith(".txt")]
    except FileNotFoundError:
        return []
get_labeled_data(self, **kwargs)

Get the labeled examples from a dataset (annotation file).

Takes the dataset_name argument from the kwargs.

Parameters:

Name Type Description Default
**kwargs Any

Keyword arguments containing the dataset_name to retrieve.

{}

Returns:

Type Description
List[Tuple[Any, Any]]

A list of tuples containing (example, label) for each labeled example.

Exceptions:

Type Description
ValueError

Dataset name is required to retrieve labeled data.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_labeled_data(self, **kwargs: Any) -> List[Tuple[Any, Any]]:
    """Get the labeled examples from a dataset (annotation file).

    Takes the `dataset_name` argument from the kwargs.

    Args:
        **kwargs: Keyword arguments containing the `dataset_name` to retrieve.

    Returns:
        A list of tuples containing (example, label) for each labeled
        example.

    Raises:
        ValueError: Dataset name is required to retrieve labeled data.
    """
    if dataset_name := kwargs.get("dataset_name"):
        return self.get_dataset(dataset_name=dataset_name)
    else:
        raise ValueError(
            "Dataset name (`dataset_name`) is required to retrieve labeled data."
        )
get_unlabeled_data(self, **kwargs)

Get the unlabeled examples from a dataset (annotation file).

Parameters:

Name Type Description Default
**kwargs Any

keyword arguments.

{}

Exceptions:

Type Description
NotImplementedError

Pigeon annotator does not support retrieving unlabeled data.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_unlabeled_data(self, **kwargs: Any) -> Any:
    """Get the unlabeled examples from a dataset (annotation file).

    Args:
        **kwargs: keyword arguments.

    Raises:
        NotImplementedError: Pigeon annotator does not support retrieving unlabeled data.
    """
    raise NotImplementedError(
        "Pigeon annotator does not support retrieving unlabeled data."
    )
get_url(self)

Get the URL of the Pigeon annotator.

Exceptions:

Type Description
NotImplementedError

Pigeon annotator does not have a URL.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_url(self) -> str:
    """Get the URL of the Pigeon annotator.

    Raises:
        NotImplementedError: Pigeon annotator does not have a URL.
    """
    raise NotImplementedError("Pigeon annotator does not have a URL.")
get_url_for_dataset(self, dataset_name)

Get the URL of the Pigeon annotator for a specific dataset.

Parameters:

Name Type Description Default
dataset_name str

Name of the dataset (annotation file).

required

Exceptions:

Type Description
NotImplementedError

Pigeon annotator does not have a URL.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def get_url_for_dataset(self, dataset_name: str) -> str:
    """Get the URL of the Pigeon annotator for a specific dataset.

    Args:
        dataset_name: Name of the dataset (annotation file).

    Raises:
        NotImplementedError: Pigeon annotator does not have a URL.
    """
    raise NotImplementedError("Pigeon annotator does not have a URL.")
launch(self, **kwargs)

Launch the Pigeon annotator in the Jupyter notebook.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments to pass to the annotation client.

{}

Exceptions:

Type Description
NotImplementedError

Pigeon annotator does not support launching with a URL.

Source code in zenml/integrations/pigeon/annotators/pigeon_annotator.py
def launch(self, **kwargs: Any) -> None:
    """Launch the Pigeon annotator in the Jupyter notebook.

    Args:
        **kwargs: Additional keyword arguments to pass to the annotation client.

    Raises:
        NotImplementedError: Pigeon annotator does not support launching with a URL.
    """
    raise NotImplementedError(
        "Pigeon annotator does not support launching with a URL."
    )

flavors special

Pigeon integration flavors.

pigeon_annotator_flavor

Pigeon annotator flavor.

PigeonAnnotatorConfig (BaseAnnotatorConfig, PigeonAnnotatorSettings, AuthenticationConfigMixin)

Config for the Pigeon annotator.

Attributes:

Name Type Description
output_dir str

The directory to store the annotations.

notebook_only ClassVar[bool]

Whether the annotator only works within a notebook.

Source code in zenml/integrations/pigeon/flavors/pigeon_annotator_flavor.py
class PigeonAnnotatorConfig(
    BaseAnnotatorConfig, PigeonAnnotatorSettings, AuthenticationConfigMixin
):
    """Config for the Pigeon annotator.

    Attributes:
        output_dir: The directory to store the annotations.
        notebook_only: Whether the annotator only works within a notebook.
    """

    output_dir: str = "annotations"
PigeonAnnotatorFlavor (BaseAnnotatorFlavor)

Pigeon annotator flavor.

Source code in zenml/integrations/pigeon/flavors/pigeon_annotator_flavor.py
class PigeonAnnotatorFlavor(BaseAnnotatorFlavor):
    """Pigeon annotator flavor."""

    @property
    def name(self) -> str:
        """Name of the flavor.

        Returns:
            The name of the flavor.
        """
        return PIGEON_ANNOTATOR_FLAVOR

    @property
    def docs_url(self) -> Optional[str]:
        """A url to point at docs explaining this flavor.

        Returns:
            A flavor docs url.
        """
        return self.generate_default_docs_url()

    @property
    def sdk_docs_url(self) -> Optional[str]:
        """A url to point at SDK docs explaining this flavor.

        Returns:
            A flavor SDK docs url.
        """
        return self.generate_default_sdk_docs_url()

    @property
    def logo_url(self) -> str:
        """A url to represent the flavor in the dashboard.

        Returns:
            The flavor logo.
        """
        return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/annotator/pigeon.png"

    @property
    def config_class(self) -> Type[PigeonAnnotatorConfig]:
        """Returns `PigeonAnnotatorConfig` config class.

        Returns:
            The config class.
        """
        return PigeonAnnotatorConfig

    @property
    def implementation_class(self) -> Type["PigeonAnnotator"]:
        """Implementation class for this flavor.

        Returns:
            The implementation class.
        """
        from zenml.integrations.pigeon.annotators import PigeonAnnotator

        return PigeonAnnotator
config_class: Type[zenml.integrations.pigeon.flavors.pigeon_annotator_flavor.PigeonAnnotatorConfig] property readonly

Returns PigeonAnnotatorConfig config class.

Returns:

Type Description
Type[zenml.integrations.pigeon.flavors.pigeon_annotator_flavor.PigeonAnnotatorConfig]

The config class.

docs_url: Optional[str] property readonly

A url to point at docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor docs url.

implementation_class: Type[PigeonAnnotator] property readonly

Implementation class for this flavor.

Returns:

Type Description
Type[PigeonAnnotator]

The implementation class.

logo_url: str property readonly

A url to represent the flavor in the dashboard.

Returns:

Type Description
str

The flavor logo.

name: str property readonly

Name of the flavor.

Returns:

Type Description
str

The name of the flavor.

sdk_docs_url: Optional[str] property readonly

A url to point at SDK docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor SDK docs url.

PigeonAnnotatorSettings (BaseSettings)

Settings for the Pigeon annotator.

Source code in zenml/integrations/pigeon/flavors/pigeon_annotator_flavor.py
class PigeonAnnotatorSettings(BaseSettings):
    """Settings for the Pigeon annotator."""