Coverage for sources/librovore/processors.py: 80%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' Site processors. ''' 

22 

23 

24from . import __ 

25from . import exceptions as _exceptions 

26from . import interfaces as _interfaces 

27from . import state as _state 

28 

29 

30class Processor( __.immut.DataclassProtocol ): 

31 ''' Abstract base class for documentation source detectors. ''' 

32 

33 name: str 

34 

35 @property 

36 @__.abc.abstractmethod 

37 def capabilities( self ) -> _interfaces.ProcessorCapabilities: 

38 ''' Returns processor capabilities for advertisement. ''' 

39 raise NotImplementedError 

40 

41 @__.abc.abstractmethod 

42 async def detect( 

43 self, auxdata: _state.Globals, source: str 

44 ) -> 'Detection': 

45 ''' Detects if can process documentation from source. ''' 

46 raise NotImplementedError 

47 

48 

49InventoryProcessorsRegistry: __.typx.TypeAlias = ( 

50 __.accret.ValidatorDictionary[ str, Processor ] ) 

51StructureProcessorsRegistry: __.typx.TypeAlias = ( 

52 __.accret.ValidatorDictionary[ str, Processor ] ) 

53 

54 

55class Detection( __.immut.DataclassProtocol ): 

56 ''' Abstract base class for documentation detector selections. ''' 

57 

58 processor: Processor 

59 confidence: float 

60 timestamp: float = __.dcls.field( default_factory = __.time.time ) 

61 

62 @property 

63 def capabilities( self ) -> _interfaces.ProcessorCapabilities: 

64 ''' Returns capabilities for processor. ''' 

65 return self.processor.capabilities 

66 

67 def __post_init__( self ) -> None: 

68 ''' Validates confidence is in valid range [0.0, 1.0]. ''' 

69 if not ( 0.0 <= self.confidence <= 1.0 ): 

70 raise _exceptions.DetectionConfidenceInvalidity( self.confidence ) 

71 

72 @classmethod 

73 @__.abc.abstractmethod 

74 async def from_source( 

75 selfclass, 

76 auxdata: _state.Globals, 

77 processor: Processor, 

78 source: str, 

79 ) -> __.typx.Self: 

80 ''' Constructs detection from source location. ''' 

81 raise NotImplementedError 

82 

83 

84class InventoryDetection( Detection ): 

85 ''' Base class for inventory detection results. ''' 

86 

87 @__.abc.abstractmethod 

88 async def filter_inventory( 

89 self, 

90 auxdata: _state.Globals, 

91 source: str, /, *, 

92 filters: __.cabc.Mapping[ str, __.typx.Any ], 

93 details: _interfaces.InventoryQueryDetails = ( 

94 _interfaces.InventoryQueryDetails.Documentation ), 

95 ) -> list[ dict[ str, __.typx.Any ] ]: 

96 ''' Extracts and filters inventory objects from source. ''' 

97 raise NotImplementedError 

98 

99 

100class StructureDetection( Detection ): 

101 ''' Base class for structure detection results. ''' 

102 

103 @__.abc.abstractmethod 

104 async def extract_contents( 

105 self, 

106 auxdata: _state.Globals, 

107 source: str, 

108 objects: __.cabc.Sequence[ __.cabc.Mapping[ str, __.typx.Any ] ], /, *, 

109 include_snippets: bool = True, 

110 ) -> list[ dict[ str, __.typx.Any ] ]: 

111 ''' Extracts documentation content for specified objects. ''' 

112 raise NotImplementedError 

113 

114 

115DetectionsByProcessor: __.typx.TypeAlias = __.cabc.Mapping[ str, Detection ] 

116 

117 

118class DetectionsForLocation( __.immut.DataclassObject ): 

119 ''' Detections for location. ''' 

120 

121 source: str 

122 detections: DetectionsByProcessor 

123 detection_optimal: __.typx.Optional[ Detection ] 

124 time_detection_ms: int 

125 

126 

127def _inventory_validator( name: str, value: Processor ) -> bool: 

128 return isinstance( value, Processor ) 

129 

130def _structure_validator( name: str, value: Processor ) -> bool: 

131 return isinstance( value, Processor ) 

132 

133 

134inventory_processors: InventoryProcessorsRegistry = ( 

135 __.accret.ValidatorDictionary( _inventory_validator ) ) 

136structure_processors: StructureProcessorsRegistry = ( 

137 __.accret.ValidatorDictionary( _structure_validator ) )