Coverage for python/lsst/images/cli/_reformat.py: 88%

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11from __future__ import annotations 

12 

13__all__ = ("reformat",) 

14 

15import os 

16import tempfile 

17 

18import click 

19 

20from ..serialization import ArchiveReadError, backend_for_path, read, write 

21 

22 

23@click.command(name="reformat") 

24@click.argument("input", type=click.Path(exists=True, dir_okay=False)) 

25@click.argument("output", type=click.Path(dir_okay=False)) 

26@click.option("--overwrite", is_flag=True, default=False, help="Overwrite OUTPUT if it exists.") 

27def reformat(input: str, output: str, overwrite: bool) -> None: 

28 """Rewrite an lsst.images file in a different container format. 

29 

30 Reads INPUT and writes it back out to OUTPUT, choosing the format from 

31 OUTPUT's extension. This is the easy way to, for example, turn a FITS 

32 file into an NDF for testing. 

33 """ 

34 try: 

35 backend = backend_for_path(output) 

36 except ValueError as err: 

37 raise click.ClickException(str(err)) from None 

38 

39 output_abs = os.path.realpath(output) 

40 if os.path.realpath(input) == output_abs: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 raise click.ClickException("INPUT and OUTPUT must be different paths.") 

42 

43 if os.path.exists(output_abs) and not overwrite: 

44 raise click.ClickException(f"{output!r} already exists; pass --overwrite to replace it.") 

45 

46 try: 

47 obj = read(input) 

48 except (ValueError, ArchiveReadError) as err: 

49 raise click.ClickException(f"Could not read {input!r}: {err}") from None 

50 

51 # Write to a temporary file in the output's directory and move it into 

52 # place only after a successful write, so a failure never destroys an 

53 # existing OUTPUT (the backends refuse to overwrite, so the temporary path 

54 # must not already exist). 

55 output_dir = os.path.dirname(output_abs) 

56 with tempfile.TemporaryDirectory(dir=output_dir) as staging: 

57 staged = os.path.join(staging, os.path.basename(output_abs)) 

58 write(obj, staged) 

59 os.replace(staged, output_abs) 

60 click.echo(f"Wrote {output} ({backend.name}).")