diff --git a/ammico/display.py b/ammico/display.py index 5248c2a..a69cdad 100644 --- a/ammico/display.py +++ b/ammico/display.py @@ -510,7 +510,7 @@ class AnalysisExplorer: detector_class = identify_function( image_copy, analysis_type=setting_summary_analysis_type, - summary_model_type=setting_summary_model, + model_type=setting_summary_model, list_of_questions=[setting_summary_list_of_questions] if (setting_summary_list_of_questions is not None) else None, diff --git a/ammico/notebooks/facial_expressions.ipynb b/ammico/notebooks/facial_expressions.ipynb index 244ae4d..4e11400 100644 --- a/ammico/notebooks/facial_expressions.ipynb +++ b/ammico/notebooks/facial_expressions.ipynb @@ -61,7 +61,7 @@ "id": "a2bd2153", "metadata": {}, "source": [ - "We select a subset of image files to try facial expression detection on, see the `limit` keyword. The `find_files` function finds image files within a given directory:" + "We select a subset of image files to try facial expression detection on, see the `limit` keyword. The `find_files` function finds image files within a given directory and initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis::" ] }, { @@ -73,31 +73,12 @@ "source": [ "# Here you need to provide the path to your google drive folder\n", "# or local folder containing the images\n", - "images = ammico.find_files(\n", + "mydict = ammico.find_files(\n", " path=\"/content/drive/MyDrive/misinformation-data/\",\n", " limit=10,\n", ")" ] }, - { - "attachments": {}, - "cell_type": "markdown", - "id": "705e7328", - "metadata": {}, - "source": [ - "We need to initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "b37c0c91", - "metadata": {}, - "outputs": [], - "source": [ - "mydict = ammico.initialize_dict(images)" - ] - }, { "attachments": {}, "cell_type": "markdown", diff --git a/ammico/notebooks/get-text-from-image.ipynb b/ammico/notebooks/get-text-from-image.ipynb index 989f5c1..598f9bd 100644 --- a/ammico/notebooks/get-text-from-image.ipynb +++ b/ammico/notebooks/get-text-from-image.ipynb @@ -64,7 +64,7 @@ "id": "fddba721", "metadata": {}, "source": [ - "We select a subset of image files to try the text extraction on, see the `limit` keyword. The `find_files` function finds image files within a given directory: " + "We select a subset of image files to try the text extraction on, see the `limit` keyword. The `find_files` function finds image files within a given directory and initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis: " ] }, { @@ -76,28 +76,11 @@ "source": [ "# Here you need to provide the path to your google drive folder\n", "# or local folder containing the images\n", - "images = ammico.find_files(\n", + "mydict = ammico.find_files(\n", " path=\"/content/drive/MyDrive/misinformation-data/\",\n", " limit=10,\n", - ")" - ] - }, - { - "cell_type": "markdown", - "id": "3a7dfe11", - "metadata": {}, - "source": [ - "We need to initialize the main dictionary that contains all information for the images and is updated through each subsequent analysis:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "8b32409f", - "metadata": {}, - "outputs": [], - "source": [ - "mydict = ammico.initialize_dict(images)" + ")\n", + "mydict" ] }, { diff --git a/ammico/notebooks/image_summary.ipynb b/ammico/notebooks/image_summary.ipynb index 79ee1ff..e99dfe2 100644 --- a/ammico/notebooks/image_summary.ipynb +++ b/ammico/notebooks/image_summary.ipynb @@ -63,21 +63,12 @@ "source": [ "# Here you need to provide the path to your google drive folder\n", "# or local folder containing the images\n", - "images = ammico.find_files(\n", + "mydict = ammico.find_files(\n", " path=\"/content/drive/MyDrive/misinformation-data/\",\n", - " limit=10,\n", - ")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "mydict = ammico.initialize_dict(images)" + " #path=\"../../data/images/\",\n", + " limit=2,\n", + ")\n", + "mydict" ] }, { @@ -112,9 +103,8 @@ }, "outputs": [], "source": [ - "obj = ammico.SummaryDetector(mydict)\n", - "summary_model, summary_vis_processors = obj.load_model(model_type=\"base\") # here we load the base model to the memory. This can dramatically speed up the calculation process then.\n", - "# summary_model, summary_vis_processors = ammico.load_model(\"large\")" + "obj = ammico.SummaryDetector(mydict, analysis_type=\"summary\", model_type=\"base\") # here we load the base model to the memory. This can dramatically speed up the calculation process then.\n", + "#obj = ammico.SummaryDetector(mydict, analysis_type=\"summary\", model_type=\"large\")" ] }, { @@ -126,12 +116,16 @@ "outputs": [], "source": [ "for key in mydict:\n", - " mydict[key] = ammico.SummaryDetector(\n", - " mydict[key], # here we pass the dictionary containing the images\n", - " analysis_type=\"summary\", # here we specify the type of analysis to perform (summary, questions, summary_and_questions)\n", - " summary_model=summary_model, # here we pass the model to use for the analysis\n", - " summary_vis_processors=summary_vis_processors # here we pass the visual processors to use for the analysis\n", - " ).analyse_image()" + " mydict[key] = obj.analyse_image(analysis_type=\"summary\", subdict = mydict[key])" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "mydict" ] }, { @@ -229,11 +223,7 @@ "metadata": {}, "outputs": [], "source": [ - "(\n", - " summary_vqa_model, \n", - " summary_vqa_vis_processors, \n", - " summary_vqa_txt_processors \n", - ") = obj.load_vqa_model() # here we load the VQA model to the memory. This can dramatically speed up the calculation process then.\n" + "obj = ammico.SummaryDetector(mydict, analysis_type=\"questions\", list_of_questions = list_of_questions)" ] }, { @@ -243,13 +233,16 @@ "outputs": [], "source": [ "for key in mydict:\n", - " mydict[key] = ammico.SummaryDetector(\n", - " mydict[key],\n", - " analysis_type=\"questions\",\n", - " summary_vqa_model=summary_vqa_model,\n", - " summary_vqa_vis_processors=summary_vqa_vis_processors,\n", - " summary_vqa_txt_processors=summary_vqa_txt_processors, \n", - " ).analyse_questions(list_of_questions)" + " mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions = list_of_questions)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "mydict" ] }, { @@ -260,6 +253,15 @@ "Or you can perform two types of analysis at a time `analysis_type=\"summary_and_questions\"`." ] }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "obj = ammico.SummaryDetector(mydict, analysis_type=\"summary_and_questions\", model_type=\"base\", list_of_questions = list_of_questions)" + ] + }, { "cell_type": "code", "execution_count": null, @@ -267,15 +269,7 @@ "outputs": [], "source": [ "for key in mydict:\n", - " mydict[key] = ammico.SummaryDetector(\n", - " mydict[key],\n", - " analysis_type=\"summary_and_questions\",\n", - " summary_model=summary_model, \n", - " summary_vis_processors=summary_vis_processors,\n", - " summary_vqa_model=summary_vqa_model,\n", - " summary_vqa_vis_processors=summary_vqa_vis_processors,\n", - " summary_vqa_txt_processors=summary_vqa_txt_processors, \n", - " ).analyse_questions(list_of_questions)" + " mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"summary_and_questions\", list_of_questions = list_of_questions)" ] }, { @@ -336,6 +330,230 @@ "analysis_explorer = ammico.AnalysisExplorer(mydict)\n", "analysis_explorer.run_server(port=8055)" ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# New models\n", + "This is very heavy models. They requare approx 60GB of RAM and they can use 20+GB memory GPUs for acceleration." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "obj = ammico.SummaryDetector(mydict, analysis_type = \"summary_and_questions\", model_type = \"blip2_t5_caption_coco_flant5xl\", device_type= \"cpu\")\n", + "# list of the new models that can be used:\n", + "# \"blip2_t5_pretrain_flant5xxl\",\n", + "# \"blip2_t5_pretrain_flant5xl\",\n", + "# \"blip2_t5_caption_coco_flant5xl\",\n", + "# \"blip2_opt_pretrain_opt2.7b\",\n", + "# \"blip2_opt_pretrain_opt6.7b\",\n", + "# \"blip2_opt_caption_coco_opt2.7b\",\n", + "# \"blip2_opt_caption_coco_opt6.7b\",\n", + "\n", + "# You can use `pretrain_` model types for zero-shot image-to-text generation with prompts.\n", + "# Or you can use `caption_coco_`` model types to generate coco-style captions.\n", + "# `flant5` and `opt` means that the model equipped with FlanT5 and OPT LLMs respectively.\n", + "\n", + "#also you can perform all calculation on cpu if you set device_type= \"cpu\" or gpu if you set device_type= \"cuda\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "for key in mydict:\n", + " mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"summary_and_questions\")\n", + "\n", + "# analysis_type can be \n", + "# \"summary\",\n", + "# \"questions\",\n", + "# \"summary_and_questions\"." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "mydict" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "You can also pass a list of questions to this cell if `analysis_type=\"summary_and_questions\"` or `analysis_type=\"questions\"`. But the format of questions has changed in new models. \n", + "\n", + "Here is an example of a list of questions:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "list_of_questions = [\n", + " \"Question: Are there people in the image? Answer:\",\n", + " \"Question: What is this picture about? Answer:\",\n", + "]" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "for key in mydict:\n", + " mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions=list_of_questions)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "You can also pass a question with previous answers as context into this model and pass in questions like this one to get a more accurate answer:\n", + "\n", + "You can combine as many questions as you want in a single query as a list." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "list_of_questions = [\n", + " \"Question: What country is in the picture? Answer: USA. Question: Why? Answer: Because there is an American flag in the background . Question: Where it comes from? Answer:\",\n", + " \"Question: Which city is this? Answer: Frankfurt. Question: why?\",\n", + "]" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "for key in mydict:\n", + " mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions=list_of_questions)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "mydict" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "You can also ask sequential questions if you pass the argument `cosequential_questions=True`. This means that the answers to previous questions will be passed as context to the next question. However, this method will work a bit slower, because for each image the answers to the questions will not be calculated simultaneously, but sequentially. " + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "list_of_questions = [\n", + " \"Question: Is this picture taken inside or outside? Answer:\",\n", + " \"Question: Why? Answer:\",\n", + "]" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "for key in mydict:\n", + " mydict[key] = obj.analyse_image(subdict = mydict[key], analysis_type=\"questions\", list_of_questions=list_of_questions, consequential_questions=True)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "mydict" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Convert to dataframe and write csv\n", + "\n", + "Convert the dictionary of dictionarys into a dictionary with lists:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "outdict = ammico.append_data_to_dict(mydict)\n", + "df = ammico.dump_df(outdict)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "mydict" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Check the dataframe:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "df.head(10)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Write the csv file:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "df.to_csv(\"/content/drive/MyDrive/misinformation-data/data_out.csv\")" + ] } ], "metadata": { @@ -354,7 +572,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.11.3" + "version": "3.10.13" }, "vscode": { "interpreter": { diff --git a/ammico/notebooks/multimodal_search.ipynb b/ammico/notebooks/multimodal_search.ipynb index b2ed911..242d393 100644 --- a/ammico/notebooks/multimodal_search.ipynb +++ b/ammico/notebooks/multimodal_search.ipynb @@ -66,22 +66,11 @@ }, "outputs": [], "source": [ - "images = ammico.find_files(\n", + "mydict = ammico.find_files(\n", " path=\"/content/drive/MyDrive/misinformation-data/\",\n", " limit=10,\n", - ")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "adf3db21-1f8b-4d44-bbef-ef0acf4623a0", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "mydict = ammico.initialize_dict(images)" + ")\n", + "mydict" ] }, { @@ -497,7 +486,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.11.3" + "version": "3.10.13" } }, "nbformat": 4, diff --git a/ammico/notebooks/objects_expression.ipynb b/ammico/notebooks/objects_expression.ipynb index aca2671..d6b7d47 100644 --- a/ammico/notebooks/objects_expression.ipynb +++ b/ammico/notebooks/objects_expression.ipynb @@ -67,19 +67,11 @@ "source": [ "# Here you need to provide the path to your google drive folder\n", "# or local folder containing the images\n", - "images = ammico.find_files(\n", + "mydict = ammico.find_files(\n", " path=\"/content/drive/MyDrive/misinformation-data/\",\n", " limit=10,\n", - ")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "mydict = ammico.initialize_dict(images)" + ")\n", + "mydict" ] }, { diff --git a/ammico/summary.py b/ammico/summary.py index 21e5fe9..27b701a 100644 --- a/ammico/summary.py +++ b/ammico/summary.py @@ -2,27 +2,56 @@ from ammico.utils import AnalysisMethod from torch import cuda, no_grad from PIL import Image from lavis.models import load_model_and_preprocess +from typing import Optional class SummaryDetector(AnalysisMethod): + allowed_model_types = [ + "base", + "large", + ] + allowed_new_model_types = [ + "blip2_t5_pretrain_flant5xxl", + "blip2_t5_pretrain_flant5xl", + "blip2_t5_caption_coco_flant5xl", + "blip2_opt_pretrain_opt2.7b", + "blip2_opt_pretrain_opt6.7b", + "blip2_opt_caption_coco_opt2.7b", + "blip2_opt_caption_coco_opt6.7b", + ] + all_allowed_model_types = allowed_model_types + allowed_new_model_types + allowed_analysis_types = ["summary", "questions", "summary_and_questions"] + def __init__( self, subdict: dict = {}, - summary_model_type: str = "base", + model_type: str = "base", analysis_type: str = "summary_and_questions", - list_of_questions: str = None, + list_of_questions: Optional[list[str]] = None, summary_model=None, summary_vis_processors=None, summary_vqa_model=None, summary_vqa_vis_processors=None, summary_vqa_txt_processors=None, + summary_vqa_model_new=None, + summary_vqa_vis_processors_new=None, + summary_vqa_txt_processors_new=None, + device_type: Optional[str] = None, ) -> None: """ SummaryDetector class for analysing images using the blip_caption model. Args: subdict (dict, optional): Dictionary containing the image to be analysed. Defaults to {}. - summary_model_type (str, optional): Type of blip_caption model to use. Can be "base" or "large". Defaults to "base". + + model_type (str, optional): Type of model to use. Can be "base" or "large" for blip_caption. Or can be one of the new models: + "blip2_t5_pretrain_flant5xxl", + "blip2_t5_pretrain_flant5xl", + "blip2_t5_caption_coco_flant5xl", + "blip2_opt_pretrain_opt2.7b", + "blip2_opt_pretrain_opt6.7b", + "blip2_opt_caption_coco_opt2.7b", + "blip2_opt_caption_coco_opt6.7b". Defaults to "base". analysis_type (str, optional): Type of analysis to perform. Can be "summary", "questions" or "summary_and_questions". Defaults to "summary_and_questions". list_of_questions (list, optional): List of questions to answer. Defaults to ["Are there people in the image?", "What is this picture about?"]. summary_model ([type], optional): blip_caption model. Defaults to None. @@ -30,6 +59,9 @@ class SummaryDetector(AnalysisMethod): summary_vqa_model ([type], optional): blip_vqa model. Defaults to None. summary_vqa_vis_processors ([type], optional): Preprocessors for vqa visual inputs. Defaults to None. summary_vqa_txt_processors ([type], optional): Preprocessors for vqa text inputs. Defaults to None. + summary_vqa_model_new ([type], optional): new_vqa model. Defaults to None. + summary_vqa_vis_processors_new ([type], optional): Preprocessors for vqa visual inputs. Defaults to None. + summary_vqa_txt_processors_new ([type], optional): Preprocessors for vqa text inputs. Defaults to None. Raises: ValueError: If analysis_type is not one of "summary", "questions" or "summary_and_questions". @@ -39,47 +71,67 @@ class SummaryDetector(AnalysisMethod): """ super().__init__(subdict) - if analysis_type not in ["summary", "questions", "summary_and_questions"]: + # check if analysis_type is valid + if analysis_type not in self.allowed_analysis_types: raise ValueError( - "analysis_type must be one of 'summary', 'questions' or 'summary_and_questions'" + "analysis_type must be one of {}".format(self.allowed_analysis_types) ) - self.summary_device = "cuda" if cuda.is_available() else "cpu" - allowed_model_types = ["base", "large"] - if summary_model_type not in allowed_model_types: + # check if device_type is valid + if device_type is None: + self.summary_device = "cuda" if cuda.is_available() else "cpu" + elif device_type not in ["cuda", "cpu"]: + raise ValueError("device_type must be one of {}".format(["cuda", "cpu"])) + else: + self.summary_device = device_type + # check if model_type is valid + if model_type not in self.all_allowed_model_types: raise ValueError( "Model type is not allowed - please select one of {}".format( - allowed_model_types + self.all_allowed_model_types ) ) - self.summary_model_type = summary_model_type + self.model_type = model_type self.analysis_type = analysis_type - if list_of_questions is None: + # check if list_of_questions is valid + if list_of_questions is None and model_type in self.allowed_model_types: self.list_of_questions = [ "Are there people in the image?", "What is this picture about?", ] + elif list_of_questions is None and model_type in self.allowed_new_model_types: + self.list_of_questions = [ + "Question: Are there people in the image? Answer:", + "Question: What is this picture about? Answer:", + ] elif (not isinstance(list_of_questions, list)) or ( not all(isinstance(i, str) for i in list_of_questions) ): - raise ValueError("list_of_questions must be a list of string (questions)") + raise ValueError( + "list_of_questions must be a list of string (questions)" + ) # add sequence of questions else: self.list_of_questions = list_of_questions + # load models and preprocessors if ( - (summary_model is None) + model_type in self.allowed_model_types + and (summary_model is None) and (summary_vis_processors is None) - and (analysis_type != "questions") + and (analysis_type == "summary" or analysis_type == "summary_and_questions") ): self.summary_model, self.summary_vis_processors = self.load_model( - model_type=summary_model_type + model_type=model_type ) else: self.summary_model = summary_model self.summary_vis_processors = summary_vis_processors if ( - (summary_vqa_model is None) + model_type in self.allowed_model_types + and (summary_vqa_model is None) and (summary_vqa_vis_processors is None) and (summary_vqa_txt_processors is None) - and (analysis_type != "summary") + and ( + analysis_type == "questions" or analysis_type == "summary_and_questions" + ) ): ( self.summary_vqa_model, @@ -90,6 +142,21 @@ class SummaryDetector(AnalysisMethod): self.summary_vqa_model = summary_vqa_model self.summary_vqa_vis_processors = summary_vqa_vis_processors self.summary_vqa_txt_processors = summary_vqa_txt_processors + if ( + model_type in self.allowed_new_model_types + and (summary_vqa_model_new is None) + and (summary_vqa_vis_processors_new is None) + and (summary_vqa_txt_processors_new is None) + ): + ( + self.summary_vqa_model_new, + self.summary_vqa_vis_processors_new, + self.summary_vqa_txt_processors_new, + ) = self.load_new_model(model_type=model_type) + else: + self.summary_vqa_model_new = summary_vqa_model_new + self.summary_vqa_vis_processors_new = summary_vqa_vis_processors_new + self.summary_vqa_txt_processors_new = summary_vqa_txt_processors_new def load_model_base(self): """ @@ -98,8 +165,8 @@ class SummaryDetector(AnalysisMethod): Args: Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. + summary_model (torch.nn.Module): model. + summary_vis_processors (dict): preprocessors for visual inputs. """ summary_model, summary_vis_processors, _ = load_model_and_preprocess( name="blip_caption", @@ -116,8 +183,8 @@ class SummaryDetector(AnalysisMethod): Args: Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. + summary_model (torch.nn.Module): model. + summary_vis_processors (dict): preprocessors for visual inputs. """ summary_model, summary_vis_processors, _ = load_model_and_preprocess( name="blip_caption", @@ -135,8 +202,8 @@ class SummaryDetector(AnalysisMethod): model_type (str): type of the model. Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. + summary_model (torch.nn.Module): model. + summary_vis_processors (dict): preprocessors for visual inputs. """ select_model = { "base": SummaryDetector.load_model_base, @@ -152,9 +219,9 @@ class SummaryDetector(AnalysisMethod): Args: Returns: - model (torch.nn.Module): model. - vis_processors (dict): preprocessors for visual inputs. - txt_processors (dict): preprocessors for text inputs. + summary_vqa_model (torch.nn.Module): model. + summary_vqa_vis_processors (dict): preprocessors for visual inputs. + summary_vqa_txt_processors (dict): preprocessors for text inputs. """ ( @@ -169,99 +236,405 @@ class SummaryDetector(AnalysisMethod): ) return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors - def analyse_image(self): + def analyse_image( + self, + analysis_type: Optional[str] = None, + subdict: dict = None, + list_of_questions: Optional[list[str]] = None, + consequential_questions: bool = False, + ): """ Analyse image with blip_caption model. Args: + analysis_type (str): type of the analysis. + subdict (dict): dictionary with analising pictures. + list_of_questions (list[str]): list of questions. + consequential_questions (bool): whether to ask consequential questions. Works only for new BLIP2 models. Returns: self.subdict (dict): dictionary with analysis results. """ - if self.analysis_type == "summary_and_questions": - self.analyse_summary() - self.analyse_questions(self.list_of_questions) - elif self.analysis_type == "summary": - self.analyse_summary() - elif self.analysis_type == "questions": - self.analyse_questions(self.list_of_questions) + if analysis_type is None: + analysis_type = self.analysis_type + if subdict is not None: + self.subdict = subdict + if list_of_questions is not None: + self.list_of_questions = list_of_questions + if analysis_type == "summary_and_questions": + if ( + self.model_type in self.allowed_model_types + and self.analysis_type != "summary_and_questions" + ): # if model_type is not new and required model is absent + if self.summary_model is None: # load summary model if it is not loaded + self.summary_model, self.summary_vis_processors = self.load_model( + model_type=self.model_type + ) + elif ( + self.summary_vqa_model is None + ): # load vqa model if it is not loaded + ( + self.summary_vqa_model, + self.summary_vqa_vis_processors, + self.summary_vqa_txt_processors, + ) = self.load_vqa_model() + self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis + self.analyse_summary(nondeterministic_summaries=True) + self.analyse_questions(self.list_of_questions, consequential_questions) + elif analysis_type == "summary": + if ( + (self.model_type in self.allowed_model_types) + and (self.analysis_type == "questions") + and (self.summary_model is None) + ): # if model_type is not new and required model is absent + ( + self.summary_model, + self.summary_vis_processors, + ) = self.load_model( # load summary model if it is not loaded + model_type=self.model_type + ) + self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis + self.analyse_summary(nondeterministic_summaries=True) + elif analysis_type == "questions": + if ( + (self.model_type in self.allowed_model_types) + and (self.analysis_type == "summary") + and (self.summary_vqa_model is None) + ): # if model_type is not new and required model is absent + ( + self.summary_vqa_model, # load vqa model if it is not loaded + self.summary_vqa_vis_processors, + self.summary_vqa_txt_processors, + ) = self.load_vqa_model() + self.analysis_type = "summary_and_questions" # now all models are loaded, so you can perform any analysis + self.analyse_questions(self.list_of_questions, consequential_questions) + else: + raise ValueError( + "analysis_type must be one of {}".format(self.allowed_analysis_types) + ) return self.subdict - def analyse_summary(self): + def analyse_summary(self, nondeterministic_summaries: bool = True): """ Create 1 constant and 3 non deterministic captions for image. Args: + nondeterministic_summaries (bool): whether to create 3 non deterministic captions. Returns: self.subdict (dict): dictionary with analysis results. """ - + if self.model_type in self.allowed_model_types: + vis_processors = self.summary_vis_processors + model = self.summary_model + elif self.model_type in self.allowed_new_model_types: + vis_processors = self.summary_vqa_vis_processors_new + model = self.summary_vqa_model_new + else: + raise ValueError( + "Model type is not allowed - please select one of {}".format( + self.all_allowed_model_types + ) + ) path = self.subdict["filename"] raw_image = Image.open(path).convert("RGB") - image = ( - self.summary_vis_processors["eval"](raw_image) - .unsqueeze(0) - .to(self.summary_device) - ) + image = vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device) with no_grad(): - self.subdict["const_image_summary"] = self.summary_model.generate( - {"image": image} - )[0] - self.subdict["3_non-deterministic summary"] = self.summary_model.generate( - {"image": image}, use_nucleus_sampling=True, num_captions=3 - ) + self.subdict["const_image_summary"] = model.generate({"image": image})[0] + if nondeterministic_summaries: + self.subdict["3_non-deterministic summary"] = model.generate( + {"image": image}, use_nucleus_sampling=True, num_captions=3 + ) return self.subdict - def analyse_questions(self, list_of_questions: list[str]) -> dict: + def analyse_questions( + self, list_of_questions: list[str], consequential_questions: bool = False + ) -> dict: """ Generate answers to free-form questions about image written in natural language. Args: list_of_questions (list[str]): list of questions. + consequential_questions (bool): whether to ask consequential questions. Works only for new BLIP2 models. Returns: self.subdict (dict): dictionary with answers to questions. """ - if ( - (self.summary_vqa_model is None) - and (self.summary_vqa_vis_processors is None) - and (self.summary_vqa_txt_processors is None) - ): - ( - self.summary_vqa_model, - self.summary_vqa_vis_processors, - self.summary_vqa_txt_processors, - ) = load_model_and_preprocess( - name="blip_vqa", - model_type="vqav2", - is_eval=True, - device=self.summary_device, - ) + model, vis_processors, txt_processors, model_old = self.check_model() if len(list_of_questions) > 0: path = self.subdict["filename"] raw_image = Image.open(path).convert("RGB") image = ( - self.summary_vqa_vis_processors["eval"](raw_image) - .unsqueeze(0) - .to(self.summary_device) + vis_processors["eval"](raw_image).unsqueeze(0).to(self.summary_device) ) question_batch = [] - for quest in list_of_questions: - question_batch.append(self.summary_vqa_txt_processors["eval"](quest)) + list_of_questions_processed = [] + + if model_old: + for quest in list_of_questions: + list_of_questions_processed.append(txt_processors["eval"](quest)) + else: + for quest in list_of_questions: + list_of_questions_processed.append((str)(quest)) + + for quest in list_of_questions_processed: + question_batch.append(quest) batch_size = len(list_of_questions) image_batch = image.repeat(batch_size, 1, 1, 1) - with no_grad(): - answers_batch = self.summary_vqa_model.predict_answers( - samples={"image": image_batch, "text_input": question_batch}, - inference_method="generate", + if not consequential_questions: + with no_grad(): + if model_old: + answers_batch = model.predict_answers( + samples={ + "image": image_batch, + "text_input": question_batch, + }, + inference_method="generate", + ) + else: + answers_batch = model.generate( + {"image": image_batch, "prompt": question_batch} + ) + + for q, a in zip(list_of_questions, answers_batch): + self.subdict[q] = a + + if consequential_questions and not model_old: + query_with_context = "" + for quest in question_batch: + query_with_context = query_with_context + quest + with no_grad(): + answer = model.generate( + {"image": image, "prompt": query_with_context} + ) + self.subdict[query_with_context] = answer[0] + query_with_context = query_with_context + " " + answer[0] + ". " + elif consequential_questions and model_old: + raise ValueError( + "Consequential questions are not allowed for old models" ) - - for q, a in zip(list_of_questions, answers_batch): - self.subdict[q] = a - else: print("Please, enter list of questions") return self.subdict + + def check_model(self): + """ + Check model type and return appropriate model and preprocessors. + + Args: + + Returns: + model (nn.Module): model. + vis_processors (dict): visual preprocessor. + txt_processors (dict): text preprocessor. + model_old (bool): whether model is old or new. + """ + if self.model_type in self.allowed_model_types: + vis_processors = self.summary_vqa_vis_processors + model = self.summary_vqa_model + txt_processors = self.summary_vqa_txt_processors + model_old = True + elif self.model_type in self.allowed_new_model_types: + vis_processors = self.summary_vqa_vis_processors_new + model = self.summary_vqa_model_new + txt_processors = self.summary_vqa_txt_processors_new + model_old = False + else: + raise ValueError( + "Model type is not allowed - please select one of {}".format( + self.all_allowed_model_types + ) + ) + + return model, vis_processors, txt_processors, model_old + + def load_new_model(self, model_type: str): + """ + Load new BLIP2 models. + + Args: + model_type (str): type of the model. + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + select_model = { + "blip2_t5_pretrain_flant5xxl": SummaryDetector.load_model_blip2_t5_pretrain_flant5xxl, + "blip2_t5_pretrain_flant5xl": SummaryDetector.load_model_blip2_t5_pretrain_flant5xl, + "blip2_t5_caption_coco_flant5xl": SummaryDetector.load_model_blip2_t5_caption_coco_flant5xl, + "blip2_opt_pretrain_opt2.7b": SummaryDetector.load_model_blip2_opt_pretrain_opt27b, + "blip2_opt_pretrain_opt6.7b": SummaryDetector.load_model_base_blip2_opt_pretrain_opt67b, + "blip2_opt_caption_coco_opt2.7b": SummaryDetector.load_model_blip2_opt_caption_coco_opt27b, + "blip2_opt_caption_coco_opt6.7b": SummaryDetector.load_model_base_blip2_opt_caption_coco_opt67b, + } + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = select_model[model_type](self) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_blip2_t5_pretrain_flant5xxl(self): + """ + Load BLIP2 model with FLAN-T5 XXL architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_t5", + model_type="pretrain_flant5xxl", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_blip2_t5_pretrain_flant5xl(self): + """ + Load BLIP2 model with FLAN-T5 XL architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_t5", + model_type="pretrain_flant5xl", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_blip2_t5_caption_coco_flant5xl(self): + """ + Load BLIP2 model with caption_coco_flant5xl architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_t5", + model_type="caption_coco_flant5xl", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_blip2_opt_pretrain_opt27b(self): + """ + Load BLIP2 model with pretrain_opt2 architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_opt", + model_type="pretrain_opt2.7b", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_base_blip2_opt_pretrain_opt67b(self): + """ + Load BLIP2 model with pretrain_opt6.7b architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_opt", + model_type="pretrain_opt6.7b", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_blip2_opt_caption_coco_opt27b(self): + """ + Load BLIP2 model with caption_coco_opt2.7b architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_opt", + model_type="caption_coco_opt2.7b", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors + + def load_model_base_blip2_opt_caption_coco_opt67b(self): + """ + Load BLIP2 model with caption_coco_opt6.7b architecture. + + Args: + + Returns: + model (torch.nn.Module): model. + vis_processors (dict): preprocessors for visual inputs. + txt_processors (dict): preprocessors for text inputs. + """ + ( + summary_vqa_model, + summary_vqa_vis_processors, + summary_vqa_txt_processors, + ) = load_model_and_preprocess( + name="blip2_opt", + model_type="caption_coco_opt6.7b", + is_eval=True, + device=self.summary_device, + ) + return summary_vqa_model, summary_vqa_vis_processors, summary_vqa_txt_processors diff --git a/ammico/test/test_summary.py b/ammico/test/test_summary.py index b638841..0fe8224 100644 --- a/ammico/test/test_summary.py +++ b/ammico/test/test_summary.py @@ -125,11 +125,11 @@ def test_advanced_init_summary(): sd = sm.SummaryDetector({}) assert sd.summary_model assert sd.summary_vis_processors - sd = sm.SummaryDetector({}, summary_model_type="large") + sd = sm.SummaryDetector({}, model_type="large") assert sd.summary_model assert sd.summary_vis_processors with pytest.raises(ValueError): - sm.SummaryDetector({}, summary_model_type="bla") + sm.SummaryDetector({}, model_type="bla") ( summary_vqa_model, summary_vqa_vis_processors,