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

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 results as _results 

28from . import state as _state 

29 

30 

31class Processor( __.immut.DataclassProtocol ): 

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

33 

34 name: str 

35 

36 @property 

37 @__.abc.abstractmethod 

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

39 ''' Returns processor capabilities for advertisement. ''' 

40 raise NotImplementedError 

41 

42 @__.abc.abstractmethod 

43 async def detect( 

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

45 ) -> 'Detection': 

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

47 raise NotImplementedError 

48 

49 

50InventoryProcessorsRegistry: __.typx.TypeAlias = ( 

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

52StructureProcessorsRegistry: __.typx.TypeAlias = ( 

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

54 

55 

56class Detection( __.immut.DataclassProtocol ): 

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

58 

59 processor: Processor 

60 confidence: float 

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

62 

63 @property 

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

65 ''' Returns capabilities for processor. ''' 

66 return self.processor.capabilities 

67 

68 def __post_init__( self ) -> None: 

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

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

71 raise _exceptions.DetectionConfidenceInvalidity( self.confidence ) 

72 

73 @classmethod 

74 @__.abc.abstractmethod 

75 async def from_source( 

76 selfclass, 

77 auxdata: _state.Globals, 

78 processor: Processor, 

79 source: str, 

80 ) -> __.typx.Self: 

81 ''' Constructs detection from source location. ''' 

82 raise NotImplementedError 

83 

84 

85class InventoryDetection( Detection ): 

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

87 

88 @__.abc.abstractmethod 

89 async def filter_inventory( 

90 self, 

91 auxdata: _state.Globals, 

92 source: str, /, *, 

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

94 details: _interfaces.InventoryQueryDetails = ( 

95 _interfaces.InventoryQueryDetails.Documentation ), 

96 ) -> tuple[ _results.InventoryObject, ... ]: 

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

98 raise NotImplementedError 

99 

100 

101class StructureDetection( Detection ): 

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

103 

104 @__.abc.abstractmethod 

105 async def extract_contents( 

106 self, 

107 auxdata: _state.Globals, 

108 source: str, 

109 objects: __.cabc.Sequence[ _results.InventoryObject ], /, 

110 ) -> tuple[ _results.ContentDocument, ... ]: 

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 ) )