Coverage for sources/librovore/results.py: 30%
300 statements
« prev ^ index » next coverage.py v7.10.5, created at 2025-08-29 01:14 +0000
« prev ^ index » next coverage.py v7.10.5, created at 2025-08-29 01:14 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
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#============================================================================#
21''' Results structures.
23 Search results, inventory objects, content documents, etc....
24'''
27from . import __
30_CONTENT_PREVIEW_LIMIT = 100
33class InventoryObject( __.immut.DataclassObject ):
34 ''' Universal inventory object with complete source attribution.
36 Represents a single documentation object from any inventory source
37 with standardized fields and format-specific metadata container.
38 '''
40 name: __.typx.Annotated[
41 str,
42 __.ddoc.Doc( "Primary object identifier from inventory source." ),
43 ]
44 uri: __.typx.Annotated[
45 str,
46 __.ddoc.Doc( "Relative URI to object documentation content." ),
47 ]
48 inventory_type: __.typx.Annotated[
49 str,
50 __.ddoc.Doc(
51 "Inventory format identifier (e.g., sphinx_objects_inv)." ),
52 ]
53 location_url: __.typx.Annotated[
54 str, __.ddoc.Doc(
55 "Complete URL to inventory location for attribution." )
56 ]
57 display_name: __.typx.Annotated[
58 __.typx.Optional[ str ],
59 __.ddoc.Doc( "Human-readable name if different from name." ),
60 ] = None
61 specifics: __.typx.Annotated[
62 __.immut.Dictionary[ str, __.typx.Any ],
63 __.ddoc.Doc(
64 "Format-specific metadata (domain, role, priority, etc.)." ),
65 ] = __.dcls.field( default_factory = lambda: __.immut.Dictionary( ) )
68 @property
69 def effective_display_name( self ) -> str:
70 ''' Effective display name. Might be same as name. '''
71 if self.display_name is not None:
72 return self.display_name
73 return self.name
75 @__.abc.abstractmethod
76 def render_specifics_json(
77 self
78 ) -> __.immut.Dictionary[ str, __.typx.Any ]:
79 ''' Renders specifics for JSON output. '''
80 raise NotImplementedError
82 @__.abc.abstractmethod
83 def render_specifics_markdown(
84 self, /, *,
85 reveal_internals: __.typx.Annotated[
86 bool,
87 __.ddoc.Doc( '''
88 Controls whether implementation-specific details (internal
89 field names, version numbers, priority scores) are included.
90 When False, only user-facing information is shown.
91 ''' ),
92 ] = True,
93 ) -> tuple[ str, ... ]:
94 ''' Renders specifics as Markdown lines for CLI display. '''
95 raise NotImplementedError
97 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
98 ''' Renders complete object as JSON-compatible dictionary. '''
99 base = __.immut.Dictionary[
100 str, __.typx.Any
101 ](
102 name = self.name,
103 uri = self.uri,
104 inventory_type = self.inventory_type,
105 location_url = self.location_url,
106 display_name = self.display_name,
107 effective_display_name = self.effective_display_name,
108 )
109 formatted_specifics = self.render_specifics_json( )
110 result_dict = dict( base )
111 result_dict.update( dict( formatted_specifics ) )
112 return __.immut.Dictionary[ str, __.typx.Any ]( result_dict )
114 def render_as_markdown(
115 self, /, *,
116 reveal_internals: __.typx.Annotated[
117 bool,
118 __.ddoc.Doc( "Controls whether internal details are shown." ),
119 ] = True,
120 ) -> tuple[ str, ... ]:
121 ''' Renders complete object as Markdown lines for display. '''
122 lines = [ f"### `{self.effective_display_name}`" ]
123 if reveal_internals:
124 lines.append( f"**URI:** {self.uri}" )
125 lines.append( f"**Type:** {self.inventory_type}" )
126 lines.append( f"**Location:** {self.location_url}" )
127 specifics_lines = self.render_specifics_markdown(
128 reveal_internals = reveal_internals )
129 lines.extend( specifics_lines )
130 return tuple( lines )
133class ContentDocument( __.immut.DataclassObject ):
134 ''' Documentation content with extracted metadata and snippets. '''
136 inventory_object: __.typx.Annotated[
137 InventoryObject,
138 __.ddoc.Doc( "Location inventory object for this content." ),
139 ]
140 signature: __.typx.Annotated[
141 str,
142 __.ddoc.Doc( "Extracted function/class signature." ),
143 ] = ''
144 description: __.typx.Annotated[
145 str,
146 __.ddoc.Doc( "Extracted object description or summary." ),
147 ] = ''
148 content_snippet: __.typx.Annotated[
149 str,
150 __.ddoc.Doc( "Relevant content excerpt for search context." ),
151 ] = ''
152 documentation_url: __.typx.Annotated[
153 str,
154 __.ddoc.Doc( "Complete URL to full documentation page." ),
155 ] = ''
156 extraction_metadata: __.typx.Annotated[
157 __.immut.Dictionary[ str, __.typx.Any ],
158 __.ddoc.Doc( "Metadata from structure processor extraction." ),
159 ] = __.dcls.field( default_factory = lambda: __.immut.Dictionary( ) )
161 @property
162 def has_meaningful_content( self ) -> bool:
163 ''' Returns True if document contains useful extracted content. '''
164 return bool(
165 self.signature or self.description or self.content_snippet )
167 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
168 ''' Renders complete document as JSON-compatible dictionary. '''
169 return __.immut.Dictionary[
170 str, __.typx.Any
171 ](
172 inventory_object = dict( self.inventory_object.render_as_json( ) ),
173 signature = self.signature,
174 description = self.description,
175 content_snippet = self.content_snippet,
176 documentation_url = self.documentation_url,
177 extraction_metadata = dict( self.extraction_metadata ),
178 has_meaningful_content = self.has_meaningful_content,
179 )
181 def render_as_markdown(
182 self, /, *,
183 reveal_internals: __.typx.Annotated[
184 bool,
185 __.ddoc.Doc( "Controls whether internal details are shown." ),
186 ] = True,
187 ) -> tuple[ str, ... ]:
188 ''' Renders complete document as Markdown lines for display. '''
189 lines = [ f"### `{self.inventory_object.effective_display_name}`" ]
190 if self.signature:
191 lines.append( f"**Signature:** `{self.signature}`" )
192 if self.description:
193 lines.append( f"**Description:** {self.description}" )
194 if self.content_snippet:
195 lines.append( f"**Content:** {self.content_snippet}" )
196 if self.documentation_url:
197 lines.append( f"**URL:** {self.documentation_url}" )
198 if reveal_internals:
199 inventory_lines = self.inventory_object.render_specifics_markdown(
200 reveal_internals = True )
201 lines.extend( inventory_lines )
202 return tuple( lines )
205class ErrorInfo( __.immut.DataclassObject ):
206 ''' Structured error information for processor failures. '''
208 type: __.typx.Annotated[
209 str,
210 __.ddoc.Doc(
211 "Error type identifier (e.g., 'processor_unavailable')." ),
212 ]
213 title: __.typx.Annotated[
214 str,
215 __.ddoc.Doc( "Human-readable error title." ),
216 ]
217 message: __.typx.Annotated[
218 str,
219 __.ddoc.Doc( "Detailed error description." ),
220 ]
221 suggestion: __.typx.Annotated[
222 __.typx.Optional[ str ],
223 __.ddoc.Doc( "Suggested remediation steps." ),
224 ] = None
226 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
227 ''' Renders error info as JSON-compatible dictionary. '''
228 return __.immut.Dictionary[
229 str, __.typx.Any
230 ](
231 type = self.type,
232 title = self.title,
233 message = self.message,
234 suggestion = self.suggestion,
235 )
238class ErrorResponse( __.immut.DataclassObject ):
239 ''' Error response wrapper maintaining query context. '''
241 location: __.typx.Annotated[
242 str,
243 __.ddoc.Doc( "Primary location URL for failed query." ),
244 ]
245 query: __.typx.Annotated[
246 str,
247 __.ddoc.Doc( "Search term or query string that failed." ),
248 ]
249 error: __.typx.Annotated[
250 ErrorInfo,
251 __.ddoc.Doc( "Detailed error information." ),
252 ]
254 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
255 ''' Renders error response as JSON-compatible dictionary. '''
256 return __.immut.Dictionary(
257 location = self.location,
258 query = self.query,
259 error = __.immut.Dictionary[
260 str, __.typx.Any
261 ](
262 type = self.error.type,
263 title = self.error.title,
264 message = self.error.message,
265 suggestion = self.error.suggestion,
266 ),
267 )
269 def render_as_markdown(
270 self, /, *,
271 reveal_internals: __.typx.Annotated[
272 bool,
273 __.ddoc.Doc( "Controls whether internal details are shown." ),
274 ] = True,
275 ) -> tuple[ str, ... ]:
276 ''' Renders error response as Markdown lines for display. '''
277 lines = [ f"## Error: {self.error.title}" ]
278 lines.append( f"**Message:** {self.error.message}" )
279 if self.error.suggestion:
280 lines.append( f"**Suggestion:** {self.error.suggestion}" )
281 if reveal_internals:
282 lines.append( f"**Location:** {self.location}" )
283 lines.append( f"**Query:** {self.query}" )
284 lines.append( f"**Error Type:** {self.error.type}" )
285 return tuple( lines )
288class InventoryLocationInfo( __.immut.DataclassObject ):
289 ''' Information about detected inventory location and processor. '''
291 inventory_type: __.typx.Annotated[
292 str,
293 __.ddoc.Doc( "Inventory format type identifier." ),
294 ]
295 location_url: __.typx.Annotated[
296 str,
297 __.ddoc.Doc( "Complete URL to inventory location." ),
298 ]
299 processor_name: __.typx.Annotated[
300 str,
301 __.ddoc.Doc( "Name of processor handling this location." ),
302 ]
303 confidence: __.typx.Annotated[
304 float,
305 __.ddoc.Doc( "Detection confidence score (0.0-1.0)." ),
306 ]
307 object_count: __.typx.Annotated[
308 int,
309 __.ddoc.Doc( "Total objects available in this inventory." ),
310 ]
312 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
313 ''' Renders location info as JSON-compatible dictionary. '''
314 return __.immut.Dictionary(
315 inventory_type = self.inventory_type,
316 location_url = self.location_url,
317 processor_name = self.processor_name,
318 confidence = self.confidence,
319 object_count = self.object_count,
320 )
323class SearchMetadata( __.immut.DataclassObject ):
324 ''' Search operation metadata and performance statistics. '''
326 results_count: __.typx.Annotated[
327 int,
328 __.ddoc.Doc( "Number of results returned to user." ),
329 ]
330 results_max: __.typx.Annotated[
331 int,
332 __.ddoc.Doc( "Maximum results requested by user." ),
333 ]
334 matches_total: __.typx.Annotated[
335 __.typx.Optional[ int ],
336 __.ddoc.Doc( "Total matching objects before limit applied." ),
337 ] = None
338 search_time_ms: __.typx.Annotated[
339 __.typx.Optional[ int ],
340 __.ddoc.Doc( "Search execution time in milliseconds." ),
341 ] = None
343 @property
344 def results_truncated( self ) -> bool:
345 ''' Returns True if results were limited by results_max. '''
346 if self.matches_total is None:
347 return False
348 return self.results_count < self.matches_total
350 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
351 ''' Renders search metadata as JSON-compatible dictionary. '''
352 return __.immut.Dictionary(
353 results_count = self.results_count,
354 results_max = self.results_max,
355 matches_total = self.matches_total,
356 search_time_ms = self.search_time_ms,
357 results_truncated = self.results_truncated,
358 )
361class SearchResult( __.immut.DataclassObject ):
362 ''' Search result with inventory object and match metadata. '''
364 inventory_object: __.typx.Annotated[
365 InventoryObject,
366 __.ddoc.Doc( "Matched inventory object with metadata." ),
367 ]
368 score: __.typx.Annotated[
369 float,
370 __.ddoc.Doc( "Search relevance score (0.0-1.0)." ),
371 ]
372 match_reasons: __.typx.Annotated[
373 tuple[ str, ... ],
374 __.ddoc.Doc( "Detailed reasons for search match." ),
375 ]
377 @classmethod
378 def from_inventory_object(
379 cls,
380 inventory_object: InventoryObject, *,
381 score: float,
382 match_reasons: __.cabc.Sequence[ str ],
383 ) -> __.typx.Self:
384 ''' Produces search result from inventory object with scoring. '''
385 return cls(
386 inventory_object = inventory_object,
387 score = score,
388 match_reasons = tuple( match_reasons ) )
390 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
391 ''' Renders search result as JSON-compatible dictionary. '''
392 return __.immut.Dictionary[
393 str, __.typx.Any
394 ](
395 inventory_object = dict( self.inventory_object.render_as_json( ) ),
396 score = self.score,
397 match_reasons = list( self.match_reasons ),
398 )
400 def render_as_markdown(
401 self, /, *,
402 reveal_internals: __.typx.Annotated[
403 bool,
404 __.ddoc.Doc( "Controls whether internal details are shown." ),
405 ] = True,
406 ) -> tuple[ str, ... ]:
407 ''' Renders search result as Markdown lines for display. '''
408 title = "### `{name}` (Score: {score:.2f})".format(
409 name = self.inventory_object.effective_display_name,
410 score = self.score )
411 lines = [ title ]
412 if reveal_internals and self.match_reasons:
413 reasons = ', '.join( self.match_reasons )
414 lines.append( "**Match reasons:** {reasons}".format(
415 reasons = reasons ) )
416 inventory_lines = self.inventory_object.render_as_markdown(
417 reveal_internals = reveal_internals )
418 lines.extend( inventory_lines[ 1: ] ) # Skip duplicate title line
419 return tuple( lines )
422class ContentQueryResult( __.immut.DataclassObject ):
423 ''' Complete result structure for content queries. '''
425 location: __.typx.Annotated[
426 str,
427 __.ddoc.Doc( "Primary location URL for this query." ),
428 ]
429 query: __.typx.Annotated[
430 str,
431 __.ddoc.Doc( "Search term or query string used." ),
432 ]
433 documents: __.typx.Annotated[
434 tuple[ ContentDocument, ... ],
435 __.ddoc.Doc( "Documentation content for matching objects." ) ]
436 search_metadata: __.typx.Annotated[
437 SearchMetadata,
438 __.ddoc.Doc( "Search execution and result metadata." ),
439 ]
440 inventory_locations: __.typx.Annotated[
441 tuple[ InventoryLocationInfo, ... ],
442 __.ddoc.Doc( "Information about inventory locations used." ),
443 ]
445 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
446 ''' Renders content query result as JSON-compatible dictionary. '''
447 documents_json = [
448 dict( doc.render_as_json( ) ) for doc in self.documents ]
449 locations_json = [
450 dict( loc.render_as_json( ) ) for loc in self.inventory_locations ]
451 return __.immut.Dictionary[
452 str, __.typx.Any
453 ](
454 location = self.location,
455 query = self.query,
456 documents = documents_json,
457 search_metadata = dict( self.search_metadata.render_as_json( ) ),
458 inventory_locations = locations_json,
459 )
461 def render_as_markdown(
462 self, /, *,
463 reveal_internals: __.typx.Annotated[
464 bool,
465 __.ddoc.Doc( "Controls whether internal details are shown." ),
466 ] = True,
467 lines_max: __.typx.Annotated[
468 __.typx.Optional[ int ],
469 __.ddoc.Doc( "Maximum lines to display per content result." ),
470 ] = None,
471 ) -> tuple[ str, ... ]:
472 ''' Renders content query result as Markdown lines for display. '''
473 title = "# Content Query Results"
474 if lines_max is not None:
475 title += " (truncated)"
476 lines = [ title ]
477 lines.append( "**Query:** {query}".format( query = self.query ) )
478 if reveal_internals:
479 lines.append( "**Location:** {location}".format(
480 location = self.location ) )
481 lines.append( "**Results:** {count} of {max}".format(
482 count = self.search_metadata.results_count,
483 max = self.search_metadata.results_max ) )
484 if self.documents:
485 lines.append( "" )
486 lines.append( "## Documents" )
487 for index, doc in enumerate( self.documents, 1 ):
488 separator = "\n📄 ── Document {} ──────────────────── 📄\n"
489 lines.append( separator.format( index ) )
490 lines.append( "**URL:** {url}".format(
491 url = doc.documentation_url ) )
492 if doc.signature:
493 lines.append( "**Signature:** {signature}".format(
494 signature = doc.signature ) )
495 if doc.description:
496 lines.append( "**Description:** {description}".format(
497 description = doc.description ) )
498 if doc.content_snippet:
499 content = doc.content_snippet
500 if lines_max is not None:
501 content_lines = content.split( '\n' )
502 if len( content_lines ) > lines_max:
503 content_lines = content_lines[ :lines_max ]
504 content_lines.append(
505 "... (truncated at {lines_max} lines)".format(
506 lines_max = lines_max ) )
507 content = '\n'.join( content_lines )
508 lines.append( "**Content:** {content}".format(
509 content = content ) )
510 return tuple( lines )
513class InventoryQueryResult( __.immut.DataclassObject ):
514 ''' Complete result structure for inventory queries. '''
516 location: __.typx.Annotated[
517 str,
518 __.ddoc.Doc( "Primary location URL for this query." ),
519 ]
520 query: __.typx.Annotated[
521 str,
522 __.ddoc.Doc( "Search term or query string used." ),
523 ]
524 objects: __.typx.Annotated[
525 tuple[ InventoryObject, ... ],
526 __.ddoc.Doc( "Inventory objects matching search criteria." ),
527 ]
528 search_metadata: __.typx.Annotated[
529 SearchMetadata,
530 __.ddoc.Doc( "Search execution and result metadata." ),
531 ]
532 inventory_locations: __.typx.Annotated[
533 tuple[ InventoryLocationInfo, ... ],
534 __.ddoc.Doc( "Information about inventory locations used." ),
535 ]
537 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
538 ''' Renders inventory query result as JSON-compatible dictionary. '''
539 objects_json = [
540 dict( obj.render_as_json( ) ) for obj in self.objects ]
541 locations_json = [
542 dict( loc.render_as_json( ) ) for loc in self.inventory_locations ]
543 return __.immut.Dictionary[
544 str, __.typx.Any
545 ](
546 location = self.location,
547 query = self.query,
548 objects = objects_json,
549 search_metadata = dict( self.search_metadata.render_as_json( ) ),
550 inventory_locations = locations_json,
551 )
553 def render_as_markdown(
554 self, /, *,
555 reveal_internals: __.typx.Annotated[
556 bool,
557 __.ddoc.Doc( "Controls whether internal details are shown." ),
558 ] = True,
559 ) -> tuple[ str, ... ]:
560 ''' Renders inventory query result as Markdown lines for display. '''
561 lines = [ "# Inventory Query Results" ]
562 lines.append( "**Query:** {query}".format( query = self.query ) )
563 if reveal_internals:
564 lines.append( "**Location:** {location}".format(
565 location = self.location ) )
566 lines.append( "**Results:** {count} of {max}".format(
567 count = self.search_metadata.results_count,
568 max = self.search_metadata.results_max ) )
569 if self.objects:
570 lines.append( "" )
571 lines.append( "## Objects" )
572 for index, obj in enumerate( self.objects, 1 ):
573 separator = "\n📦 ── Object {} ─────────────────────── 📦\n"
574 lines.append( separator.format( index ) )
575 obj_lines = obj.render_as_markdown(
576 reveal_internals = reveal_internals )
577 lines.extend( obj_lines )
578 return tuple( lines )
581class Detection( __.immut.DataclassObject ):
582 ''' Processor detection information with confidence scoring. '''
584 processor_name: __.typx.Annotated[
585 str,
586 __.ddoc.Doc( "Name of the processor that can handle this location." ),
587 ]
588 confidence: __.typx.Annotated[
589 float,
590 __.ddoc.Doc( "Detection confidence score (0.0-1.0)." ),
591 ]
592 processor_type: __.typx.Annotated[
593 str,
594 __.ddoc.Doc( "Type of processor (inventory, structure)." ),
595 ]
596 detection_metadata: __.typx.Annotated[
597 __.immut.Dictionary[ str, __.typx.Any ],
598 __.ddoc.Doc( "Processor-specific detection metadata." ),
599 ] = __.dcls.field( default_factory = lambda: __.immut.Dictionary( ) )
601 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
602 ''' Renders detection as JSON-compatible dictionary. '''
603 return __.immut.Dictionary[
604 str, __.typx.Any
605 ](
606 processor_name = self.processor_name,
607 confidence = self.confidence,
608 processor_type = self.processor_type,
609 detection_metadata = dict( self.detection_metadata ),
610 )
613class DetectionsResult( __.immut.DataclassObject ):
614 ''' Detection results with processor selection and timing metadata. '''
616 source: __.typx.Annotated[
617 str,
618 __.ddoc.Doc( "Primary location URL for detection operation." ),
619 ]
620 detections: __.typx.Annotated[
621 tuple[ Detection, ... ],
622 __.ddoc.Doc( "All processor detections found for location." ),
623 ]
624 detection_optimal: __.typx.Annotated[
625 __.typx.Optional[ Detection ],
626 __.ddoc.Doc( "Best detection result based on confidence scoring." ),
627 ]
628 time_detection_ms: __.typx.Annotated[
629 int,
630 __.ddoc.Doc( "Detection operation time in milliseconds." ),
631 ]
634 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
635 ''' Renders detection results as JSON-compatible dictionary. '''
636 detections_json = [
637 dict( detection.render_as_json( ) )
638 for detection in self.detections ]
639 return __.immut.Dictionary[
640 str, __.typx.Any
641 ](
642 source = self.source,
643 detections = detections_json,
644 detection_optimal = (
645 dict( self.detection_optimal.render_as_json( ) )
646 if self.detection_optimal else None ),
647 time_detection_ms = self.time_detection_ms,
648 )
650 def render_as_markdown(
651 self, /, *,
652 reveal_internals: __.typx.Annotated[
653 bool,
654 __.ddoc.Doc( "Controls whether internal details are shown." ),
655 ] = True,
656 ) -> tuple[ str, ... ]:
657 ''' Renders detection results as Markdown lines for display. '''
658 lines = [ "# Detection Results" ]
659 if reveal_internals:
660 lines.append( "**Source:** {source}".format(
661 source = self.source ) )
662 lines.append( "**Detection time:** {time}ms".format(
663 time = self.time_detection_ms ) )
664 if self.detection_optimal:
665 lines.append( "**Optimal processor:** {name} ({type})".format(
666 name = self.detection_optimal.processor_name,
667 type = self.detection_optimal.processor_type ) )
668 lines.append( "**Confidence:** {conf:.2f}".format(
669 conf = self.detection_optimal.confidence ) )
670 else:
671 lines.append( "**No optimal processor found**" )
672 if reveal_internals and self.detections:
673 lines.append( "" )
674 lines.append( "## All Detections" )
675 detection_lines = [
676 "- **{name}** ({type}): {conf:.2f}".format(
677 name = detection.processor_name,
678 type = detection.processor_type,
679 conf = detection.confidence )
680 for detection in self.detections ]
681 lines.extend( detection_lines )
682 return tuple( lines )
685class ProcessorInfo( __.immut.DataclassObject ):
686 ''' Information about a processor and its capabilities. '''
688 processor_name: __.typx.Annotated[
689 str,
690 __.ddoc.Doc( "Name of the processor for identification." ),
691 ]
692 processor_type: __.typx.Annotated[
693 str,
694 __.ddoc.Doc( "Type of processor (inventory, structure)." ),
695 ]
696 capabilities: __.typx.Annotated[
697 __.typx.Any, # Will be _interfaces.ProcessorCapabilities after import
698 __.ddoc.Doc( "Complete capability description for processor." ),
699 ]
701 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
702 ''' Renders processor info as JSON-compatible dictionary. '''
703 return __.immut.Dictionary[
704 str, __.typx.Any
705 ](
706 processor_name = self.processor_name,
707 processor_type = self.processor_type,
708 capabilities = serialize_for_json( self.capabilities ),
709 )
711 def render_as_markdown(
712 self, /, *,
713 reveal_internals: __.typx.Annotated[
714 bool,
715 __.ddoc.Doc( "Controls whether internal details are shown." ),
716 ] = True,
717 ) -> tuple[ str, ... ]:
718 ''' Renders processor info as Markdown lines for display. '''
719 lines = [ f"### `{self.processor_name}` ({self.processor_type})" ]
720 if reveal_internals:
721 lines.append( f"**Version:** {self.capabilities.version}" )
722 if self.capabilities.results_limit_max:
723 lines.append(
724 f"**Max results:** {self.capabilities.results_limit_max}" )
725 if self.capabilities.response_time_typical:
726 lines.append(
727 f"**Response time:** "
728 f"{self.capabilities.response_time_typical}" )
729 if self.capabilities.notes:
730 lines.append( f"**Notes:** {self.capabilities.notes}" )
731 if self.capabilities.supported_filters:
732 lines.append( "" )
733 lines.append( "**Supported filters:**" )
734 for filter_cap in self.capabilities.supported_filters:
735 filter_line = f"- `{filter_cap.name}` ({filter_cap.type})"
736 if filter_cap.required:
737 filter_line += " *required*"
738 lines.append( filter_line )
739 if filter_cap.description:
740 lines.append( f" {filter_cap.description}" )
741 if filter_cap.values:
742 values_str = ', '.join( filter_cap.values )
743 lines.append( f" Values: {values_str}" )
744 return tuple( lines )
747class ProcessorsSurveyResult( __.immut.DataclassObject ):
748 ''' Survey results listing available processors and capabilities. '''
750 genus: __.typx.Annotated[
751 __.typx.Any, # Will be _interfaces.ProcessorGenera after import
752 __.ddoc.Doc(
753 "Processor genus that was surveyed (inventory or structure)." ),
754 ]
755 filter_name: __.typx.Annotated[
756 __.typx.Optional[ str ],
757 __.ddoc.Doc( "Optional processor name filter applied to survey." ),
758 ] = None
759 processors: __.typx.Annotated[
760 tuple[ ProcessorInfo, ... ],
761 __.ddoc.Doc( "Available processors matching survey criteria." ),
762 ]
763 survey_time_ms: __.typx.Annotated[
764 int,
765 __.ddoc.Doc( "Survey operation time in milliseconds." ),
766 ]
768 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
769 ''' Renders survey results as JSON-compatible dictionary. '''
770 processors_json = [
771 dict( processor.render_as_json( ) )
772 for processor in self.processors ]
773 return __.immut.Dictionary[
774 str, __.typx.Any
775 ](
776 genus = (
777 self.genus.value if hasattr( self.genus, 'value' )
778 else str( self.genus ) ),
779 filter_name = self.filter_name,
780 processors = processors_json,
781 survey_time_ms = self.survey_time_ms,
782 )
784 def render_as_markdown(
785 self, /, *,
786 reveal_internals: __.typx.Annotated[
787 bool,
788 __.ddoc.Doc( "Controls whether internal details are shown." ),
789 ] = True,
790 ) -> tuple[ str, ... ]:
791 ''' Renders survey results as Markdown lines for display. '''
792 genus_name = (
793 self.genus.value if hasattr( self.genus, 'value' )
794 else str( self.genus ) )
795 title = f"# Processor Survey Results ({genus_name})"
796 lines = [ title ]
797 if reveal_internals:
798 lines.append( f"**Survey time:** {self.survey_time_ms}ms" )
799 if self.filter_name:
800 lines.append( f"**Filter:** {self.filter_name}" )
801 lines.append( f"**Processors found:** {len( self.processors )}" )
802 if self.processors:
803 lines.append( "" )
804 for i, processor in enumerate( self.processors, 1 ):
805 lines.append( f"📦 ── Processor {i} ──────────" )
806 processor_lines = processor.render_as_markdown(
807 reveal_internals = reveal_internals )
808 lines.extend( processor_lines )
809 if i < len( self.processors ):
810 lines.append( "" )
811 return tuple( lines )
814def serialize_for_json( objct: __.typx.Any ) -> __.typx.Any:
815 ''' Legacy serialization for non-structured objects (dicts). '''
816 if __.dcls.is_dataclass( objct ):
817 return _serialize_dataclass_for_json( objct )
818 if isinstance( objct, ( list, tuple, set, frozenset ) ):
819 return _serialize_sequence_for_json( objct )
820 if isinstance( objct, ( dict, __.immut.Dictionary ) ):
821 return _serialize_mapping_for_json( objct )
822 return objct
826def _serialize_dataclass_for_json(
827 obj: __.typx.Any,
828) -> dict[ str, __.typx.Any ]:
829 ''' Serializes dataclass objects to JSON-compatible dictionaries. '''
830 result: dict[ str, __.typx.Any ] = { }
831 for field in __.dcls.fields( obj ):
832 if field.name.startswith( '_' ):
833 continue
834 value = getattr( obj, field.name )
835 result[ field.name ] = serialize_for_json( value )
836 return result
839def _serialize_mapping_for_json(
840 obj: __.typx.Any
841) -> dict[ __.typx.Any, __.typx.Any ]:
842 ''' Serializes mapping-like objects to JSON-compatible dictionaries. '''
843 return {
844 key: serialize_for_json( value )
845 for key, value in obj.items( )
846 }
849def _serialize_sequence_for_json( obj: __.typx.Any ) -> list[ __.typx.Any ]:
850 ''' Serializes sequence-like objects to JSON-compatible lists. '''
851 return [ serialize_for_json( item ) for item in obj ]
854ContentDocuments: __.typx.TypeAlias = __.cabc.Sequence[ ContentDocument ]
855InventoryObjects: __.typx.TypeAlias = __.cabc.Sequence[ InventoryObject ]
856SearchResults: __.typx.TypeAlias = __.cabc.Sequence[ SearchResult ]
858ContentResult: __.typx.TypeAlias = ContentQueryResult | ErrorResponse
859InventoryResult: __.typx.TypeAlias = InventoryQueryResult | ErrorResponse
860DetectionsResultUnion: __.typx.TypeAlias = DetectionsResult | ErrorResponse
861ProcessorsSurveyResultUnion: __.typx.TypeAlias = (
862 ProcessorsSurveyResult | ErrorResponse )