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

1import logging 

2import shutil 

3from typing import Annotated, Optional 

4 

5import fileseq 

6import typer 

7from rich.progress import ( 

8 BarColumn, 

9 MofNCompleteColumn, 

10 Progress, 

11 TaskProgressColumn, 

12 TextColumn, 

13 TimeRemainingColumn, 

14) 

15 

16from vfx_seqtools import common_options 

17from vfx_seqtools.decorators import attach_hook 

18from vfx_seqtools.parser import replace_hash_and_at_with_framenumber 

19 

20 

21def do_action(tupl: tuple) -> None: 

22 src, dst, frame, is_dryrun, be_verbose, be_interactive, be_strict, logger = tupl 

23 substituted_src = replace_hash_and_at_with_framenumber(src, frame) 

24 substituted_dst = replace_hash_and_at_with_framenumber(dst, frame) 

25 if is_dryrun: 

26 logger.info(f"dry-run: COPY {substituted_src} {substituted_dst}") 

27 else: 

28 if be_verbose: 

29 logger.info(f"COPY: {substituted_src} {substituted_dst}") 

30 try: 

31 if be_interactive: 

32 confirm = input(f"copy {substituted_src} -> {substituted_dst}? (y/n): ") 

33 if confirm.lower() != "y": 

34 return 

35 shutil.copy2(substituted_src, substituted_dst) 

36 except Exception as e: 

37 if be_strict: 

38 raise 

39 logger.warning( 

40 f"Error copying file: {substituted_src} -> {substituted_dst}, skipping it. Use --strict to stop on errors. {e}" 

41 ) 

42 

43 

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

45@attach_hook(common_options.frame_range_options, hook_output_kwarg="frame_range") 

46@attach_hook(common_options.frame_seq_options, hook_output_kwarg="frame_seq") 

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

48@attach_hook(common_options.interactive_option, hook_output_kwarg="be_interactive") 

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

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

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

52def seqcp( 

53 src: Annotated[str, typer.Argument(help="The source name pattern for the frames.")], 

54 dst: Annotated[ 

55 str, typer.Argument(help="The destination name pattern for the frames.") 

56 ], 

57 logger: logging.Logger, 

58 be_verbose: Optional[bool] = False, 

59 be_strict: Optional[bool] = False, 

60 be_interactive: Optional[bool] = False, 

61 frame_range: tuple[int, int, int] = (0, 0, 0), 

62 frame_seq: str = "", 

63 is_dryrun: Optional[bool] = False, 

64 show_version: Optional[bool] = False, 

65) -> None: 

66 """ 

67 Copy files from <SRC> to <DST>, using the provided framerange. # (padded) and @ (unpadded) are replaced with the frame number. 

68 

69 'seqcp file.####.exr newfile.####.exr -f 1-10' - copy frames 1-10 of file.####.exr to newfile, with 4-padded numbering like '0001'. 

70 

71 frame offsets using '+' and '-' are supported. 

72 

73 'seqcp file.@.exr newfile.@+10.exr -f 1-10' - copy frames 1-10 of file.@.exr to newfile, with a +10 frame number offset. 

74 """ 

75 if frame_seq: 

76 frames = fileseq.FrameSet(frame_seq) 

77 elif frame_range[0] != 0 or frame_range[1] != 0: 

78 frames = fileseq.FrameSet(f"{frame_range[0]}-{frame_range[1]}x{frame_range[2]}") 

79 else: 

80 frames = [] 

81 

82 if is_dryrun: 

83 for frame in frames: 

84 do_action( 

85 ( 

86 src, 

87 dst, 

88 frame, 

89 is_dryrun, 

90 be_verbose, 

91 be_interactive, 

92 be_strict, 

93 logger, 

94 ) 

95 ) 

96 else: 

97 # skip progress bar if interactive 

98 if be_interactive: 

99 for frame in frames: 

100 do_action( 

101 ( 

102 src, 

103 dst, 

104 frame, 

105 is_dryrun, 

106 be_verbose, 

107 be_interactive, 

108 be_strict, 

109 logger, 

110 ) 

111 ) 

112 else: 

113 from rich.logging import RichHandler 

114 

115 logging.basicConfig( 

116 level=logging.INFO, 

117 format="%(message)s", 

118 datefmt="[%X]", 

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

120 ) 

121 logger = logging.getLogger("rich") 

122 progress = Progress( 

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

124 MofNCompleteColumn(), 

125 BarColumn(), 

126 TaskProgressColumn(), 

127 TimeRemainingColumn(), 

128 ) 

129 with progress: 

130 for frame in progress.track(frames, description="Working..."): 

131 do_action( 

132 ( 

133 src, 

134 dst, 

135 frame, 

136 is_dryrun, 

137 be_verbose, 

138 be_interactive, 

139 be_strict, 

140 logger, 

141 ) 

142 )