Coverage for src/bioimageio/spec/_description_impl.py: 84%

1"""implementation details for building a bioimage.io resource description""" 

2 

3import collections.abc 

4from typing import Any, Callable, List, Literal, Mapping, Optional, Type, TypeVar, Union 

5 

6from ._internal.common_nodes import InvalidDescr, ResourceDescrBase 

7from ._internal.field_validation import issue_warning 

8from ._internal.io import BioimageioYamlContentView 

9from ._internal.types import FormatVersionPlaceholder 

10from ._internal.validation_context import ValidationContext, get_validation_context 

11from .summary import ( 

12 ErrorEntry, 

13 ValidationDetail, 

14) 

15 

16ResourceDescrT = TypeVar("ResourceDescrT", bound=ResourceDescrBase) 

17 

18 

19DISCOVER: Literal["discover"] = "discover" 

20"""placeholder for whatever format version an RDF specifies""" 

21 

22 

23def get_rd_class_impl( 

24 typ: Any, 

25 format_version: Any, 

26 descriptions_map: Mapping[Optional[str], Mapping[str, Type[ResourceDescrT]]], 

27 fallback_to_latest: bool, 

28) -> Type[ResourceDescrT]: 

29 """get the resource description class for the given type and format version""" 

30 assert None in descriptions_map 

31 assert all("latest" in version_map for version_map in descriptions_map.values()) 

32 assert all( 

33 fv == "latest" or fv.count(".") == 1 

34 for version_map in descriptions_map.values() 

35 for fv in version_map 

36 ) 

37 if not isinstance(typ, str) or typ not in descriptions_map: 

38 typ = None 

39 

40 format_version = str(format_version) 

41 if format_version == "latest" or (ndots := format_version.count(".")) == 1: 

42 use_format_version = format_version 

43 elif ndots == 0: 

44 use_format_version = format_version + ".0" 

45 else: 

46 assert ndots > 1 

47 use_format_version = format_version[: format_version.rfind(".")] 

48 

49 descr_versions = descriptions_map[typ] 

50 if use_format_version not in descr_versions: 

51 if fallback_to_latest: 

52 issue_warning( 

53 "Unsupported format version '{value}' for type '{typ}'.", 

54 value=format_version, 

55 field="format_version", 

56 log_depth=3, 

57 msg_context={"typ": typ}, 

58 ) 

59 use_format_version = "latest" 

60 else: 

61 raise ValueError( 

62 f"Unsupported format version '{format_version}' for type '{typ}'." 

63 + " Supported format versions are: {', '.join(sorted(fv for fv in descr_versions))}" 

64 ) 

65 

66 return descr_versions[use_format_version] 

67 

68 

69def build_description_impl( 

70 content: BioimageioYamlContentView, 

71 /, 

72 *, 

73 context: Optional[ValidationContext] = None, 

74 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

75 get_rd_class: Callable[[Any, Any, bool], Type[ResourceDescrT]], 

76) -> Union[ResourceDescrT, InvalidDescr]: 

77 context = context or get_validation_context() 

78 errors: List[ErrorEntry] = [] 

79 if isinstance(content, collections.abc.Mapping): 

80 for minimum in ("type", "format_version"): 

81 if minimum not in content: 

82 errors.append( 

83 ErrorEntry( 

84 loc=(minimum,), msg=f"Missing field '{minimum}'", type="error" 

85 ) 

86 ) 

87 elif not isinstance(content[minimum], str): 

88 errors.append( 

89 ErrorEntry( 

90 loc=(minimum,), 

91 msg=f"Invalid type '{type(content[minimum])}'", 

92 type="error", 

93 ) 

94 ) 

95 else: 

96 errors.append( 

97 ErrorEntry( 

98 loc=(), msg=f"Invalid content of type '{type(content)}'", type="error" 

99 ) 

100 ) 

101 content = {} 

102 

103 if errors: 

104 ret = InvalidDescr(**content) # pyright: ignore[reportArgumentType] 

105 ret.validation_summary.add_detail( 

106 ValidationDetail( 

107 name="extract fields to chose description class", 

108 status="failed", 

109 errors=errors, 

110 context=context.summary, 

111 ) 

112 ) 

113 return ret 

114 

115 typ = content["type"] 

116 # check format_version argument before loading as 'discover' 

117 # to throw an exception for an invalid format_version early 

118 if str(format_version).lower() != DISCOVER: 

119 as_rd_class = get_rd_class(typ, format_version, False) 

120 else: 

121 as_rd_class = None 

122 # always load with discovered format_version first 

123 rd_class = get_rd_class(typ, content["format_version"], True) 

124 rd = rd_class.load(content, context=context) 

125 

126 if as_rd_class is not None and as_rd_class is not rd_class: 

127 # load with requested format_version 

128 discover_details = rd.validation_summary.details 

129 rd = as_rd_class.load(content, context=context) 

130 assert rd.validation_summary is not None 

131 rd.validation_summary.details[:0] = discover_details 

132 

133 return rd