Coverage for /opt/hostedtoolcache/Python/3.10.17/x64/lib/python3.10/site-packages/vfx_seqtools/actions/seqchk.py: 93%

1import glob 

2import logging 

3import pathlib 

4from typing import Annotated, Optional 

5 

6import fileseq 

7import OpenEXR 

8import typer 

9from PIL import Image 

10from rich.progress import ( 

11 BarColumn, 

12 MofNCompleteColumn, 

13 Progress, 

14 TaskProgressColumn, 

15 TextColumn, 

16 TimeRemainingColumn, 

17) 

18 

19from vfx_seqtools import common_options 

20from vfx_seqtools.decorators import attach_hook 

21 

22 

23def do_action(tupl: tuple) -> bool: 

24 framefile, is_dryrun, be_verbose, be_strict, logger = tupl 

25 if is_dryrun: 

26 logger.info(f"dry-run: CHECK {framefile}") 

27 return True 

28 else: 

29 # Run the command using subprocess or any other method 

30 if be_verbose: 

31 logger.info(f"CHECK {framefile}") 

32 

33 try: 

34 framepath = pathlib.Path(framefile) 

35 if framepath.suffix.lower() == ".exr": 

36 with OpenEXR.File(framefile): 

37 # header = infile.header() 

38 # print(f"type={header['type']}") 

39 # print(f"compression={header['compression']}") 

40 

41 # RGB = infile.channels()["RGB"].pixels 

42 # height, width = RGB.shape[0:2] 

43 # print(f"width={width}, height={height}") 

44 # print(f"channels={infile.channels()}") 

45 pass 

46 else: 

47 Image.open(framefile).verify() 

48 # print(f"{verify_response} was returned by verify") 

49 Image.open(framefile) 

50 # print(f"{im.filename.split('/')[-1]} is a valid image") 

51 # im.show() 

52 return True 

53 except Exception: 

54 return False 

55 

56 

57@attach_hook(common_options.logging_options, hook_output_kwarg="logger") 

58@attach_hook(common_options.dry_run_option, hook_output_kwarg="is_dryrun") 

59@attach_hook(common_options.verbose_option, hook_output_kwarg="be_verbose") 

60@attach_hook(common_options.strict_option, hook_output_kwarg="be_strict") 

61@attach_hook(common_options.version_option, hook_output_kwarg="show_version") 

62@attach_hook(common_options.sequence_only_option, hook_output_kwarg="only_on_sequences") 

63def seqchk( 

64 show_version: bool, 

65 logger: logging.Logger, 

66 filepattern: Annotated[ 

67 str, 

68 typer.Argument( 

69 help="An optional file pattern to use, use quotes or escape shell wildcards like '?' and '*'." 

70 ), 

71 ] = "", 

72 be_verbose: Optional[bool] = False, 

73 be_strict: Optional[bool] = False, 

74 is_dryrun: Optional[bool] = False, 

75 only_on_sequences: bool = False, 

76) -> None: 

77 """ 

78 Check file sequences. 

79 

80 seqchk - will check all file sequences in the current directory. 

81 

82 seqchk file.\*.exr - will check all file sequences matching the pattern 'file.*.exr' (escaping shell wildcards). 

83 

84 seqchk "file.????.exr" - will check all file sequences matching the pattern 'file.????.exr' (quoting shell wildcards). 

85 

86 """ 

87 if filepattern: 

88 files = glob.glob(filepattern) 

89 seqs = fileseq.findSequencesInList(files) 

90 else: 

91 seqs = fileseq.findSequencesOnDisk(".") 

92 

93 for seq in seqs: 

94 if only_on_sequences and "@" not in str(seq) and "#" not in str(seq): 

95 continue 

96 if is_dryrun: 

97 for framefile in list(seq): 

98 do_action((framefile, is_dryrun, be_verbose, be_strict, logger)) 

99 else: 

100 bad_frames = [] 

101 from rich.logging import RichHandler 

102 

103 logging.basicConfig( 

104 level=logging.INFO, 

105 format="%(message)s", 

106 datefmt="[%X]", 

107 handlers=[RichHandler(rich_tracebacks=True)], 

108 ) 

109 logger = logging.getLogger("rich") 

110 progress = Progress( 

111 TextColumn("[progress.description]{task.description}"), 

112 MofNCompleteColumn(), 

113 BarColumn(), 

114 TaskProgressColumn(), 

115 TimeRemainingColumn(), 

116 ) 

117 with progress: 

118 for framefile in progress.track( 

119 list(seq), description=f"Checking {str(seq)}..." 

120 ): 

121 check_ok = do_action( 

122 (framefile, is_dryrun, be_verbose, be_strict, logger) 

123 ) 

124 if not check_ok: 

125 bad_frames.append(framefile) 

126 

127 if bad_frames: 

128 if len(bad_frames) == 1: 

129 print(f"{str(seq)}: Bad frame: {bad_frames[0]}") 

130 else: 

131 badseqs = fileseq.findSequencesInList(bad_frames) 

132 # print(f"Bad sequences: {badseqs}") 

133 print(f"{str(seq)}: Bad frames: {badseqs[0].frameSet()}")