Coverage for sources/ictr/dispatchers.py: 94%
239 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-26 12:43 +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''' Dispatchers for messages and inspections to scribes. '''
24from . import __
25from . import configuration as _cfg
26from . import exceptions as _exceptions
27from . import flavors as _flavors
28from . import printers as _printers
29from . import reporters as _reporters
30from . import textualizers as _texts
33_installer_mutex: __.threads.Lock = __.threads.Lock( )
34_registrar_mutex: __.threads.Lock = __.threads.Lock( )
35_self_addresscfg: _cfg.AddressConfiguration = _cfg.AddressConfiguration(
36 flavors = __.immut.Dictionary(
37 note = _cfg.FlavorConfiguration(
38 compositor_factory = (
39 _texts.produce_compositor_factory_default(
40 introducer = 'NOTE| ' ) ) ),
41 error = _cfg.FlavorConfiguration(
42 compositor_factory = (
43 _texts.produce_compositor_factory_default(
44 introducer = 'ERROR| ' ) ) ) ) )
45_validate_arguments = (
46 __.validate_arguments(
47 globalvars = globals( ),
48 errorclass = _exceptions.ArgumentClassInvalidity ) )
51class AddressesConfigurationsRegistry(
52 __.accret.Dictionary[ str, _cfg.AddressConfiguration ]
53):
54 # TODO: Use 'accret.ValidatorDictionary'.
55 ''' Accretive dictionary specifically for address registrations. '''
57 def __init__(
58 self,
59 *iterables: __.DictionaryPositionalArgument[
60 str, _cfg.AddressConfiguration ],
61 **entries: __.DictionaryNominativeArgument[
62 _cfg.AddressConfiguration ],
63 ):
64 super( ).__init__( { __.package_name: _self_addresscfg } )
65 self.update( *iterables, **entries )
68class Omniflavor( __.enum.Enum ):
69 ''' Singleton to match any flavor. '''
71 Instance = __.enum.auto( )
74ActiveFlavors: __.typx.TypeAlias = Omniflavor | frozenset[ _flavors.Flavor ]
75ActiveFlavorsLiberal: __.typx.TypeAlias = __.typx.Union[
76 Omniflavor,
77 __.cabc.Sequence[ _flavors.Flavor ],
78 __.cabc.Set[ _flavors.Flavor ],
79]
80ActiveFlavorsRegistry: __.typx.TypeAlias = (
81 __.immut.Dictionary[ __.typx.Optional[ str ], ActiveFlavors ] )
82ActiveFlavorsRegistryLiberal: __.typx.TypeAlias = (
83 __.cabc.Mapping[ __.typx.Optional[ str ], ActiveFlavorsLiberal ] )
84AddressesConfigurationsRegistryLiberal: __.typx.TypeAlias = (
85 __.cabc.Mapping[ str, _cfg.AddressConfiguration ] )
86ReportersRegistry: __.typx.TypeAlias = (
87 __.accret.Dictionary[
88 tuple[ str, _flavors.Flavor ], _reporters.Reporter ] )
89TraceLevelsRegistry: __.typx.TypeAlias = (
90 __.immut.Dictionary[ __.typx.Optional[ str ], int ] )
91TraceLevelsRegistryLiberal: __.typx.TypeAlias = (
92 __.cabc.Mapping[ __.typx.Optional[ str ], int ] )
95def _provide_active_flavors_default( ) -> ActiveFlavorsRegistry:
96 ''' Provides default set of globally active flavors. '''
97 flavors = set( _flavors.flavor_specifications_standard.keys( ) )
98 flavors.update( _flavors.flavor_aliases_standard.keys( ) )
99 return __.immut.Dictionary( { None: frozenset( flavors ) } )
102builtins_alias_default: __.typx.Annotated[
103 str,
104 __.typx.Doc(
105 ''' Default alias for global dispatcher in builtins module. ''' ),
106] = 'ictr'
107addresscfgs: __.typx.Annotated[
108 AddressesConfigurationsRegistry,
109 __.typx.Doc( ''' Global registry of address configurations. ''' ),
110] = AddressesConfigurationsRegistry( )
111omniflavor: __.typx.Annotated[
112 Omniflavor, __.typx.Doc( ''' Matches any flavor. ''' )
113] = Omniflavor.Instance
116class Dispatcher( __.immut.DataclassObject ):
117 ''' Provides reporter instances. '''
119 active_flavors: __.typx.Annotated[
120 ActiveFlavorsRegistry,
121 __.typx.Doc(
122 ''' Mapping of addresses to active flavor sets.
124 Key ``None`` applies globally. Address-specific entries
125 override globals for that address.
126 ''' ),
127 ] = __.dcls.field( default_factory = _provide_active_flavors_default )
128 addresscfgs: __.typx.Annotated[
129 AddressesConfigurationsRegistry,
130 __.typx.Doc(
131 ''' Registry of per-address configurations.
133 Addresses inherit configuration from their parent packages.
134 Top-level packages inherit from general instance
135 configruration.
136 ''' ),
137 ] = __.dcls.field( default_factory = lambda: addresscfgs )
138 generalcfg: __.typx.Annotated[
139 _cfg.DispatcherConfiguration,
140 __.typx.Doc(
141 ''' General configuration.
143 Top of configuration inheritance hierarchy.
144 Default is suitable for application use.
145 ''' ),
146 ] = __.dcls.field( default_factory = _cfg.DispatcherConfiguration )
147 printer_factories: __.typx.Annotated[
148 _printers.PrinterFactoriesUnion,
149 __.typx.Doc(
150 ''' Factories which produce callables to output text somewhere.
152 A factory takes two arguments, address and flavor, and
153 returns a callable which takes one argument, either a record
154 or the string produced by a textualizer.
156 May also be writable text stream instead of a factory.
157 ''' ),
158 ] = ( _printers.produce_printer_factory_default( __.sys.stderr ), )
159 reporters: __.typx.Annotated[
160 ReportersRegistry,
161 __.typx.Doc(
162 ''' Cache of reporter instances by address and flavor. ''' ),
163 ] = __.dcls.field( default_factory = ReportersRegistry )
164 # TODO? Move reporters mutex into reporters registry.
165 reporters_mutex: __.typx.Annotated[
166 __.threads.Lock,
167 __.typx.Doc( ''' Access lock for cache of reporter instances. ''' ),
168 ] = __.dcls.field( default_factory = __.threads.Lock )
169 trace_levels: __.typx.Annotated[
170 TraceLevelsRegistry,
171 __.typx.Doc(
172 ''' Mapping of addresses to maximum trace depths.
174 Key ``None`` applies globally. Address-specific entries
175 override globals for that address.
176 ''' ),
177 ] = __.dcls.field(
178 default_factory = lambda: __.immut.Dictionary( { None: -1 } ) )
180 @_validate_arguments
181 def __call__(
182 self,
183 flavor: _flavors.Flavor, *,
184 address: __.Absential[ str ] = __.absent,
185 ) -> _reporters.Reporter:
186 ''' Produces and caches message reporter. '''
187 address = (
188 _discover_invoker_module_name( ) if __.is_absent( address )
189 else address )
190 cache_index = ( address, flavor )
191 if cache_index in self.reporters:
192 with self.reporters_mutex:
193 return self.reporters[ cache_index ]
194 configuration = _produce_ic_configuration( self, address, flavor )
195 if isinstance( flavor, int ):
196 trace_level = (
197 _calculate_effective_trace_level( self.trace_levels, address) )
198 active = flavor <= trace_level
199 elif isinstance( flavor, str ): # pragma: no branch
200 active_flavors = (
201 _calculate_effective_flavors( self.active_flavors, address ) )
202 active = (
203 isinstance( active_flavors, Omniflavor )
204 or flavor in active_flavors )
205 compositor = configuration[ 'compositor_factory' ]( address, flavor )
206 printers = _resolve_printers( self.printer_factories, address, flavor )
207 reporter = _reporters.Reporter(
208 active = active, address = address, flavor = flavor,
209 compositor = compositor, printers = printers )
210 with self.reporters_mutex:
211 self.reporters[ cache_index ] = reporter
212 return reporter
214 @_validate_arguments
215 def install(
216 self, alias: __.Absential[ str ] = __.absent
217 ) -> __.typx.Self:
218 ''' Installs dispatcher into builtins with provided alias.
220 Replaces an existing dispatcher. Preserves global address
221 configurations.
223 Library developers should call :py:func:`register_address` instead.
224 '''
225 import builtins
226 if __.is_absent( alias ): alias = builtins_alias_default
227 with _installer_mutex:
228 dispatcher_o = getattr( builtins, alias, None )
229 if isinstance( dispatcher_o, Dispatcher ):
230 self( 'note', address = __name__ )(
231 "Installed dispatcher is being replaced." )
232 setattr( builtins, alias, self )
233 else:
234 __.install_builtin_safely(
235 alias, self, _exceptions.AttributeNondisplacement )
236 return self
238 @_validate_arguments
239 def register_address(
240 self,
241 name: __.Absential[ str ] = __.absent,
242 configuration: __.Absential[ _cfg.AddressConfiguration ] = __.absent,
243 ) -> __.typx.Self:
244 ''' Registers configuration for address.
246 If no address is given, then the invoking module name is
247 inferred.
249 If no configuration is provided, then a default is generated.
250 '''
251 if __.is_absent( name ): 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 name = _discover_invoker_module_name( )
253 if __.is_absent( configuration ): 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 configuration = _cfg.AddressConfiguration( )
255 with _registrar_mutex:
256 self.addresscfgs[ name ] = configuration
257 return self
260ActiveFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
261 __.Absential[ ActiveFlavorsLiberal | ActiveFlavorsRegistryLiberal ],
262 __.typx.Doc(
263 ''' Flavors to activate.
265 Can be collection, which applies globally across all registered
266 addresses. Or, can be mapping of addresses to sets.
268 Address-specific entries merge with global entries.
269 ''' ),
270]
271AddressArgument: __.typx.TypeAlias = __.typx.Annotated[
272 __.Absential[ str ],
273 __.typx.Doc(
274 ''' Address to register.
276 If absent, infers the invoking module name as the address.
277 ''' ),
278]
279AddresscfgsArgument: __.typx.TypeAlias = __.typx.Annotated[
280 __.Absential[ AddressesConfigurationsRegistryLiberal ],
281 __.typx.Doc(
282 ''' Address configurations for the dispatcher.
284 If absent, defaults to global addresses registry.
285 ''' ),
286]
287CompositorFactoryArgument: __.typx.TypeAlias = __.typx.Annotated[
288 __.Absential[ _texts.CompositorFactory ],
289 __.typx.Doc(
290 ''' Factory which produces compositor callable.
292 Takes address and flavor as arguments.
293 Returns compositor to convert record content to a string.
294 ''' ),
295]
296EvnActiveFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
297 __.Absential[ __.typx.Optional[ str ] ],
298 __.typx.Doc(
299 ''' Name of environment variable for active flavors or ``None``.
301 If absent, then a default environment variable name is used.
303 If ``None``, then active flavors are not parsed from the process
304 environment.
306 If active flavors are supplied directly to a function,
307 which also accepts this argument, then active flavors are not
308 parsed from the process environment.
309 ''' ),
310]
311EvnTraceLevelsArgument: __.typx.TypeAlias = __.typx.Annotated[
312 __.Absential[ __.typx.Optional[ str ] ],
313 __.typx.Doc(
314 ''' Name of environment variable for trace levels or ``None``.
316 If absent, then a default environment variable name is used.
318 If ``None``, then trace levels are not parsed from the process
319 environment.
321 If trace levels are supplied directly to a function,
322 which also accepts this argument, then trace levels are not
323 parsed from the process environment.
324 ''' ),
325]
326FlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
327 __.Absential[ _cfg.FlavorsRegistryLiberal ],
328 __.typx.Doc( ''' Registry of flavor identifiers to configurations. ''' ),
329]
330GeneralcfgArgument: __.typx.TypeAlias = __.typx.Annotated[
331 __.Absential[ _cfg.DispatcherConfiguration ],
332 __.typx.Doc(
333 ''' General configuration for the dispatcher.
335 Top of configuration inheritance hierarchy. If absent,
336 defaults to a suitable configuration for application use.
337 ''' ),
338]
339InstallAliasArgument: __.typx.TypeAlias = __.typx.Annotated[
340 __.Absential[ str ],
341 __.typx.Doc(
342 ''' Alias under which the dispatcher is installed in builtins. ''' ),
343]
344IntroducerArgument: __.typx.TypeAlias = __.typx.Annotated[
345 __.Absential[ _texts.IntroducerUnion ],
346 __.typx.Doc(
347 ''' String or factory which produces message introduction.
349 Factory takes textualization control, address, and flavor as
350 arguments. Returns introduction string.
351 ''' ),
352]
353PrinterFactoriesArgument: __.typx.TypeAlias = __.typx.Annotated[
354 __.Absential[ _printers.PrinterFactoriesUnion ],
355 __.typx.Doc(
356 ''' Factories which produce callables to output text somewhere.
358 A factory take two arguments, address and flavor, and
359 returns a callable which takes one argument, either a record or
360 the string produced by a textualizer.
362 May also be writable text stream instead of a factory.
364 If absent, uses a default.
365 ''' ),
366]
367TraceLevelsArgument: __.typx.TypeAlias = __.typx.Annotated[
368 __.Absential[ int | TraceLevelsRegistryLiberal ],
369 __.typx.Doc(
370 ''' Maximum trace depths.
372 Can be an integer, which applies globally across all registered
373 addresses. Or, can be a mapping of addresses to integers.
375 Address-specific entries override global entries.
376 ''' ),
377]
380def active_flavors_from_environment(
381 evname: __.Absential[ str ] = __.absent
382) -> ActiveFlavorsRegistry:
383 ''' Extracts active flavors from named environment variable. '''
384 active_flavors: ActiveFlavorsRegistryLiberal = { }
385 name = 'ICTR_ACTIVE_FLAVORS' if __.is_absent( evname ) else evname
386 value = __.os.getenv( name )
387 if value is None:
388 return _provide_active_flavors_default( )
389 for part in value.split( '+' ):
390 if not part: continue 390 ↛ 389line 390 didn't jump to line 389 because the continue on line 390 wasn't executed
391 if ':' in part:
392 address, flavors = part.split( ':', 1 )
393 else: address, flavors = None, part
394 match flavors:
395 case '*': active_flavors[ address ] = omniflavor 395 ↛ 389line 395 didn't jump to line 389 because
396 case _: active_flavors[ address ] = flavors.split( ',' )
397 return __.immut.Dictionary( {
398 address:
399 flavors if isinstance( flavors, Omniflavor )
400 else frozenset( flavors )
401 for address, flavors in active_flavors.items( ) } )
404def trace_levels_from_environment(
405 evname: __.Absential[ str ] = __.absent
406) -> TraceLevelsRegistry:
407 ''' Extracts trace levels from named environment variable. '''
408 trace_levels: TraceLevelsRegistryLiberal = { None: -1 }
409 name = 'ICTR_TRACE_LEVELS' if __.is_absent( evname ) else evname
410 value = __.os.getenv( name, '' )
411 for part in value.split( '+' ):
412 if not part: continue
413 if ':' in part: address, level = part.split( ':', 1 ) 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was always true
414 else: address, level = None, part
415 if not level.isdigit( ):
416 __.warnings.warn(
417 f"Non-integer trace level {level!r} "
418 f"in environment variable {name!r}." )
419 continue
420 trace_levels[ address ] = int( level )
421 return __.immut.Dictionary( trace_levels )
424@_validate_arguments
425def install( # noqa: PLR0913, PLR0917
426 alias: InstallAliasArgument = __.absent,
427 active_flavors: ActiveFlavorsArgument = __.absent,
428 generalcfg: GeneralcfgArgument = __.absent,
429 printer_factories: PrinterFactoriesArgument = __.absent,
430 trace_levels: TraceLevelsArgument = __.absent,
431 evname_active_flavors: EvnActiveFlavorsArgument = __.absent,
432 evname_trace_levels: EvnTraceLevelsArgument = __.absent,
433) -> Dispatcher:
434 ''' Produces dispatcher and installs it into builtins with alias.
436 Replaces an existing dispatcher, preserving global address
437 configurations.
439 Library developers should call :py:func:`register_address` instead.
440 '''
441 dispatcher = produce_dispatcher(
442 active_flavors = active_flavors,
443 generalcfg = generalcfg,
444 printer_factories = printer_factories,
445 trace_levels = trace_levels,
446 evname_active_flavors = evname_active_flavors,
447 evname_trace_levels = evname_trace_levels )
448 return dispatcher.install( alias = alias )
451@_validate_arguments
452def produce_dispatcher( # noqa: PLR0913, PLR0917
453 active_flavors: ActiveFlavorsArgument = __.absent,
454 generalcfg: GeneralcfgArgument = __.absent,
455 addresscfgs: AddresscfgsArgument = __.absent,
456 printer_factories: PrinterFactoriesArgument = __.absent,
457 trace_levels: TraceLevelsArgument = __.absent,
458 evname_active_flavors: EvnActiveFlavorsArgument = __.absent,
459 evname_trace_levels: EvnTraceLevelsArgument = __.absent,
460) -> Dispatcher:
461 ''' Produces dispatcher with some shorthand argument values. '''
462 # TODO: Deeper validation of active flavors and trace levels.
463 # TODO: Deeper validation of printer factory.
464 initargs: dict[ str, __.typx.Any ] = { }
465 if not __.is_absent( generalcfg ):
466 initargs[ 'generalcfg' ] = generalcfg
467 if not __.is_absent( addresscfgs ):
468 initargs[ 'addresscfgs' ] = AddressesConfigurationsRegistry(
469 { address: configuration for address, configuration
470 in addresscfgs.items( ) } )
471 if not __.is_absent( printer_factories ):
472 initargs[ 'printer_factories' ] = printer_factories
473 _add_dispatcher_initarg_active_flavors(
474 initargs, active_flavors, evname_active_flavors )
475 _add_dispatcher_initarg_trace_levels(
476 initargs, trace_levels, evname_trace_levels )
477 return Dispatcher( **initargs )
480@_validate_arguments
481def register_address(
482 name: AddressArgument = __.absent,
483 flavors: FlavorsArgument = __.absent,
484 compositor_factory: CompositorFactoryArgument = __.absent,
485 introducer: IntroducerArgument = __.absent,
486) -> _cfg.AddressConfiguration:
487 ''' Registers address configuration on the builtin dispatcher.
489 If no dispatcher exists in builtins, installs one which produces null
490 printers.
492 Intended for library developers to configure debugging flavors
493 without overriding anything set by the application or other libraries.
494 Application developers should call :py:func:`install` instead.
495 '''
496 import builtins
497 dispatcher = getattr( builtins, builtins_alias_default, None )
498 if not isinstance( dispatcher, Dispatcher ):
499 dispatcher = Dispatcher( )
500 __.install_builtin_safely(
501 builtins_alias_default,
502 dispatcher,
503 _exceptions.AttributeNondisplacement )
504 nomargs: dict[ str, __.typx.Any ] = { }
505 if not __.is_absent( flavors ): 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true
506 nomargs[ 'flavors' ] = __.immut.Dictionary( flavors )
507 if not __.is_absent( compositor_factory ):
508 nomargs[ 'compositor_factory' ] = compositor_factory
509 if not __.is_absent( introducer ): 509 ↛ 510line 509 didn't jump to line 510 because the condition on line 509 was never true
510 nomargs[ 'introducer' ] = introducer
511 configuration = _cfg.AddressConfiguration( **nomargs )
512 dispatcher.register_address( name = name, configuration = configuration )
513 return configuration
516def _add_dispatcher_initarg_active_flavors(
517 initargs: dict[ str, __.typx.Any ],
518 active_flavors: ActiveFlavorsArgument = __.absent,
519 evname_active_flavors: EvnActiveFlavorsArgument = __.absent,
520) -> None:
521 name = 'active_flavors'
522 if not __.is_absent( active_flavors ):
523 if isinstance( active_flavors, Omniflavor ): 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true
524 initargs[ name ] = __.immut.Dictionary(
525 { None: active_flavors } )
526 elif isinstance( active_flavors, ( __.cabc.Sequence, __.cabc.Set ) ):
527 initargs[ name ] = __.immut.Dictionary(
528 { None: frozenset( active_flavors ) } )
529 else:
530 initargs[ name ] = __.immut.Dictionary( {
531 address:
532 flavors if isinstance( flavors, Omniflavor )
533 else frozenset( flavors )
534 for address, flavors in active_flavors.items( ) } )
535 elif evname_active_flavors is not None: 535 ↛ exitline 535 didn't return from function '_add_dispatcher_initarg_active_flavors' because the condition on line 535 was always true
536 initargs[ name ] = (
537 active_flavors_from_environment( evname = evname_active_flavors ) )
540def _add_dispatcher_initarg_trace_levels(
541 initargs: dict[ str, __.typx.Any ],
542 trace_levels: TraceLevelsArgument = __.absent,
543 evname_trace_levels: EvnTraceLevelsArgument = __.absent,
544) -> None:
545 name = 'trace_levels'
546 if not __.is_absent( trace_levels ):
547 if isinstance( trace_levels, int ): 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true
548 initargs[ name ] = __.immut.Dictionary( { None: trace_levels } )
549 else:
550 trace_levels_: TraceLevelsRegistryLiberal = { None: -1 }
551 trace_levels_.update( trace_levels )
552 initargs[ name ] = __.immut.Dictionary( trace_levels_ )
553 elif evname_trace_levels is not None: 553 ↛ exitline 553 didn't return from function '_add_dispatcher_initarg_trace_levels' because the condition on line 553 was always true
554 initargs[ name ] = (
555 trace_levels_from_environment( evname = evname_trace_levels ) )
558def _calculate_effective_flavors(
559 flavors: ActiveFlavorsRegistry, address: str
560) -> ActiveFlavors:
561 result_ = flavors.get( None ) or frozenset( )
562 if isinstance( result_, Omniflavor ): return result_ 562 ↛ exitline 562 didn't return from function '_calculate_effective_flavors' because the return on line 562 wasn't executed
563 result = result_
564 for address_ in _iterate_address_ancestry( address ):
565 if address_ in flavors:
566 result_ = flavors.get( address_ ) or frozenset( )
567 if isinstance( result_, Omniflavor ): return result_ 567 ↛ exitline 567 didn't return from function '_calculate_effective_flavors' because the return on line 567 wasn't executed
568 result |= result_
569 return result
572def _calculate_effective_trace_level(
573 levels: TraceLevelsRegistry, address: str
574) -> int:
575 result = levels.get( None, -1 )
576 for address_ in _iterate_address_ancestry( address ):
577 if address_ in levels:
578 result = levels[ address_ ]
579 return result
582def _dict_from_dataclass( objct: object ) -> dict[ str, __.typx.Any ]:
583 # objct = __.typx.cast( _typeshed.DataclassInstance, objct )
584 return {
585 field.name: getattr( objct, field.name )
586 for field in __.dcls.fields( objct ) # pyright: ignore[reportArgumentType]
587 if not field.name.startswith( '_' ) }
590def _discover_invoker_module_name( ) -> str:
591 frame = __.inspect.currentframe( )
592 while frame: # pragma: no branch
593 module = __.inspect.getmodule( frame )
594 if module is None:
595 if '<stdin>' == frame.f_code.co_filename: # pragma: no cover
596 name = '__main__'
597 break
598 raise _exceptions.ModuleInferenceFailure
599 name = module.__name__
600 if not name.startswith( f"{__.package_name}." ): break
601 frame = frame.f_back
602 return name
605def _is_text_stream(
606 factory: _printers.PrinterFactoryUnion
607) -> __.typx.TypeIs[ __.typx.TextIO ]:
608 ''' Runtime check for text streams (type guard for static analysis). '''
609 return isinstance( factory, __.io.TextIOBase )
612def _iterate_address_ancestry( name: str ) -> __.cabc.Iterator[ str ]:
613 parts = name.split( '.' )
614 for i in range( len( parts ) ):
615 yield '.'.join( parts[ : i + 1 ] )
618def _merge_ic_configuration(
619 base: dict[ str, __.typx.Any ], update_objct: object,
620) -> dict[ str, __.typx.Any ]:
621 update: dict[ str, __.typx.Any ] = _dict_from_dataclass( update_objct )
622 result: dict[ str, __.typx.Any ] = { }
623 result[ 'flavors' ] = (
624 dict( base.get( 'flavors', dict( ) ) )
625 | dict( update.get( 'flavors', dict( ) ) ) )
626 for ename in ( 'compositor_factory', 'introducer' ):
627 uvalue = update.get( ename )
628 if uvalue is not None: result[ ename ] = uvalue
629 elif ename in base: result[ ename ] = base[ ename ]
630 return result
633def _produce_ic_configuration(
634 dispatcher: Dispatcher, address: str, flavor: _flavors.Flavor
635) -> __.immut.Dictionary[ str, __.typx.Any ]:
636 fconfigs: list[ _cfg.FlavorConfiguration ] = [ ]
637 dconfig = dispatcher.generalcfg
638 configd: dict[ str, __.typx.Any ] = {
639 field.name: getattr( dconfig, field.name )
640 for field in __.dcls.fields( dconfig )
641 if not field.name.startswith( '_' ) }
642 if flavor in dconfig.flavors:
643 fconfigs.append( dconfig.flavors[ flavor ] )
644 for address_ in _iterate_address_ancestry( address ):
645 if address_ not in dispatcher.addresscfgs: continue
646 mconfig = dispatcher.addresscfgs[ address_ ]
647 configd = _merge_ic_configuration( configd, mconfig )
648 if flavor in mconfig.flavors:
649 fconfigs.append( mconfig.flavors[ flavor ] )
650 if not fconfigs: raise _exceptions.FlavorInavailability( flavor )
651 # Apply collected flavor configs after general and address configs.
652 # (Applied in top-down order for correct overrides.)
653 for fconfig in fconfigs:
654 configd = _merge_ic_configuration( configd, fconfig )
655 return __.immut.Dictionary( configd )
658def _resolve_printer(
659 factory: _printers.PrinterFactoryUnion,
660 address: str,
661 flavor: _flavors.Flavor,
662) -> _printers.Printer:
663 from .standard import Printer
664 if _is_text_stream( factory ):
665 return Printer( target = factory )
666 return factory( address, flavor )
669def _resolve_printers(
670 factories: _printers.PrinterFactoriesUnion,
671 address: str,
672 flavor: _flavors.Flavor,
673) -> _printers.Printers:
674 return tuple(
675 _resolve_printer( factory, address, flavor )
676 for factory in factories )