Coverage for haystack/components/converters/image/file_to_image.py: 100%

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

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import mimetypes 

6from dataclasses import replace 

7from pathlib import Path 

8from typing import Any, Literal 

9 

10from haystack import component, logging 

11from haystack.components.converters.image.image_utils import _encode_image_to_base64 

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

13from haystack.dataclasses import ByteStream 

14from haystack.dataclasses.image_content import ImageContent 

15from haystack.lazy_imports import LazyImport 

16 

17with LazyImport( 

18 "The 'size' parameter is set. " 

19 "Image resizing will be applied, which requires the Pillow library. " 

20 "Run 'pip install pillow'" 

21) as pillow_import: 

22 import PIL # noqa: F401 

23 

24logger = logging.getLogger(__name__) 

25 

26 

27_EMPTY_BYTE_STRING = b"" 

28 

29 

30@component 

31class ImageFileToImageContent: 

32 """ 

33 Converts image files to ImageContent objects. 

34 

35 ### Usage example 

36 ```python 

37 from haystack.components.converters.image import ImageFileToImageContent 

38 

39 converter = ImageFileToImageContent() 

40 

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

42 

43 image_contents = converter.run(sources=sources)["image_contents"] 

44 print(image_contents) 

45 

46 # [ImageContent(base64_image='...', 

47 # mime_type='image/jpeg', 

48 # detail=None, 

49 # meta={'file_path': 'image.jpg'}), 

50 # ...] 

51 ``` 

52 """ 

53 

54 def __init__( 

55 self, *, detail: Literal["auto", "high", "low"] | None = None, size: tuple[int, int] | None = None 

56 ) -> None: 

57 """ 

58 Create the ImageFileToImageContent component. 

59 

60 :param detail: Optional detail level of the image (only supported by OpenAI). One of "auto", "high", or "low". 

61 This will be passed to the created ImageContent objects. 

62 :param size: If provided, resizes the image to fit within the specified dimensions (width, height) while 

63 maintaining aspect ratio. This reduces file size, memory usage, and processing time, which is beneficial 

64 when working with models that have resolution constraints or when transmitting images to remote services. 

65 """ 

66 self.detail = detail 

67 self.size = size 

68 

69 if self.size is not None: 

70 pillow_import.check() 

71 

72 @component.output_types(image_contents=list[ImageContent]) 

73 def run( 

74 self, 

75 sources: list[str | Path | ByteStream], 

76 meta: dict[str, Any] | list[dict[str, Any]] | None = None, 

77 *, 

78 detail: Literal["auto", "high", "low"] | None = None, 

79 size: tuple[int, int] | None = None, 

80 ) -> dict[str, list[ImageContent]]: 

81 """ 

82 Converts files to ImageContent objects. 

83 

84 :param sources: 

85 List of file paths or ByteStream objects to convert. 

86 :param meta: 

87 Optional metadata to attach to the ImageContent objects. 

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

89 If it's a single dictionary, its content is added to the metadata of all produced ImageContent objects. 

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

91 For ByteStream objects, their `meta` is added to the output ImageContent objects. 

92 :param detail: 

93 Optional detail level of the image (only supported by OpenAI). One of "auto", "high", or "low". 

94 This will be passed to the created ImageContent objects. 

95 If not provided, the detail level will be the one set in the constructor. 

96 :param size: If provided, resizes the image to fit within the specified dimensions (width, height) while 

97 maintaining aspect ratio. This reduces file size, memory usage, and processing time, which is beneficial 

98 when working with models that have resolution constraints or when transmitting images to remote services. 

99 If not provided, the size value will be the one set in the constructor. 

100 

101 :returns: 

102 A dictionary with the following keys: 

103 - `image_contents`: A list of ImageContent objects. 

104 """ 

105 if not sources: 

106 return {"image_contents": []} 

107 

108 resolved_detail = detail or self.detail 

109 resolved_size = size or self.size 

110 

111 # Check import 

112 if resolved_size: 

113 pillow_import.check() 

114 

115 image_contents = [] 

116 

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

118 

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

120 if isinstance(source, str): 

121 source = Path(source) 

122 

123 try: 

124 bytestream = get_bytestream_from_source(source) 

125 except Exception as e: 

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

127 continue 

128 

129 if bytestream.mime_type is None and isinstance(source, Path): 

130 bytestream = replace(bytestream, mime_type=mimetypes.guess_type(source.as_posix())[0]) 

131 

132 if bytestream.data == _EMPTY_BYTE_STRING: 

133 logger.warning("File {source} is empty. Skipping it.", source=source) 

134 continue 

135 

136 try: 

137 inferred_mime_type, base64_image = _encode_image_to_base64(bytestream=bytestream, size=resolved_size) 

138 except Exception as e: 

139 logger.warning( 

140 "Could not convert file {source}. Skipping it. Error message: {error}", source=source, error=e 

141 ) 

142 continue 

143 

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

145 image_content = ImageContent( 

146 base64_image=base64_image, mime_type=inferred_mime_type, meta=merged_metadata, detail=resolved_detail 

147 ) 

148 image_contents.append(image_content) 

149 

150 return {"image_contents": image_contents}