Coverage for sources/ictruck/vehicles.py: 100%

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. '''

24from __future__ import annotations

26import icecream as _icecream

28from . import __

29from . import configuration as _cfg

30from . import exceptions as _exceptions

33# pylint: disable=import-error,import-private-name

34if __.typx.TYPE_CHECKING: # pragma: no cover

35 import _typeshed

36# pylint: enable=import-error,import-private-name

39_validate_arguments = (

40 __.validate_arguments(

41 globalvars = globals( ),

42 errorclass = _exceptions.ArgumentClassInvalidity ) )

45builtins_alias_default: __.typx.Annotated[

46 str,

47 __.typx.Doc( ''' Default alias for global truck in builtins module. ''' ),

48] = 'ictr'

51class Truck( metaclass = __.ImmutableCompleteDataclass ):

52 ''' Vends flavors of Icecream debugger. '''

54 # pylint: disable=invalid-field-call

55 active_flavors: __.typx.Annotated[

56 ActiveFlavorsRegistry,

57 __.typx.Doc(

58 ''' Mapping of module names to active flavor sets.

60 Key ``None`` applies globally. Module-specific entries

61 override globals for that module.

62 ''' ),

63 ] = __.dcls.field( default_factory = __.ImmutableDictionary ) # pyright: ignore

64 generalcfg: __.typx.Annotated[

65 _cfg.VehicleConfiguration,

66 __.typx.Doc(

67 ''' General configuration.

69 Top of configuration inheritance hierarchy.

70 Default is suitable for application use.

71 ''' ),

72 ] = __.dcls.field( default_factory = _cfg.VehicleConfiguration )

73 modulecfgs: __.typx.Annotated[

74 __.AccretiveDictionary[ str, _cfg.ModuleConfiguration ],

75 __.typx.Doc(

76 ''' Registry of per-module configurations.

78 Modules inherit configuration from their parent packages.

79 Top-level packages inherit from general instance

80 configruration.

81 ''' ),

82 ] = __.dcls.field( default_factory = __.AccretiveDictionary ) # pyright: ignore

83 printer_factory: __.typx.Annotated[

84 PrinterFactoryUnion,

85 __.typx.Doc(

86 ''' Factory which produces callables to output text somewhere.

88 May also be writable text stream.

89 Factories take two arguments, module name and flavor, and

90 return a callable which takes one argument, the string

91 produced by a formatter.

92 ''' ),

93 ] = __.dcls.field(

94 default_factory = (

95 lambda: lambda mname, flavor: _icecream.DEFAULT_OUTPUT_FUNCTION ) )

96 trace_levels: __.typx.Annotated[

97 TraceLevelsRegistry,

98 __.typx.Doc(

99 ''' Mapping of module names to maximum trace depths.

100

101 Key ``None`` applies globally. Module-specific entries

102 override globals for that module.

103 ''' ),

104 ] = __.dcls.field(

105 default_factory = lambda: __.ImmutableDictionary( { None: -1 } ) )

106 _debuggers: __.typx.Annotated[

107 __.AccretiveDictionary[

108 tuple[ str, _cfg.Flavor ], _icecream.IceCreamDebugger ],

109 __.typx.Doc(

110 ''' Cache of debugger instances by module and flavor. ''' ),

111 ] = __.dcls.field( default_factory = __.AccretiveDictionary ) # pyright: ignore

112 _debuggers_lock: __.typx.Annotated[

113 __.threads.Lock,

114 __.typx.Doc( ''' Access lock for cache of debugger instances. ''' ),

115 ] = __.dcls.field( default_factory = __.threads.Lock )

116 # pylint: enable=invalid-field-call

117

118 @_validate_arguments

119 def __call__( self, flavor: _cfg.Flavor ) -> _icecream.IceCreamDebugger:

120 ''' Vends flavor of Icecream debugger. '''

121 mname = _discover_invoker_module_name( )

122 cache_index = ( mname, flavor )

123 if cache_index in self._debuggers: # pylint: disable=unsupported-membership-test

124 with self._debuggers_lock: # pylint: disable=not-context-manager

125 return self._debuggers[ cache_index ] # pylint: disable=unsubscriptable-object

126 configuration = _produce_ic_configuration( self, mname, flavor )

127 control = _cfg.FormatterControl( )

128 initargs = _calculate_ic_initargs(

129 self, configuration, control, mname, flavor )

130 debugger = _icecream.IceCreamDebugger( **initargs )

131 if isinstance( flavor, int ):

132 trace_level = (

133 _calculate_effective_trace_level( self.trace_levels, mname) )

134 debugger.enabled = flavor <= trace_level

135 elif isinstance( flavor, str ): # pragma: no branch

136 active_flavors = (

137 _calculate_effective_flavors( self.active_flavors, mname ) )

138 debugger.enabled = flavor in active_flavors

139 with self._debuggers_lock: # pylint: disable=not-context-manager

140 self._debuggers[ cache_index ] = debugger # pylint: disable=unsupported-assignment-operation

141 return debugger

142

143 @_validate_arguments

144 def register_module(

145 self,

146 name: __.Absential[ str ] = __.absent,

147 configuration: __.Absential[ _cfg.ModuleConfiguration ] = __.absent,

148 ) -> None:

149 ''' Registers configuration for module.

150

151 If no module or package name is given, then the current module is

152 inferred.

153

154 If no configuration is provided, then a default is generated.

155 '''

156 if __.is_absent( name ):

157 name = _discover_invoker_module_name( )

158 if __.is_absent( configuration ):

159 configuration = _cfg.ModuleConfiguration( )

160 self.modulecfgs[ name ] = configuration # pylint: disable=unsupported-assignment-operation

161

162

163ActiveFlavors: __.typx.TypeAlias = frozenset[ _cfg.Flavor ]

164ActiveFlavorsLiberal: __.typx.TypeAlias = (

165 __.cabc.Sequence[ _cfg.Flavor ] | __.cabc.Set[ _cfg.Flavor ] )

166ActiveFlavorsRegistry: __.typx.TypeAlias = (

167 __.ImmutableDictionary[ str | None, ActiveFlavors ] )

168ActiveFlavorsRegistryLiberal: __.typx.TypeAlias = (

169 __.cabc.Mapping[ str | None, ActiveFlavorsLiberal ] )

170Printer: __.typx.TypeAlias = __.cabc.Callable[ [ str ], None ]

171PrinterFactory: __.typx.TypeAlias = (

172 __.cabc.Callable[ [ str, _cfg.Flavor ], Printer ] )

173PrinterFactoryUnion: __.typx.TypeAlias = __.io.TextIOBase | PrinterFactory

174TraceLevelsRegistry: __.typx.TypeAlias = (

175 __.ImmutableDictionary[ str | None, int ] )

176TraceLevelsRegistryLiberal: __.typx.TypeAlias = (

177 __.cabc.Mapping[ str | None, int ] )

178

179InstallAliasArgument: __.typx.TypeAlias = __.typx.Annotated[

180 str,

181 __.typx.Doc(

182 ''' Alias under which the truck is installed in builtins. ''' ),

183]

184ProduceTruckActiveFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[

185 __.Absential[ ActiveFlavorsLiberal | ActiveFlavorsRegistryLiberal ],

186 __.typx.Doc(

187 ''' Flavors to activate.

188

189 Can be collection, which applies globally across all registered

190 modules. Or, can be mapping of module names to sets.

191

192 Module-specific entries merge with global entries.

193 ''' ),

194]

195ProduceTruckFlavorsArgument: __.typx.TypeAlias = __.typx.Annotated[

196 __.Absential[ _cfg.FlavorsRegistryLiberal ],

197 __.typx.Doc( ''' Registry of flavor identifiers to configurations. ''' ),

198]

199ProduceTruckGeneralcfgArgument: __.typx.TypeAlias = __.typx.Annotated[

200 __.Absential[ _cfg.VehicleConfiguration ],

201 __.typx.Doc(

202 ''' General configuration for the truck.

203

204 Top of configuration inheritance hierarchy. If absent,

205 defaults to a suitable configuration for application use.

206 ''' ),

207]

208ProduceTruckPrinterFactoryArgument: __.typx.TypeAlias = __.typx.Annotated[

209 __.Absential[ PrinterFactoryUnion ],

210 __.typx.Doc(

211 ''' Factory which produces callables to output text somewhere.

212

213 May also be writable text stream.

214 Factories take two arguments, module name and flavor, and

215 return a callable which takes one argument, the string

216 produced by a formatter.

217

218 If absent, uses a default.

219 ''' ),

220]

221ProduceTruckTraceLevelsArgument: __.typx.TypeAlias = __.typx.Annotated[

222 __.Absential[ int | TraceLevelsRegistryLiberal ],

223 __.typx.Doc(

224 ''' Maximum trace depths.

225

226 Can be an integer, which applies globally across all registered

227 modules. Or, can be a mapping of module names to integers.

228

229 Module-specific entries override global entries.

230 ''' ),

231]

232RegisterModuleFormatterFactoryArgument: __.typx.TypeAlias = __.typx.Annotated[

233 __.Absential[ _cfg.FormatterFactory ],

234 __.typx.Doc(

235 ''' Factory which produces formatter callable.

236

237 Takes formatter control, module name, and flavor as arguments.

238 Returns formatter to convert an argument to a string.

239 ''' ),

240]

241RegisterModuleIncludeContextArgument: __.typx.TypeAlias = __.typx.Annotated[

242 __.Absential[ bool ],

243 __.typx.Doc( ''' Include stack frame with output? ''' ),

244]

245RegisterModuleNameArgument: __.typx.TypeAlias = __.typx.Annotated[

246 __.Absential[ str ],

247 __.typx.Doc(

248 ''' Name of the module to register.

249

250 If absent, infers the current module name.

251 ''' ),

252]

253RegisterModulePrefixEmitterArgument: __.typx.TypeAlias = __.typx.Annotated[

254 __.Absential[ _cfg.PrefixEmitterUnion ],

255 __.typx.Doc(

256 ''' String or factory which produces output prefix string.

257

258 Factory takes formatter control, module name, and flavor as

259 arguments. Returns prefix string.

260 ''' ),

261]

262

263@_validate_arguments

264def install(

265 alias: InstallAliasArgument = builtins_alias_default,

266 active_flavors: ProduceTruckActiveFlavorsArgument = __.absent,

267 generalcfg: ProduceTruckGeneralcfgArgument = __.absent,

268 printer_factory: ProduceTruckPrinterFactoryArgument = __.absent,

269 trace_levels: ProduceTruckTraceLevelsArgument = __.absent,

270) -> Truck:

271 ''' Installs configured truck into builtins.

272

273 Application developers should call this early before importing

274 library packages which may also use the builtin truck.

275

276 Library developers should call :py:func:`register_module` instead.

277 '''

278 truck = produce_truck(

279 active_flavors = active_flavors,

280 generalcfg = generalcfg,

281 printer_factory = printer_factory,

282 trace_levels = trace_levels )

283 __.install_builtin_safely(

284 alias, truck, _exceptions.AttributeNondisplacement )

285 return truck

286

287

288@_validate_arguments

289def produce_truck(

290 active_flavors: ProduceTruckActiveFlavorsArgument = __.absent,

291 generalcfg: ProduceTruckGeneralcfgArgument = __.absent,

292 printer_factory: ProduceTruckPrinterFactoryArgument = __.absent,

293 trace_levels: ProduceTruckTraceLevelsArgument = __.absent,

294) -> Truck:

295 ''' Produces icecream truck with some shorthand argument values. '''

296 # TODO: Deeper validation of active flavors and trace levels.

297 # TODO: Deeper validation of printer factory.

298 nomargs: dict[ str, __.typx.Any ] = { }

299 if not __.is_absent( generalcfg ):

300 nomargs[ 'generalcfg' ] = generalcfg

301 if not __.is_absent( printer_factory ):

302 nomargs[ 'printer_factory' ] = printer_factory

303 if not __.is_absent( active_flavors ):

304 if isinstance( active_flavors, ( __.cabc.Sequence, __.cabc.Set ) ):

305 nomargs[ 'active_flavors' ] = __.ImmutableDictionary(

306 { None: frozenset( active_flavors ) } )

307 else:

308 nomargs[ 'active_flavors' ] = __.ImmutableDictionary( {

309 mname: frozenset( flavors )

310 for mname, flavors in active_flavors.items( ) } )

311 if not __.is_absent( trace_levels ):

312 if isinstance( trace_levels, int ):

313 nomargs[ 'trace_levels' ] = __.ImmutableDictionary(

314 { None: trace_levels } )

315 else:

316 nomargs[ 'trace_levels' ] = __.ImmutableDictionary( trace_levels )

317 return Truck( **nomargs )

318

319

320@_validate_arguments

321def register_module(

322 name: RegisterModuleNameArgument = __.absent,

323 flavors: ProduceTruckFlavorsArgument = __.absent,

324 formatter_factory: RegisterModuleFormatterFactoryArgument = __.absent,

325 include_context: RegisterModuleIncludeContextArgument = __.absent,

326 prefix_emitter: RegisterModulePrefixEmitterArgument = __.absent,

327) -> None:

328 ''' Registers module configuration on the builtin truck.

329

330 If no truck exists in builtins, installs one which produces null

331 printers.

332

333 Intended for library developers to configure debugging flavors

334 without overriding anything set by the application or other libraries.

335 Application developers should call :py:func:`install` instead.

336 '''

337 if builtins_alias_default not in __builtins__:

338 truck = Truck( printer_factory = lambda mname, flavor: lambda x: None )

339 __builtins__[ builtins_alias_default ] = truck

340 else: truck = __builtins__[ builtins_alias_default ]

341 nomargs: dict[ str, __.typx.Any ] = { }

342 if not __.is_absent( flavors ):

343 nomargs[ 'flavors' ] = __.ImmutableDictionary( flavors )

344 if not __.is_absent( formatter_factory ):

345 nomargs[ 'formatter_factory' ] = formatter_factory

346 if not __.is_absent( include_context ):

347 nomargs[ 'include_context' ] = include_context

348 if not __.is_absent( prefix_emitter ):

349 nomargs[ 'prefix_emitter' ] = prefix_emitter

350 configuration = _cfg.ModuleConfiguration( **nomargs )

351 truck.register_module( name = name, configuration = configuration )

352

353

354def _calculate_effective_flavors(

355 flavors: ActiveFlavorsRegistry, mname: str

356) -> ActiveFlavors:

357 result = set( flavors.get( None, frozenset( ) ) )

358 for mname_ in _iterate_module_name_ancestry( mname ):

359 if mname_ in flavors:

360 result |= set( flavors[ mname_ ] )

361 return frozenset( result )

362

363

364def _calculate_effective_trace_level(

365 levels: TraceLevelsRegistry, mname: str

366) -> int:

367 result = levels.get( None, -1 )

368 for mname_ in _iterate_module_name_ancestry( mname ):

369 if mname_ in levels:

370 result = levels[ mname_ ]

371 return result

372

373

374def _calculate_ic_initargs(

375 truck: Truck,

376 configuration: __.ImmutableDictionary[ str, __.typx.Any ],

377 control: _cfg.FormatterControl,

378 mname: str,

379 flavor: _cfg.Flavor,

380) -> dict[ str, __.typx.Any ]:

381 nomargs: dict[ str, __.typx.Any ] = { }

382 nomargs[ 'argToStringFunction' ] = (

383 configuration[ 'formatter_factory' ]( control, mname, flavor ) )

384 nomargs[ 'includeContext' ] = configuration[ 'include_context' ]

385 if isinstance( truck.printer_factory, __.io.TextIOBase ):

386 printer = __.funct.partial( print, file = truck.printer_factory )

387 else: printer = truck.printer_factory( mname, flavor ) # pylint: disable=not-callable

388 nomargs[ 'outputFunction' ] = printer

389 prefix_emitter = configuration[ 'prefix_emitter' ]

390 nomargs[ 'prefix' ] = (

391 prefix_emitter if isinstance( prefix_emitter, str )

392 else prefix_emitter( mname, flavor ) )

393 return nomargs

394

395

396def _dict_from_dataclass(

397 obj: _typeshed.DataclassInstance

398) -> dict[ str, __.typx.Any ]:

399 return {

400 field.name: getattr( obj, field.name )

401 for field in __.dcls.fields( obj ) }

402

403

404def _discover_invoker_module_name( ) -> str:

405 frame = __.inspect.currentframe( )

406 while frame: # pragma: no branch

407 module = __.inspect.getmodule( frame )

408 if module is None:

409 # pylint: disable=magic-value-comparison

410 if '<stdin>' == frame.f_code.co_filename: # pragma: no cover

411 name = '__main__'

412 break

413 # pylint: enable=magic-value-comparison

414 raise _exceptions.ModuleInferenceFailure

415 name = module.__name__

416 if not name.startswith( f"{__package__}." ): break

417 frame = frame.f_back

418 return name

419

420

421def _iterate_module_name_ancestry( name: str ) -> __.cabc.Iterator[ str ]:

422 parts = name.split( '.' )

423 for i in range( len( parts ) ):

424 yield '.'.join( parts[ : i + 1 ] )

425

426

427def _merge_ic_configuration(

428 base: dict[ str, __.typx.Any ], update_obj: _typeshed.DataclassInstance

429) -> dict[ str, __.typx.Any ]:

430 update: dict[ str, __.typx.Any ] = _dict_from_dataclass( update_obj )

431 result: dict[ str, __.typx.Any ] = { }

432 result[ 'flavors' ] = (

433 dict( base.get( 'flavors', dict( ) ) )

434 | dict( update.get( 'flavors', dict( ) ) ) )

435 for ename in ( 'formatter_factory', 'include_context', 'prefix_emitter' ):

436 uvalue = update.get( ename )

437 if uvalue is not None: result[ ename ] = uvalue

438 elif ename in base: result[ ename ] = base[ ename ]

439 return result

440

441

442def _produce_ic_configuration(

443 vehicle: Truck, mname: str, flavor: _cfg.Flavor

444) -> __.ImmutableDictionary[ str, __.typx.Any ]:

445 fconfigs: list[ _cfg.FlavorConfiguration ] = [ ]

446 vconfig = vehicle.generalcfg

447 configd: dict[ str, __.typx.Any ] = {

448 field.name: getattr( vconfig, field.name )

449 for field in __.dcls.fields( vconfig ) }

450 if flavor in vconfig.flavors:

451 fconfigs.append( vconfig.flavors[ flavor ] )

452 for mname_ in _iterate_module_name_ancestry( mname ):

453 if mname_ not in vehicle.modulecfgs: continue

454 mconfig = vehicle.modulecfgs[ mname_ ]

455 configd = _merge_ic_configuration( configd, mconfig )

456 if flavor in mconfig.flavors:

457 fconfigs.append( mconfig.flavors[ flavor ] )

458 if not fconfigs: raise _exceptions.FlavorInavailability( flavor )

459 # Apply collected flavor configs after general and module configs.

460 # (Applied in top-down order for correct overrides.)

461 for fconfig in fconfigs:

462 configd = _merge_ic_configuration( configd, fconfig )

463 return __.ImmutableDictionary( configd )