Coverage for haystack/components/converters/image/file_to_document.py: 96%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import os 

6from pathlib import Path 

7from typing import Any 

8 

9from haystack import Document, component, logging 

10from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata 

11from haystack.dataclasses import ByteStream 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16@component 

17class ImageFileToDocument: 

18 """ 

19 Converts image file references into empty Document objects with associated metadata. 

20 

21 This component is useful in pipelines where image file paths need to be wrapped in `Document` objects to be 

22 processed by downstream components such as the `LLMDocumentContentExtractor` or the 

23 `SentenceTransformersDocumentImageEmbedder` (available in the `sentence-transformers-haystack` integration). 

24 

25 It does **not** extract any content from the image files, instead it creates `Document` objects with `None` as 

26 their content and attaches metadata such as file path and any user-provided values. 

27 

28 ### Usage example 

29 ```python 

30 from haystack.components.converters.image import ImageFileToDocument 

31 

32 converter = ImageFileToDocument() 

33 

34 sources = ["image.jpg", "another_image.png"] 

35 

36 result = converter.run(sources=sources) 

37 documents = result["documents"] 

38 

39 print(documents) 

40 

41 # [Document(id=..., meta: {'file_path': 'image.jpg'}), 

42 # Document(id=..., meta: {'file_path': 'another_image.png'})] 

43 ``` 

44 """ 

45 

46 def __init__(self, *, store_full_path: bool = False) -> None: 

47 """ 

48 Initialize the ImageFileToDocument component. 

49 

50 :param store_full_path: 

51 If True, the full path of the file is stored in the metadata of the document. 

52 If False, only the file name is stored. 

53 """ 

54 self.store_full_path = store_full_path 

55 

56 @component.output_types(documents=list[Document]) 

57 def run( 

58 self, *, sources: list[str | Path | ByteStream], meta: dict[str, Any] | list[dict[str, Any]] | None = None 

59 ) -> dict[str, list[Document]]: 

60 """ 

61 Convert image files into empty Document objects with metadata. 

62 

63 This method accepts image file references (as file paths or ByteStreams) and creates `Document` objects 

64 without content. These documents are enriched with metadata derived from the input source and optional 

65 user-provided metadata. 

66 

67 :param sources: 

68 List of file paths or ByteStream objects to convert. 

69 :param meta: 

70 Optional metadata to attach to the documents. 

71 This value can be a list of dictionaries or a single dictionary. 

72 If it's a single dictionary, its content is added to the metadata of all produced documents. 

73 If it's a list, its length must match the number of sources, as they are zipped together. 

74 For ByteStream objects, their `meta` is added to the output documents. 

75 

76 :returns: 

77 A dictionary containing: 

78 - `documents`: A list of `Document` objects with empty content and associated metadata. 

79 """ 

80 

81 documents = [] 

82 meta_list = normalize_metadata(meta, sources_count=len(sources)) 

83 

84 for source, metadata in zip(sources, meta_list, strict=True): 

85 try: 

86 bytestream = get_bytestream_from_source(source) 

87 except Exception as e: 

88 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e) 

89 continue 

90 

91 merged_metadata = {**bytestream.meta, **metadata} 

92 

93 if not self.store_full_path and (file_path := bytestream.meta.get("file_path")): 

94 merged_metadata["file_path"] = os.path.basename(file_path) 

95 document = Document(content=None, meta=merged_metadata) 

96 documents.append(document) 

97 

98 return {"documents": documents}