Coverage for sources / ictr / dispatchers.py: 94%
236 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-12 01:33 +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( self, alias: str = builtins_alias_default ) -> __.typx.Self:
216 ''' Installs dispatcher into builtins with provided alias.
218 Replaces an existing dispatcher. Preserves global address
219 configurations.
221 Library developers should call :py:func:`register_address` instead.
222 '''
223 import builtins
224 with _installer_mutex:
225 dispatcher_o = getattr( builtins, alias, None )
226 if isinstance( dispatcher_o, Dispatcher ):
227 self( 'note', address = __name__ )(
228 "Installed dispatcher is being replaced." )
229 setattr( builtins, alias, self )
230 else:
231 __.install_builtin_safely(
232 alias, self, _exceptions.AttributeNondisplacement )
233 return self
235 @_validate_arguments
236 def register_address(
237 self,
238 name: __.Absential[ str ] = __.absent,
239 configuration: __.Absential[ _cfg.AddressConfiguration ] = __.absent,
240 ) -> __.typx.Self:
241 ''' Registers configuration for address.
243 If no address is given, then the invoking module name is
244 inferred.
246 If no configuration is provided, then a default is generated.
247 '''
248 if __.is_absent( name ): 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true
249 name = _discover_invoker_module_name( )
250 if __.is_absent( configuration ): 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
251 configuration = _cfg.AddressConfiguration( )
252 with _registrar_mutex:
253 self.addresscfgs[ name ] = configuration
254 return self
257ActiveFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
258 __.Absential[ ActiveFlavorsLiberal | ActiveFlavorsRegistryLiberal ],
259 __.typx.Doc(
260 ''' Flavors to activate.
262 Can be collection, which applies globally across all registered
263 addresses. Or, can be mapping of addresses to sets.
265 Address-specific entries merge with global entries.
266 ''' ),
267]
268AddressArgument: __.typx.TypeAlias = __.typx.Annotated[
269 __.Absential[ str ],
270 __.typx.Doc(
271 ''' Address to register.
273 If absent, infers the invoking module name as the address.
274 ''' ),
275]
276AddresscfgsArgument: __.typx.TypeAlias = __.typx.Annotated[
277 __.Absential[ AddressesConfigurationsRegistryLiberal ],
278 __.typx.Doc(
279 ''' Address configurations for the dispatcher.
281 If absent, defaults to global addresses registry.
282 ''' ),
283]
284CompositorFactoryArgument: __.typx.TypeAlias = __.typx.Annotated[
285 __.Absential[ _texts.CompositorFactory ],
286 __.typx.Doc(
287 ''' Factory which produces compositor callable.
289 Takes address and flavor as arguments.
290 Returns compositor to convert record content to a string.
291 ''' ),
292]
293EvnActiveFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
294 __.Absential[ __.typx.Optional[ str ] ],
295 __.typx.Doc(
296 ''' Name of environment variable for active flavors or ``None``.
298 If absent, then a default environment variable name is used.
300 If ``None``, then active flavors are not parsed from the process
301 environment.
303 If active flavors are supplied directly to a function,
304 which also accepts this argument, then active flavors are not
305 parsed from the process environment.
306 ''' ),
307]
308EvnTraceLevelsArgument: __.typx.TypeAlias = __.typx.Annotated[
309 __.Absential[ __.typx.Optional[ str ] ],
310 __.typx.Doc(
311 ''' Name of environment variable for trace levels or ``None``.
313 If absent, then a default environment variable name is used.
315 If ``None``, then trace levels are not parsed from the process
316 environment.
318 If trace levels are supplied directly to a function,
319 which also accepts this argument, then trace levels are not
320 parsed from the process environment.
321 ''' ),
322]
323FlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
324 __.Absential[ _cfg.FlavorsRegistryLiberal ],
325 __.typx.Doc( ''' Registry of flavor identifiers to configurations. ''' ),
326]
327GeneralcfgArgument: __.typx.TypeAlias = __.typx.Annotated[
328 __.Absential[ _cfg.DispatcherConfiguration ],
329 __.typx.Doc(
330 ''' General configuration for the dispatcher.
332 Top of configuration inheritance hierarchy. If absent,
333 defaults to a suitable configuration for application use.
334 ''' ),
335]
336InstallAliasArgument: __.typx.TypeAlias = __.typx.Annotated[
337 str,
338 __.typx.Doc(
339 ''' Alias under which the dispatcher is installed in builtins. ''' ),
340]
341IntroducerArgument: __.typx.TypeAlias = __.typx.Annotated[
342 __.Absential[ _texts.IntroducerUnion ],
343 __.typx.Doc(
344 ''' String or factory which produces message introduction.
346 Factory takes textualization control, address, and flavor as
347 arguments. Returns introduction string.
348 ''' ),
349]
350PrinterFactoriesArgument: __.typx.TypeAlias = __.typx.Annotated[
351 __.Absential[ _printers.PrinterFactoriesUnion ],
352 __.typx.Doc(
353 ''' Factories which produce callables to output text somewhere.
355 A factory take two arguments, address and flavor, and
356 returns a callable which takes one argument, either a record or
357 the string produced by a textualizer.
359 May also be writable text stream instead of a factory.
361 If absent, uses a default.
362 ''' ),
363]
364TraceLevelsArgument: __.typx.TypeAlias = __.typx.Annotated[
365 __.Absential[ int | TraceLevelsRegistryLiberal ],
366 __.typx.Doc(
367 ''' Maximum trace depths.
369 Can be an integer, which applies globally across all registered
370 addresses. Or, can be a mapping of addresses to integers.
372 Address-specific entries override global entries.
373 ''' ),
374]
377def active_flavors_from_environment(
378 evname: __.Absential[ str ] = __.absent
379) -> ActiveFlavorsRegistry:
380 ''' Extracts active flavors from named environment variable. '''
381 active_flavors: ActiveFlavorsRegistryLiberal = { }
382 name = 'ICTR_ACTIVE_FLAVORS' if __.is_absent( evname ) else evname
383 value = __.os.getenv( name )
384 if value is None:
385 return _provide_active_flavors_default( )
386 for part in value.split( '+' ):
387 if not part: continue 387 ↛ 386line 387 didn't jump to line 386 because the continue on line 387 wasn't executed
388 if ':' in part:
389 address, flavors = part.split( ':', 1 )
390 else: address, flavors = None, part
391 match flavors:
392 case '*': active_flavors[ address ] = omniflavor 392 ↛ 386line 392 didn't jump to line 386 because
393 case _: active_flavors[ address ] = flavors.split( ',' )
394 return __.immut.Dictionary( {
395 address:
396 flavors if isinstance( flavors, Omniflavor )
397 else frozenset( flavors )
398 for address, flavors in active_flavors.items( ) } )
401def trace_levels_from_environment(
402 evname: __.Absential[ str ] = __.absent
403) -> TraceLevelsRegistry:
404 ''' Extracts trace levels from named environment variable. '''
405 trace_levels: TraceLevelsRegistryLiberal = { None: -1 }
406 name = 'ICTR_TRACE_LEVELS' if __.is_absent( evname ) else evname
407 value = __.os.getenv( name, '' )
408 for part in value.split( '+' ):
409 if not part: continue
410 if ':' in part: address, level = part.split( ':', 1 ) 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was always true
411 else: address, level = None, part
412 if not level.isdigit( ):
413 __.warnings.warn(
414 f"Non-integer trace level {level!r} "
415 f"in environment variable {name!r}." )
416 continue
417 trace_levels[ address ] = int( level )
418 return __.immut.Dictionary( trace_levels )
421@_validate_arguments
422def install( # noqa: PLR0913
423 alias: InstallAliasArgument = builtins_alias_default,
424 active_flavors: ActiveFlavorsArgument = __.absent,
425 generalcfg: GeneralcfgArgument = __.absent,
426 printer_factories: PrinterFactoriesArgument = __.absent,
427 trace_levels: TraceLevelsArgument = __.absent,
428 evname_active_flavors: EvnActiveFlavorsArgument = __.absent,
429 evname_trace_levels: EvnTraceLevelsArgument = __.absent,
430) -> Dispatcher:
431 ''' Produces dispatcher and installs it into builtins with alias.
433 Replaces an existing dispatcher, preserving global address
434 configurations.
436 Library developers should call :py:func:`register_address` instead.
437 '''
438 dispatcher = produce_dispatcher(
439 active_flavors = active_flavors,
440 generalcfg = generalcfg,
441 printer_factories = printer_factories,
442 trace_levels = trace_levels,
443 evname_active_flavors = evname_active_flavors,
444 evname_trace_levels = evname_trace_levels )
445 return dispatcher.install( alias = alias )
448@_validate_arguments
449def produce_dispatcher( # noqa: PLR0913
450 active_flavors: ActiveFlavorsArgument = __.absent,
451 generalcfg: GeneralcfgArgument = __.absent,
452 addresscfgs: AddresscfgsArgument = __.absent,
453 printer_factories: PrinterFactoriesArgument = __.absent,
454 trace_levels: TraceLevelsArgument = __.absent,
455 evname_active_flavors: EvnActiveFlavorsArgument = __.absent,
456 evname_trace_levels: EvnTraceLevelsArgument = __.absent,
457) -> Dispatcher:
458 ''' Produces dispatcher with some shorthand argument values. '''
459 # TODO: Deeper validation of active flavors and trace levels.
460 # TODO: Deeper validation of printer factory.
461 initargs: dict[ str, __.typx.Any ] = { }
462 if not __.is_absent( generalcfg ):
463 initargs[ 'generalcfg' ] = generalcfg
464 if not __.is_absent( addresscfgs ):
465 initargs[ 'addresscfgs' ] = AddressesConfigurationsRegistry(
466 { address: configuration for address, configuration
467 in addresscfgs.items( ) } )
468 if not __.is_absent( printer_factories ):
469 initargs[ 'printer_factories' ] = printer_factories
470 _add_dispatcher_initarg_active_flavors(
471 initargs, active_flavors, evname_active_flavors )
472 _add_dispatcher_initarg_trace_levels(
473 initargs, trace_levels, evname_trace_levels )
474 return Dispatcher( **initargs )
477@_validate_arguments
478def register_address(
479 name: AddressArgument = __.absent,
480 flavors: FlavorsArgument = __.absent,
481 compositor_factory: CompositorFactoryArgument = __.absent,
482 introducer: IntroducerArgument = __.absent,
483) -> _cfg.AddressConfiguration:
484 ''' Registers address configuration on the builtin dispatcher.
486 If no dispatcher exists in builtins, installs one which produces null
487 printers.
489 Intended for library developers to configure debugging flavors
490 without overriding anything set by the application or other libraries.
491 Application developers should call :py:func:`install` instead.
492 '''
493 import builtins
494 dispatcher = getattr( builtins, builtins_alias_default, None )
495 if not isinstance( dispatcher, Dispatcher ):
496 dispatcher = Dispatcher( )
497 __.install_builtin_safely(
498 builtins_alias_default,
499 dispatcher,
500 _exceptions.AttributeNondisplacement )
501 nomargs: dict[ str, __.typx.Any ] = { }
502 if not __.is_absent( flavors ): 502 ↛ 503line 502 didn't jump to line 503 because the condition on line 502 was never true
503 nomargs[ 'flavors' ] = __.immut.Dictionary( flavors )
504 if not __.is_absent( compositor_factory ):
505 nomargs[ 'compositor_factory' ] = compositor_factory
506 if not __.is_absent( introducer ): 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true
507 nomargs[ 'introducer' ] = introducer
508 configuration = _cfg.AddressConfiguration( **nomargs )
509 dispatcher.register_address( name = name, configuration = configuration )
510 return configuration
513def _add_dispatcher_initarg_active_flavors(
514 initargs: dict[ str, __.typx.Any ],
515 active_flavors: ActiveFlavorsArgument = __.absent,
516 evname_active_flavors: EvnActiveFlavorsArgument = __.absent,
517) -> None:
518 name = 'active_flavors'
519 if not __.is_absent( active_flavors ):
520 if isinstance( active_flavors, Omniflavor ): 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true
521 initargs[ name ] = __.immut.Dictionary(
522 { None: active_flavors } )
523 elif isinstance( active_flavors, ( __.cabc.Sequence, __.cabc.Set ) ):
524 initargs[ name ] = __.immut.Dictionary(
525 { None: frozenset( active_flavors ) } )
526 else:
527 initargs[ name ] = __.immut.Dictionary( {
528 address:
529 flavors if isinstance( flavors, Omniflavor )
530 else frozenset( flavors )
531 for address, flavors in active_flavors.items( ) } )
532 elif evname_active_flavors is not None: 532 ↛ exitline 532 didn't return from function '_add_dispatcher_initarg_active_flavors' because the condition on line 532 was always true
533 initargs[ name ] = (
534 active_flavors_from_environment( evname = evname_active_flavors ) )
537def _add_dispatcher_initarg_trace_levels(
538 initargs: dict[ str, __.typx.Any ],
539 trace_levels: TraceLevelsArgument = __.absent,
540 evname_trace_levels: EvnTraceLevelsArgument = __.absent,
541) -> None:
542 name = 'trace_levels'
543 if not __.is_absent( trace_levels ):
544 if isinstance( trace_levels, int ): 544 ↛ 545line 544 didn't jump to line 545 because the condition on line 544 was never true
545 initargs[ name ] = __.immut.Dictionary( { None: trace_levels } )
546 else:
547 trace_levels_: TraceLevelsRegistryLiberal = { None: -1 }
548 trace_levels_.update( trace_levels )
549 initargs[ name ] = __.immut.Dictionary( trace_levels_ )
550 elif evname_trace_levels is not None: 550 ↛ exitline 550 didn't return from function '_add_dispatcher_initarg_trace_levels' because the condition on line 550 was always true
551 initargs[ name ] = (
552 trace_levels_from_environment( evname = evname_trace_levels ) )
555def _calculate_effective_flavors(
556 flavors: ActiveFlavorsRegistry, address: str
557) -> ActiveFlavors:
558 result_ = flavors.get( None ) or frozenset( )
559 if isinstance( result_, Omniflavor ): return result_ 559 ↛ exitline 559 didn't return from function '_calculate_effective_flavors' because the return on line 559 wasn't executed
560 result = result_
561 for address_ in _iterate_address_ancestry( address ):
562 if address_ in flavors:
563 result_ = flavors.get( address_ ) or frozenset( )
564 if isinstance( result_, Omniflavor ): return result_ 564 ↛ exitline 564 didn't return from function '_calculate_effective_flavors' because the return on line 564 wasn't executed
565 result |= result_
566 return result
569def _calculate_effective_trace_level(
570 levels: TraceLevelsRegistry, address: str
571) -> int:
572 result = levels.get( None, -1 )
573 for address_ in _iterate_address_ancestry( address ):
574 if address_ in levels:
575 result = levels[ address_ ]
576 return result
579def _dict_from_dataclass( objct: object ) -> dict[ str, __.typx.Any ]:
580 # objct = __.typx.cast( _typeshed.DataclassInstance, objct )
581 return {
582 field.name: getattr( objct, field.name )
583 for field in __.dcls.fields( objct ) # pyright: ignore[reportArgumentType]
584 if not field.name.startswith( '_' ) }
587def _discover_invoker_module_name( ) -> str:
588 frame = __.inspect.currentframe( )
589 while frame: # pragma: no branch
590 module = __.inspect.getmodule( frame )
591 if module is None:
592 if '<stdin>' == frame.f_code.co_filename: # pragma: no cover
593 name = '__main__'
594 break
595 raise _exceptions.ModuleInferenceFailure
596 name = module.__name__
597 if not name.startswith( f"{__.package_name}." ): break
598 frame = frame.f_back
599 return name
602def _iterate_address_ancestry( name: str ) -> __.cabc.Iterator[ str ]:
603 parts = name.split( '.' )
604 for i in range( len( parts ) ):
605 yield '.'.join( parts[ : i + 1 ] )
608def _merge_ic_configuration(
609 base: dict[ str, __.typx.Any ], update_objct: object,
610) -> dict[ str, __.typx.Any ]:
611 update: dict[ str, __.typx.Any ] = _dict_from_dataclass( update_objct )
612 result: dict[ str, __.typx.Any ] = { }
613 result[ 'flavors' ] = (
614 dict( base.get( 'flavors', dict( ) ) )
615 | dict( update.get( 'flavors', dict( ) ) ) )
616 for ename in ( 'compositor_factory', 'introducer' ):
617 uvalue = update.get( ename )
618 if uvalue is not None: result[ ename ] = uvalue
619 elif ename in base: result[ ename ] = base[ ename ]
620 return result
623def _produce_ic_configuration(
624 dispatcher: Dispatcher, address: str, flavor: _flavors.Flavor
625) -> __.immut.Dictionary[ str, __.typx.Any ]:
626 fconfigs: list[ _cfg.FlavorConfiguration ] = [ ]
627 dconfig = dispatcher.generalcfg
628 configd: dict[ str, __.typx.Any ] = {
629 field.name: getattr( dconfig, field.name )
630 for field in __.dcls.fields( dconfig )
631 if not field.name.startswith( '_' ) }
632 if flavor in dconfig.flavors:
633 fconfigs.append( dconfig.flavors[ flavor ] )
634 for address_ in _iterate_address_ancestry( address ):
635 if address_ not in dispatcher.addresscfgs: continue
636 mconfig = dispatcher.addresscfgs[ address_ ]
637 configd = _merge_ic_configuration( configd, mconfig )
638 if flavor in mconfig.flavors:
639 fconfigs.append( mconfig.flavors[ flavor ] )
640 if not fconfigs: raise _exceptions.FlavorInavailability( flavor )
641 # Apply collected flavor configs after general and address configs.
642 # (Applied in top-down order for correct overrides.)
643 for fconfig in fconfigs:
644 configd = _merge_ic_configuration( configd, fconfig )
645 return __.immut.Dictionary( configd )
648def _resolve_printer(
649 factory: _printers.PrinterFactoryUnion,
650 address: str,
651 flavor: _flavors.Flavor,
652) -> _printers.Printer:
653 from .standard import Printer
654 if isinstance( factory, __.io.TextIOBase ):
655 return Printer( target = factory )
656 return factory( address, flavor )
659def _resolve_printers(
660 factories: _printers.PrinterFactoriesUnion,
661 address: str,
662 flavor: _flavors.Flavor,
663) -> _printers.Printers:
664 return tuple(
665 _resolve_printer( factory, address, flavor )
666 for factory in factories )