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

1import logging 

2import subprocess 

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 cmds, frame, is_dryrun, be_verbose, be_strict, logger = tupl 

23 substituted_cmds = replace_hash_and_at_with_framenumber(cmds, frame) 

24 if is_dryrun: 

25 logger.info(f"dry-run: {substituted_cmds}") 

26 else: 

27 # Run the command using subprocess or any other method 

28 if be_verbose: 

29 logger.info(f"{substituted_cmds}") 

30 

31 try: 

32 subprocess.run(substituted_cmds, shell=True) 

33 except Exception as e: 

34 if be_strict: 

35 raise 

36 logger.warning( 

37 f"Command failed: {substituted_cmds}, skipping it. Use --strict to stop on errors. {e}" 

38 ) 

39 

40 

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

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

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

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

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

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

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

48def seqdo( 

49 cmds: Annotated[ 

50 str, typer.Argument(help="The command(s) to run. Use quotes to group commands.") 

51 ], 

52 logger: logging.Logger, 

53 be_verbose: Optional[bool] = False, 

54 be_strict: Optional[bool] = False, 

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

56 frame_seq: str = "", 

57 is_dryrun: Optional[bool] = False, 

58 show_version: Optional[bool] = False, 

59) -> None: 

60 """ 

61 Do command(s) for the provided framerange. Use '@' and '#' to specify frame numbers in the command. 

62 

63 For example, 'echo @' will print the frame number, and 'echo file.#.exr' will print the file name with the frame number. 

64 

65 Use 'seqdo "echo @" -f 1-5' to print the frame number for each frame in the range 1-5. 

66 

67 Use 'seqdo "echo file.####.exr" -f 6-10' to print the file name with the frame number (4-padded) for each frame in the range 6-10. 

68 """ 

69 if frame_seq: 

70 frames = fileseq.FrameSet(frame_seq) 

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

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

73 else: 

74 frames = [] 

75 

76 if is_dryrun: 

77 for frame in frames: 

78 do_action((cmds, frame, is_dryrun, be_verbose, be_strict, logger)) 

79 else: 

80 from rich.logging import RichHandler 

81 

82 logging.basicConfig( 

83 level=logging.INFO, 

84 format="%(message)s", 

85 datefmt="[%X]", 

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

87 ) 

88 logger = logging.getLogger("rich") 

89 progress = Progress( 

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

91 MofNCompleteColumn(), 

92 BarColumn(), 

93 TaskProgressColumn(), 

94 TimeRemainingColumn(), 

95 ) 

96 with progress: 

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

98 do_action((cmds, frame, is_dryrun, be_verbose, be_strict, logger))