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

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 . import __ 

25 

26 

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

28 # TODO: Protected class attributes. 

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

30 

31 Stderr = 'stderr' 

32 Stdout = 'stdout' 

33 

34 

35class ConsoleDisplay( __.immut.DataclassObject ): 

36 silence: __.typx.Annotated[ 

37 bool, 

38 __.tyro.conf.arg( 

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

40 ] = False 

41 file: __.typx.Annotated[ 

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

43 __.tyro.conf.arg( 

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

45 ] = None 

46 stream: __.typx.Annotated[ 

47 DisplayStreams, 

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

49 ] = DisplayStreams.Stderr 

50 

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

52 ''' Provides output stream for display. ''' 

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

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

55 # TODO: async context manager for async file streams 

56 # TODO: return async stream - need async printers 

57 # TODO: handle non-TextIOWrapper streams 

58 match self.stream: 

59 case DisplayStreams.Stdout: 

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

61 case DisplayStreams.Stderr: 

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

63 

64 

65class CliCommand( 

66 __.immut.DataclassProtocol, __.typx.Protocol, 

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

68): 

69 ''' CLI command. ''' 

70 

71 @__.abc.abstractmethod 

72 async def __call__( 

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

74 ) -> None: 

75 ''' Executes command with global state. ''' 

76 raise NotImplementedError 

77 

78 # TODO: provide_configuration_edits