Coverage for sources/ictruck/vehicles.py: 100%
175 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-03-31 03: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''' Vehicles which vend flavors of Icecream debugger. '''
23# TODO: Always add module configuration for 'ictruck' itself to trucks.
24# This allows for it to trace much of its own execution if its flavors
25# are activated. Suggested flavors:
26# ictruck-note: Noteworthy event.
27# ictruck-error: Error.
30from __future__ import annotations
32import icecream as _icecream
34from . import __
35from . import configuration as _cfg
36from . import exceptions as _exceptions
39# pylint: disable=import-error,import-private-name
40if __.typx.TYPE_CHECKING: # pragma: no cover
41 import _typeshed
42# pylint: enable=import-error,import-private-name
45_installer_lock: __.threads.Lock = __.threads.Lock( )
46_validate_arguments = (
47 __.validate_arguments(
48 globalvars = globals( ),
49 errorclass = _exceptions.ArgumentClassInvalidity ) )
52builtins_alias_default: __.typx.Annotated[
53 str,
54 __.typx.Doc( ''' Default alias for global truck in builtins module. ''' ),
55] = 'ictr'
58class Truck( metaclass = __.ImmutableCompleteDataclass ):
59 ''' Vends flavors of Icecream debugger. '''
61 # pylint: disable=invalid-field-call
62 active_flavors: __.typx.Annotated[
63 ActiveFlavorsRegistry,
64 __.typx.Doc(
65 ''' Mapping of module names to active flavor sets.
67 Key ``None`` applies globally. Module-specific entries
68 override globals for that module.
69 ''' ),
70 ] = __.dcls.field( default_factory = __.ImmutableDictionary ) # pyright: ignore
71 generalcfg: __.typx.Annotated[
72 _cfg.VehicleConfiguration,
73 __.typx.Doc(
74 ''' General configuration.
76 Top of configuration inheritance hierarchy.
77 Default is suitable for application use.
78 ''' ),
79 ] = __.dcls.field( default_factory = _cfg.VehicleConfiguration )
80 modulecfgs: __.typx.Annotated[
81 __.AccretiveDictionary[ str, _cfg.ModuleConfiguration ],
82 __.typx.Doc(
83 ''' Registry of per-module configurations.
85 Modules inherit configuration from their parent packages.
86 Top-level packages inherit from general instance
87 configruration.
88 ''' ),
89 ] = __.dcls.field( default_factory = __.AccretiveDictionary ) # pyright: ignore
90 printer_factory: __.typx.Annotated[
91 PrinterFactoryUnion,
92 __.typx.Doc(
93 ''' Factory which produces callables to output text somewhere.
95 May also be writable text stream.
96 Factories take two arguments, module name and flavor, and
97 return a callable which takes one argument, the string
98 produced by a formatter.
99 ''' ),
100 ] = __.dcls.field(
101 default_factory = (
102 lambda: lambda mname, flavor: _icecream.DEFAULT_OUTPUT_FUNCTION ) )
103 trace_levels: __.typx.Annotated[
104 TraceLevelsRegistry,
105 __.typx.Doc(
106 ''' Mapping of module names to maximum trace depths.
108 Key ``None`` applies globally. Module-specific entries
109 override globals for that module.
110 ''' ),
111 ] = __.dcls.field(
112 default_factory = lambda: __.ImmutableDictionary( { None: -1 } ) )
113 _debuggers: __.typx.Annotated[
114 __.AccretiveDictionary[
115 tuple[ str, _cfg.Flavor ], _icecream.IceCreamDebugger ],
116 __.typx.Doc(
117 ''' Cache of debugger instances by module and flavor. ''' ),
118 ] = __.dcls.field( default_factory = __.AccretiveDictionary ) # pyright: ignore
119 _debuggers_lock: __.typx.Annotated[
120 __.threads.Lock,
121 __.typx.Doc( ''' Access lock for cache of debugger instances. ''' ),
122 ] = __.dcls.field( default_factory = __.threads.Lock )
123 # pylint: enable=invalid-field-call
125 @_validate_arguments
126 def __call__( self, flavor: _cfg.Flavor ) -> _icecream.IceCreamDebugger:
127 ''' Vends flavor of Icecream debugger. '''
128 mname = _discover_invoker_module_name( )
129 cache_index = ( mname, flavor )
130 if cache_index in self._debuggers: # pylint: disable=unsupported-membership-test
131 with self._debuggers_lock: # pylint: disable=not-context-manager
132 return self._debuggers[ cache_index ] # pylint: disable=unsubscriptable-object
133 configuration = _produce_ic_configuration( self, mname, flavor )
134 control = _cfg.FormatterControl( )
135 initargs = _calculate_ic_initargs(
136 self, configuration, control, mname, flavor )
137 debugger = _icecream.IceCreamDebugger( **initargs )
138 if isinstance( flavor, int ):
139 trace_level = (
140 _calculate_effective_trace_level( self.trace_levels, mname) )
141 debugger.enabled = flavor <= trace_level
142 elif isinstance( flavor, str ): # pragma: no branch
143 active_flavors = (
144 _calculate_effective_flavors( self.active_flavors, mname ) )
145 debugger.enabled = flavor in active_flavors
146 with self._debuggers_lock: # pylint: disable=not-context-manager
147 self._debuggers[ cache_index ] = debugger # pylint: disable=unsupported-assignment-operation
148 return debugger
150 @_validate_arguments
151 def install( self, alias: str = builtins_alias_default ) -> __.typx.Self:
152 ''' Installs truck into builtins with provided alias.
154 Replaces an existing truck, preserving its module configurations.
156 Library developers should call :py:func:`register_module` instead.
157 '''
158 import builtins
159 with _installer_lock:
160 truck_o = getattr( builtins, alias, None )
161 if isinstance( truck_o, Truck ):
162 # TODO: self( 'ictruck-note' )( 'truck replacement', self )
163 self.modulecfgs.update( truck_o.modulecfgs )
164 setattr( builtins, alias, self )
165 else:
166 __.install_builtin_safely(
167 alias, self, _exceptions.AttributeNondisplacement )
168 return self
170 @_validate_arguments
171 def register_module(
172 self,
173 name: __.Absential[ str ] = __.absent,
174 configuration: __.Absential[ _cfg.ModuleConfiguration ] = __.absent,
175 ) -> __.typx.Self:
176 ''' Registers configuration for module.
178 If no module or package name is given, then the current module is
179 inferred.
181 If no configuration is provided, then a default is generated.
182 '''
183 if __.is_absent( name ):
184 name = _discover_invoker_module_name( )
185 if __.is_absent( configuration ):
186 configuration = _cfg.ModuleConfiguration( )
187 self.modulecfgs[ name ] = configuration # pylint: disable=unsupported-assignment-operation
188 return self
191ActiveFlavors: __.typx.TypeAlias = frozenset[ _cfg.Flavor ]
192ActiveFlavorsLiberal: __.typx.TypeAlias = (
193 __.cabc.Sequence[ _cfg.Flavor ] | __.cabc.Set[ _cfg.Flavor ] )
194ActiveFlavorsRegistry: __.typx.TypeAlias = (
195 __.ImmutableDictionary[ str | None, ActiveFlavors ] )
196ActiveFlavorsRegistryLiberal: __.typx.TypeAlias = (
197 __.cabc.Mapping[ str | None, ActiveFlavorsLiberal ] )
198Printer: __.typx.TypeAlias = __.cabc.Callable[ [ str ], None ]
199PrinterFactory: __.typx.TypeAlias = (
200 __.cabc.Callable[ [ str, _cfg.Flavor ], Printer ] )
201PrinterFactoryUnion: __.typx.TypeAlias = __.io.TextIOBase | PrinterFactory
202TraceLevelsRegistry: __.typx.TypeAlias = (
203 __.ImmutableDictionary[ str | None, int ] )
204TraceLevelsRegistryLiberal: __.typx.TypeAlias = (
205 __.cabc.Mapping[ str | None, int ] )
207InstallAliasArgument: __.typx.TypeAlias = __.typx.Annotated[
208 str,
209 __.typx.Doc(
210 ''' Alias under which the truck is installed in builtins. ''' ),
211]
212ProduceTruckActiveFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
213 __.Absential[ ActiveFlavorsLiberal | ActiveFlavorsRegistryLiberal ],
214 __.typx.Doc(
215 ''' Flavors to activate.
217 Can be collection, which applies globally across all registered
218 modules. Or, can be mapping of module names to sets.
220 Module-specific entries merge with global entries.
221 ''' ),
222]
223ProduceTruckFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[
224 __.Absential[ _cfg.FlavorsRegistryLiberal ],
225 __.typx.Doc( ''' Registry of flavor identifiers to configurations. ''' ),
226]
227ProduceTruckGeneralcfgArgument: __.typx.TypeAlias = __.typx.Annotated[
228 __.Absential[ _cfg.VehicleConfiguration ],
229 __.typx.Doc(
230 ''' General configuration for the truck.
232 Top of configuration inheritance hierarchy. If absent,
233 defaults to a suitable configuration for application use.
234 ''' ),
235]
236ProduceTruckPrinterFactoryArgument: __.typx.TypeAlias = __.typx.Annotated[
237 __.Absential[ PrinterFactoryUnion ],
238 __.typx.Doc(
239 ''' Factory which produces callables to output text somewhere.
241 May also be writable text stream.
242 Factories take two arguments, module name and flavor, and
243 return a callable which takes one argument, the string
244 produced by a formatter.
246 If absent, uses a default.
247 ''' ),
248]
249ProduceTruckTraceLevelsArgument: __.typx.TypeAlias = __.typx.Annotated[
250 __.Absential[ int | TraceLevelsRegistryLiberal ],
251 __.typx.Doc(
252 ''' Maximum trace depths.
254 Can be an integer, which applies globally across all registered
255 modules. Or, can be a mapping of module names to integers.
257 Module-specific entries override global entries.
258 ''' ),
259]
260RegisterModuleFormatterFactoryArgument: __.typx.TypeAlias = __.typx.Annotated[
261 __.Absential[ _cfg.FormatterFactory ],
262 __.typx.Doc(
263 ''' Factory which produces formatter callable.
265 Takes formatter control, module name, and flavor as arguments.
266 Returns formatter to convert an argument to a string.
267 ''' ),
268]
269RegisterModuleIncludeContextArgument: __.typx.TypeAlias = __.typx.Annotated[
270 __.Absential[ bool ],
271 __.typx.Doc( ''' Include stack frame with output? ''' ),
272]
273RegisterModuleNameArgument: __.typx.TypeAlias = __.typx.Annotated[
274 __.Absential[ str ],
275 __.typx.Doc(
276 ''' Name of the module to register.
278 If absent, infers the current module name.
279 ''' ),
280]
281RegisterModulePrefixEmitterArgument: __.typx.TypeAlias = __.typx.Annotated[
282 __.Absential[ _cfg.PrefixEmitterUnion ],
283 __.typx.Doc(
284 ''' String or factory which produces output prefix string.
286 Factory takes formatter control, module name, and flavor as
287 arguments. Returns prefix string.
288 ''' ),
289]
291@_validate_arguments
292def install(
293 alias: InstallAliasArgument = builtins_alias_default,
294 active_flavors: ProduceTruckActiveFlavorsArgument = __.absent,
295 generalcfg: ProduceTruckGeneralcfgArgument = __.absent,
296 printer_factory: ProduceTruckPrinterFactoryArgument = __.absent,
297 trace_levels: ProduceTruckTraceLevelsArgument = __.absent,
298) -> Truck:
299 ''' Produces truck and installs it into builtins with alias.
301 Replaces an existing truck, preserving its module configurations.
303 Library developers should call :py:func:`register_module` instead.
304 '''
305 truck = produce_truck(
306 active_flavors = active_flavors,
307 generalcfg = generalcfg,
308 printer_factory = printer_factory,
309 trace_levels = trace_levels )
310 return truck.install( alias = alias )
313@_validate_arguments
314def produce_truck(
315 active_flavors: ProduceTruckActiveFlavorsArgument = __.absent,
316 generalcfg: ProduceTruckGeneralcfgArgument = __.absent,
317 printer_factory: ProduceTruckPrinterFactoryArgument = __.absent,
318 trace_levels: ProduceTruckTraceLevelsArgument = __.absent,
319) -> Truck:
320 ''' Produces icecream truck with some shorthand argument values. '''
321 # TODO: Deeper validation of active flavors and trace levels.
322 # TODO: Deeper validation of printer factory.
323 nomargs: dict[ str, __.typx.Any ] = { }
324 if not __.is_absent( generalcfg ):
325 nomargs[ 'generalcfg' ] = generalcfg
326 if not __.is_absent( printer_factory ):
327 nomargs[ 'printer_factory' ] = printer_factory
328 if not __.is_absent( active_flavors ):
329 if isinstance( active_flavors, ( __.cabc.Sequence, __.cabc.Set ) ):
330 nomargs[ 'active_flavors' ] = __.ImmutableDictionary(
331 { None: frozenset( active_flavors ) } )
332 else:
333 nomargs[ 'active_flavors' ] = __.ImmutableDictionary( {
334 mname: frozenset( flavors )
335 for mname, flavors in active_flavors.items( ) } )
336 if not __.is_absent( trace_levels ):
337 if isinstance( trace_levels, int ):
338 nomargs[ 'trace_levels' ] = __.ImmutableDictionary(
339 { None: trace_levels } )
340 else:
341 nomargs[ 'trace_levels' ] = __.ImmutableDictionary( trace_levels )
342 return Truck( **nomargs )
345@_validate_arguments
346def register_module(
347 name: RegisterModuleNameArgument = __.absent,
348 flavors: ProduceTruckFlavorsArgument = __.absent,
349 formatter_factory: RegisterModuleFormatterFactoryArgument = __.absent,
350 include_context: RegisterModuleIncludeContextArgument = __.absent,
351 prefix_emitter: RegisterModulePrefixEmitterArgument = __.absent,
352) -> None:
353 ''' Registers module configuration on the builtin truck.
355 If no truck exists in builtins, installs one which produces null
356 printers.
358 Intended for library developers to configure debugging flavors
359 without overriding anything set by the application or other libraries.
360 Application developers should call :py:func:`install` instead.
361 '''
362 import builtins
363 truck = getattr( builtins, builtins_alias_default, None )
364 if not isinstance( truck, Truck ):
365 truck = Truck( printer_factory = lambda mname, flavor: lambda x: None )
366 __.install_builtin_safely(
367 builtins_alias_default,
368 truck,
369 _exceptions.AttributeNondisplacement )
370 nomargs: dict[ str, __.typx.Any ] = { }
371 if not __.is_absent( flavors ):
372 nomargs[ 'flavors' ] = __.ImmutableDictionary( flavors )
373 if not __.is_absent( formatter_factory ):
374 nomargs[ 'formatter_factory' ] = formatter_factory
375 if not __.is_absent( include_context ):
376 nomargs[ 'include_context' ] = include_context
377 if not __.is_absent( prefix_emitter ):
378 nomargs[ 'prefix_emitter' ] = prefix_emitter
379 configuration = _cfg.ModuleConfiguration( **nomargs )
380 truck.register_module( name = name, configuration = configuration )
383def _calculate_effective_flavors(
384 flavors: ActiveFlavorsRegistry, mname: str
385) -> ActiveFlavors:
386 result = set( flavors.get( None, frozenset( ) ) )
387 for mname_ in _iterate_module_name_ancestry( mname ):
388 if mname_ in flavors:
389 result |= set( flavors[ mname_ ] )
390 return frozenset( result )
393def _calculate_effective_trace_level(
394 levels: TraceLevelsRegistry, mname: str
395) -> int:
396 result = levels.get( None, -1 )
397 for mname_ in _iterate_module_name_ancestry( mname ):
398 if mname_ in levels:
399 result = levels[ mname_ ]
400 return result
403def _calculate_ic_initargs(
404 truck: Truck,
405 configuration: __.ImmutableDictionary[ str, __.typx.Any ],
406 control: _cfg.FormatterControl,
407 mname: str,
408 flavor: _cfg.Flavor,
409) -> dict[ str, __.typx.Any ]:
410 nomargs: dict[ str, __.typx.Any ] = { }
411 nomargs[ 'argToStringFunction' ] = (
412 configuration[ 'formatter_factory' ]( control, mname, flavor ) )
413 nomargs[ 'includeContext' ] = configuration[ 'include_context' ]
414 if isinstance( truck.printer_factory, __.io.TextIOBase ):
415 printer = __.funct.partial( print, file = truck.printer_factory )
416 else: printer = truck.printer_factory( mname, flavor ) # pylint: disable=not-callable
417 nomargs[ 'outputFunction' ] = printer
418 prefix_emitter = configuration[ 'prefix_emitter' ]
419 nomargs[ 'prefix' ] = (
420 prefix_emitter if isinstance( prefix_emitter, str )
421 else prefix_emitter( mname, flavor ) )
422 return nomargs
425def _dict_from_dataclass(
426 obj: _typeshed.DataclassInstance
427) -> dict[ str, __.typx.Any ]:
428 return {
429 field.name: getattr( obj, field.name )
430 for field in __.dcls.fields( obj ) }
433def _discover_invoker_module_name( ) -> str:
434 frame = __.inspect.currentframe( )
435 while frame: # pragma: no branch
436 module = __.inspect.getmodule( frame )
437 if module is None:
438 # pylint: disable=magic-value-comparison
439 if '<stdin>' == frame.f_code.co_filename: # pragma: no cover
440 name = '__main__'
441 break
442 # pylint: enable=magic-value-comparison
443 raise _exceptions.ModuleInferenceFailure
444 name = module.__name__
445 if not name.startswith( f"{__package__}." ): break
446 frame = frame.f_back
447 return name
450def _iterate_module_name_ancestry( name: str ) -> __.cabc.Iterator[ str ]:
451 parts = name.split( '.' )
452 for i in range( len( parts ) ):
453 yield '.'.join( parts[ : i + 1 ] )
456def _merge_ic_configuration(
457 base: dict[ str, __.typx.Any ], update_obj: _typeshed.DataclassInstance
458) -> dict[ str, __.typx.Any ]:
459 update: dict[ str, __.typx.Any ] = _dict_from_dataclass( update_obj )
460 result: dict[ str, __.typx.Any ] = { }
461 result[ 'flavors' ] = (
462 dict( base.get( 'flavors', dict( ) ) )
463 | dict( update.get( 'flavors', dict( ) ) ) )
464 for ename in ( 'formatter_factory', 'include_context', 'prefix_emitter' ):
465 uvalue = update.get( ename )
466 if uvalue is not None: result[ ename ] = uvalue
467 elif ename in base: result[ ename ] = base[ ename ]
468 return result
471def _produce_ic_configuration(
472 vehicle: Truck, mname: str, flavor: _cfg.Flavor
473) -> __.ImmutableDictionary[ str, __.typx.Any ]:
474 fconfigs: list[ _cfg.FlavorConfiguration ] = [ ]
475 vconfig = vehicle.generalcfg
476 configd: dict[ str, __.typx.Any ] = {
477 field.name: getattr( vconfig, field.name )
478 for field in __.dcls.fields( vconfig ) }
479 if flavor in vconfig.flavors:
480 fconfigs.append( vconfig.flavors[ flavor ] )
481 for mname_ in _iterate_module_name_ancestry( mname ):
482 if mname_ not in vehicle.modulecfgs: continue
483 mconfig = vehicle.modulecfgs[ mname_ ]
484 configd = _merge_ic_configuration( configd, mconfig )
485 if flavor in mconfig.flavors:
486 fconfigs.append( mconfig.flavors[ flavor ] )
487 if not fconfigs: raise _exceptions.FlavorInavailability( flavor )
488 # Apply collected flavor configs after general and module configs.
489 # (Applied in top-down order for correct overrides.)
490 for fconfig in fconfigs:
491 configd = _merge_ic_configuration( configd, fconfig )
492 return __.ImmutableDictionary( configd )