v3 or something: get a working demo of ibis-birdbrain back in place (#38

)

* agi or bust

* basic communication pieces in place

src/ibis_birdbrain/__init__.py

@@ -0,0 +1,6 @@
+# imports
+
+# exports
+from ibis_birdbrain.bot import Bot
+
+__all__ = ["Bot"]

src/ibis_birdbrain/__main__.py

@@ -0,0 +1,3 @@
+from ibis_birdbrain.cli import app
+
+app(prog_name="birdbrain")

src/ibis_birdbrain/attachments/__init__.py

@@ -0,0 +1,114 @@
+"""Ibis Birdbrain attachments."""
+
+# imports
+from uuid import uuid4
+from typing import Any
+from datetime import datetime
+
+from ibis.expr.types.relations import Table
+
+
+# classes
+class Attachment:
+    """Ibis Birdbrain attachment."""
+
+    content: Any
+    id: str
+    created_at: datetime
+    name: str | None
+    description: str | None
+
+    def __init__(
+        self,
+        content,
+        name=None,
+        description=None,
+    ):
+        self.id = str(uuid4())
+        self.created_at = datetime.now()
+
+        self.name = name
+        self.description = description
+        self.content = content
+
+    def encode(self) -> Table:
+        ...
+
+    def decode(self, t: Table) -> str:
+        ...
+
+    def open(self) -> Any:
+        return self.content
+
+    def __str__(self):
+        return f"""{self.__class__.__name__}
+    **guid**: {self.id}
+    **time**: {self.created_at}
+    **name**: {self.name}
+    **desc**: {self.description}"""
+
+    def __repr__(self):
+        return str(self)
+
+
+class Attachments:
+    """Ibis Birdbrain attachments."""
+
+    attachments: dict[str, Attachment]
+
+    def __init__(self, attachments: list[Attachment] = []) -> None:
+        """Initialize the attachments."""
+        self.attachments = {a.id: a for a in attachments}
+
+    def add_attachment(self, attachment: Attachment):
+        """Add an attachment to the collection."""
+        self.attachments[attachment.id] = attachment
+
+    def append(self, attachment: Attachment):
+        """Alias for add_attachment."""
+        self.add_attachment(attachment)
+
+    def __getitem__(self, id: str | int):
+        """Get an attachment from the collection."""
+        if isinstance(id, int):
+            return list(self.attachments.values())[id]
+        return self.attachments[id]
+
+    def __setitem__(self, id: str, attachment: Attachment):
+        """Set an attachment in the collection."""
+        self.attachments[id] = attachment
+
+    def __len__(self) -> int:
+        """Get the length of the collection."""
+        return len(self.attachments)
+
+    def __iter__(self):
+        """Iterate over the collection."""
+        return iter(self.attachments.keys())
+
+    def __str__(self):
+        return "\n\n".join([str(a) for a in self.attachments.values()])
+
+    def __repr__(self):
+        return str(self)
+
+
+# exports
+from ibis_birdbrain.attachments.viz import ChartAttachment
+from ibis_birdbrain.attachments.data import DataAttachment, TableAttachment
+from ibis_birdbrain.attachments.text import (
+    TextAttachment,
+    CodeAttachment,
+    WebpageAttachment,
+)
+
+__all__ = [
+    "Attachment",
+    "Attachments",
+    "DataAttachment",
+    "TableAttachment",
+    "ChartAttachment",
+    "TextAttachment",
+    "CodeAttachment",
+    "WebpageAttachment",
+]

src/ibis_birdbrain/attachments/data.py

@@ -0,0 +1,77 @@
+# imports
+import ibis
+
+from ibis.backends.base import BaseBackend
+from ibis.expr.types.relations import Table
+
+from ibis_birdbrain.attachments import Attachment
+
+# configure Ibis
+ibis.options.interactive = True
+ibis.options.repr.interactive.max_rows = 10
+ibis.options.repr.interactive.max_columns = 20
+ibis.options.repr.interactive.max_length = 20
+
+
+# classes
+class DataAttachment(Attachment):
+    """A database attachment."""
+
+    content: BaseBackend
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.con = self.content  # alias
+        if self.name is None:
+            try:
+                self.name = (
+                    self.content.current_database + "." + self.content.current_schema
+                )
+            except:
+                self.name = "unknown"
+
+        try:
+            self.sql_dialect = self.content.name
+        except:
+            self.sql_dialect = "unknown"
+        try:
+            self.description = "tables:\n\t" + "\n\t".join(
+                [t for t in self.content.list_tables()]
+            )
+        except:
+            self.description = "empty database\n"
+
+    def __str__(self):
+        return (
+            super().__str__()
+            + f"""
+    **dialect**: {self.sql_dialect}"""
+        )
+
+
+class TableAttachment(Attachment):
+    """A table attachment."""
+
+    content: Table
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        try:
+            self.name = self.content.get_name()
+        except AttributeError:
+            self.name = None
+        self.schema = self.content.schema()
+        self.description = "\n" + str(self.schema)
+
+    def encode(self) -> Table:
+        ...
+
+    def decode(self, t: Table) -> str:
+        ...
+
+    def __str__(self):
+        return (
+            super().__str__()
+            + f"""
+    **table**:\n{self.content}"""
+        )

src/ibis_birdbrain/attachments/text.py

@@ -0,0 +1,99 @@
+# imports
+from ibis_birdbrain.utils.web import webpage_to_str, open_browser
+from ibis_birdbrain.utils.strings import estimate_tokens, shorten_str
+from ibis_birdbrain.attachments import Attachment
+
+
+# classes
+class TextAttachment(Attachment):
+    """A text attachment."""
+
+    content: str
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if estimate_tokens(self.content) > 200:
+            self.display_content = (
+                shorten_str(self.content, 50)
+                + shorten_str(self.content[::-1], 50)[::-1]
+            )
+        else:
+            self.display_content = self.content
+
+    def encode(self):
+        ...
+
+    def decode(self):
+        ...
+
+    def __str__(self):
+        return (
+            super().__str__()
+            + f"""
+    **text**:\n{self.content}"""
+        )
+
+
+class WebpageAttachment(Attachment):
+    """A webpage attachment."""
+
+    content: str
+    url: str
+
+    def __init__(self, *args, url="https://ibis-project.org", **kwargs):
+        super().__init__(*args, **kwargs)
+        self.url = url
+        if self.content is None:
+            self.content = webpage_to_str(self.url)
+        if estimate_tokens(self.content) > 100:
+            self.display_content = (
+                shorten_str(self.content, 50)
+                + shorten_str(self.content[::-1], 50)[::-1]
+            )
+        else:
+            self.display_content = self.content
+
+    def encode(self):
+        ...
+
+    def decode(self):
+        ...
+
+    def __str__(self):
+        return (
+            super().__str__()
+            + f"""
+    **url**: {self.url}
+    **content**:\n{self.display_content}"""
+        )
+
+    def open(self, browser=False):
+        if browser:
+            open_browser(self.url)
+        else:
+            return self.url
+
+
+class CodeAttachment(TextAttachment):
+    """A code attachment."""
+
+    content: str
+    language: str
+
+    def __init__(self, language="python", *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.language = language
+
+    def encode(self):
+        ...
+
+    def decode(self):
+        ...
+
+    def __str__(self):
+        return (
+            super().__str__()
+            + f"""
+    **language**: {self.language}
+    **code**:\n{self.content}"""
+        )

src/ibis_birdbrain/attachments/viz.py

@@ -0,0 +1,27 @@
+# imports
+from plotly.graph_objs import Figure
+
+from ibis_birdbrain.attachments import Attachment
+
+
+# classes
+class ChartAttachment(Attachment):
+    """A chart attachment."""
+
+    content: Figure
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def encode(self):
+        ...
+
+    def decode(self):
+        ...
+
+    def __str__(self):
+        return (
+            super().__str__()
+            + f"""
+    **chart**:\n{self.content}"""
+        )