Coverage for sources/emcdproj/interfaces.py: 50%

20 statements  

« prev     ^ index     » next       coverage.py v7.8.2, created at 2025-05-27 20:29 +0000

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

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#============================================================================# 

19 

20 

21''' Abstract bases and interfaces. ''' 

22 

23 

24from __future__ import annotations 

25 

26from . import __ 

27 

28 

29class DisplayStreams( __.enum.Enum ): # TODO: Python 3.11: StrEnum 

30 # TODO: Protected class attributes. 

31 ''' Stream upon which to place output. ''' 

32 

33 Stderr = 'stderr' 

34 Stdout = 'stdout' 

35 

36 

37class ConsoleDisplay( 

38 metaclass = __.ImmutableDataclass, 

39): 

40 silence: __.typx.Annotated[ 

41 bool, 

42 __.tyro.conf.arg( 

43 aliases = ( '--quiet', '--silent', ), prefix_name = False ), 

44 ] = False 

45 file: __.typx.Annotated[ 

46 __.typx.Optional[ __.Path ], 

47 __.tyro.conf.arg( 

48 name = 'console-capture-file', prefix_name = False ), 

49 ] = None 

50 stream: __.typx.Annotated[ 

51 DisplayStreams, 

52 __.tyro.conf.arg( name = 'console-stream', prefix_name = False ), 

53 ] = DisplayStreams.Stderr 

54 

55 async def provide_stream( self ) -> __.io.TextIOWrapper: 

56 ''' Provides output stream for display. ''' 

57 # TODO: register file stream as a process-lifetime exit 

58 if self.file: return open( self.file, 'w' ) 

59 # TODO: async context manager for async file streams 

60 # TODO: return async stream - need async printers 

61 # TODO: handle non-TextIOWrapper streams 

62 match self.stream: 

63 case DisplayStreams.Stdout: 

64 return __.sys.stdout # pyright: ignore[reportReturnType] 

65 case DisplayStreams.Stderr: 

66 return __.sys.stderr # pyright: ignore[reportReturnType] 

67 

68 

69class CliCommand( 

70 __.typx.Protocol, 

71 metaclass = __.ImmutableProtocolDataclass, 

72 decorators = ( __.typx.runtime_checkable, ), 

73): 

74 ''' CLI command. ''' 

75 

76 @__.abc.abstractmethod 

77 async def __call__( 

78 self, auxdata: __.Globals, display: ConsoleDisplay 

79 ) -> None: 

80 ''' Executes command with global state. ''' 

81 raise NotImplementedError 

82 

83 # TODO: provide_configuration_edits