# Copyright 2016 - 2023 Ternaris
# SPDX-License-Identifier: Apache-2.0
"""Generated declinate CLI."""

# DO NOT EDIT THIS FILE MANUALLY
# pyright: basic
# ruff: noqa: PGH004
# ruff: noqa

from __future__ import annotations

import argparse
import os
import re
import sys
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from typing import Callable, NoReturn


class HelpFormatter(argparse.HelpFormatter):
    """Help formatter."""

    def __init__(
        self,
        prog: str,
        indent_increment: int = 2,
        max_help_position: int = 24,
        width: int | None = None,
    ) -> None:
        """Init."""
        super().__init__(prog, indent_increment, max_help_position, width)
        self._width = min(self._width, 78)

    def _fill_text(self, text: str, width: int, indent: str) -> str:
        """Reformat individual paragraphs."""
        parent = super()._fill_text

        def idn(text: str) -> str:
            if re.match(r'(?m)(^.*?:$)|(^ )', text):
                return text
            return parent(text, width, indent)

        return '\n\n'.join(map(idn, text.split('\n\n')))

    def _metavar_formatter(
        self,
        action: argparse.Action,
        default_metavar: str,
    ) -> Callable[[int], tuple[str, ...]]:
        if isinstance(action, argparse._SubParsersAction) and action.choices:
            choice_strs = [
                k for k, v in action.choices.items() if not v.description.startswith('SUPPRESS.')
            ]
            result = f'{{{",".join(choice_strs)}}}'
            return lambda x: (result,) * x
        return super()._metavar_formatter(action, default_metavar)


class ArgumentParser(argparse.ArgumentParser):
    """Argument parser."""

    def _check_value(self, action: argparse.Action, value: str) -> None:
        """Filter suppressed actions."""
        if isinstance(action, argparse._SubParsersAction):
            if action.choices and value not in action.choices:
                msg = f'invalid action: {value!r}'
                raise argparse.ArgumentError(action, msg)
        else:
            super()._check_value(action, value)


# fmt: off
PARSER = ArgumentParser(
    allow_abbrev=False,
    argument_default=argparse.SUPPRESS,
    formatter_class=HelpFormatter,
    description=(
        'Merge, filter, and convert rosbags.\n'
        '\n'
        'This tool reads messages from source rosbags and writes them into a\n'
        'destination rosbag.\n'
        '\n'
        'When multiple source rosbags are provided, their connections are merged,\n'
        'and messages are written in correct chronological order.\n'
        '\n'
        'Source rosbags are automatically decompressed, destination rosbags can\n'
        'optionally be compressed.\n'
        '\n'
        'Source rosbag connections can be filtered by excluding or including based\n'
        'on topics and/or message types. Exclusions take precedence over inclusions.\n'
        '\n'
        'Source and destination rosbag versions do not have to match. The desired\n'
        'target version is derived from the destination filename, and conversion is\n'
        'handled automatically. When no additional parameters are specified,\n'
        'only the container type (rosbag1 <=> rosbag2) and the serialization\n'
        'format (ros1 <=> cdr) will be changed. Message definitions will be copied\n'
        'over as is from the source.\n'
        '\n'
        'Experimental: If the destination typestore is set to anything other than\n'
        "'copy', an automatic conversion of messages will be performed. For each\n"
        'message type conversion, it will:\n'
        '\n'
        '    - Detect trivial field renames.\n'
        '    - Drop fields that have been removed.\n'
        '    - Add a default value for fields that have been added.\n'
        '\n'
        'Usage Examples:\n'
        '\n'
        '    Convert bag from rosbag1 to rosbag2:\n'
        '        rosbags-convert --src example.bag --dst ros2_bagdir\n'
        '\n'
        '    Convert bag from rosbag1 to rosbag2, using per file compression for destination:\n'
        '        rosbags-convert --src example.bag --dst ros2_bagdir --compress zstd\n'
        '\n'
        '    Convert bag from rosbag1 to rosbag2, upgrade types to iron:\n'
        '        rosbags-convert --src example.bag --dst ros2_bagdir --dst-typestore ros2_iron\n'
        '\n'
        '    Convert bag from legacy rosbag2 (with humble types) to rosbag1:\n'
        '        rosbags-convert --src ros2_bagdir --dst dst.bag --src_typestore ros2_humble\n'
        '\n'
        '    Copy only image topics:\n'
        '        rosbags-convert --src src.bag --dst dst.bag --include-topic sensor_msgs/msg/Image'
    ),
)
PARSER_dststore = PARSER.add_argument_group(
    'Destination typestore',
).add_mutually_exclusive_group()  # fmt: skip
PARSER_msgtypes = PARSER.add_argument_group(
    'Select message types to exclude or include using the ROS2 type name schema',
)  # fmt: skip
PARSER_srcstore = PARSER.add_argument_group(
    'Default source typestore, if sources do not include types',
).add_mutually_exclusive_group()  # fmt: skip
PARSER_topics = PARSER.add_argument_group(
    'Select topics to exclude or include',
)  # fmt: skip
PARSER.add_argument(
    '--src',
    action='extend',
    nargs='*',
    metavar='PATH',
    help='Rosbag files to read from.',
    dest='srcs',
    required=True,
    type=Path,
)
PARSER.add_argument(
    '--dst',
    help='Destination path to write rosbag to.',
    dest='dst',
    required=True,
    type=Path,
)
PARSER.add_argument(
    '--dst-storage',
    choices=['sqlite3', 'mcap'],
    help="Destination file storage backend. (default='sqlite3')",
    dest='dst_storage',
    type=str,
)
PARSER.add_argument(
    '--dst-version',
    help='Destination file format version. (default=9)',
    dest='dst_version',
    type=int,
)
PARSER.add_argument(
    '--compress',
    choices=['none', 'bz2', 'lz4', 'zstd'],
    help="Compression algorithm. Rosbag1 supports 'bz2' or 'lz4'. Rosbag2 supports 'zstd'. (default='none')",
    dest='compress',
    type=str,
)
PARSER.add_argument(
    '--compress-mode',
    choices=['file', 'message', 'storage'],
    help="Compression mode for rosbag2. (default='file')",
    dest='compress_mode',
    type=str,
)
PARSER_srcstore.add_argument(
    '--src-typestore',
    choices=['empty', 'latest', 'ros1_noetic', 'ros2_dashing', 'ros2_eloquent', 'ros2_foxy', 'ros2_galactic', 'ros2_humble', 'ros2_iron', 'ros2_jazzy', 'ros2_kilted', 'ros2_lyrical'],
    help="Source typestore name. (default='ros2_foxy')",
    dest='src_typestore',
    type=str,
)
PARSER_srcstore.add_argument(
    '--src-typestore-ref',
    help="Source typestore import location. (default='')",
    dest='src_typestore_ref',
    type=str,
)
PARSER_dststore.add_argument(
    '--dst-typestore',
    choices=['copy', 'empty', 'latest', 'ros1_noetic', 'ros2_dashing', 'ros2_eloquent', 'ros2_foxy', 'ros2_galactic', 'ros2_humble', 'ros2_iron', 'ros2_jazzy', 'ros2_kilted', 'ros2_lyrical'],
    help="Destination typestore name. (default='copy')",
    dest='dst_typestore',
    type=str,
)
PARSER_dststore.add_argument(
    '--dst-typestore-ref',
    help="Destination typestore import location. (default='')",
    dest='dst_typestore_ref',
    type=str,
)
PARSER_topics.add_argument(
    '--exclude-topic',
    action='extend',
    nargs='*',
    metavar='TOPIC',
    help='Topics to exclude from conversion, even if included explicitly. (default=())',
    dest='exclude_topics',
    type=str,
)
PARSER_topics.add_argument(
    '--include-topic',
    action='extend',
    nargs='*',
    metavar='TOPIC',
    help='Topics to include in conversion, instead of all. (default=())',
    dest='include_topics',
    type=str,
)
PARSER_msgtypes.add_argument(
    '--exclude-msgtype',
    action='extend',
    nargs='*',
    metavar='MSGTYPE',
    help='Message types to exclude from conversion, even if included explicitly. (default=())',
    dest='exclude_msgtypes',
    type=str,
)
PARSER_msgtypes.add_argument(
    '--include-msgtype',
    action='extend',
    nargs='*',
    metavar='MSGTYPE',
    help='Message types to include in conversion, instead of all. (default=())',
    dest='include_msgtypes',
    type=str,
)

# fmt: on
def custom_complete(action: argparse.Action, args: list[str], arg: str) -> list[str]:
    """Complete action."""
    ret: list[str] = []
    addargs: dict[str, str] = {}

    return ret


COMPLETER_BASH = """\
_completer__%(name)s() {
    local IFS=$'\\n'
    for reply in $(env _COMPLETER=bash COMP_LINE="$COMP_LINE" COMP_POINT=$COMP_POINT $1); do
        IFS=',' read type value descr <<< "$reply"
        if [[ $type == "directory" ]]; then
            compopt -o dirnames
        elif [[ $type == "file" ]]; then
            compopt -o default
        elif [[ $type == "string" ]]; then
            COMPREPLY+=($value)
        fi
    done
    return 0
}
complete -o nosort -F _completer__%(name)s %(name)s;
"""

COMPLETER_ZSH = """\
#compdef %(name)s
_completer__%(name)s() {
    local -a values values_descrs
    (( ! $+commands[%(name)s] )) && return 1
    for reply in "${(@f)$(env _COMPLETER=zsh COMP_LINE="$BUFFER" COMP_POINT=$CURSOR %(name)s)}"; do
       IFS="," read type value descr <<< "$reply"
       if [[ "$type" == "directory" ]]; then
           _path_files -/
       elif [[ "$type" == "file" ]]; then
           _path_files -f
       elif [[ "$type" == "string" ]]; then
           if [[ -n "$descr" ]]; then
               values_descrs+=("$value":"$descr")
           else
               values+=("$value")
           fi
       fi
    done
    if [ -n "$values_descrs" ]; then
        _describe -V unsorted values_descrs -U
    fi
    if [ -n "$values" ]; then
        compadd -U -V unsorted -a values
    fi
}
compdef _completer__%(name)s %(name)s;
"""


def generate_source_completion() -> None:
    """Generate snippet for sourcing from shell."""
    completer = os.getenv('_COMPLETER')
    assert completer
    completers = {
        'bash': COMPLETER_BASH,
        'zsh': COMPLETER_ZSH,
    }
    print(completers[completer] % {'name': Path(sys.argv[0]).name})


def consume_action(action: argparse.Action, arg: str) -> None:
    """Consume arg in action."""
    if action.choices and arg not in action.choices:
        raise ValueError


def consume_parser(
    parser: ArgumentParser,
    excluded: list[argparse.Action],
    arg: str,
) -> tuple[ArgumentParser, argparse.Action | None]:
    """Consume parser."""
    for action in parser._actions:
        if isinstance(action, argparse._SubParsersAction):
            for option, subaction in action.choices.items():
                if option == arg:
                    return subaction, None

        if arg.startswith('-'):
            if any(x == arg for x in action.option_strings):
                break
        else:
            if not action.option_strings:
                break
    else:
        raise ValueError

    if not isinstance(action, argparse._CountAction) and not (
        isinstance(action.nargs, str) and action.nargs in '*+'
    ):
        excluded.append(action)
    for group in parser._mutually_exclusive_groups:
        if action in group._group_actions:
            for subaction in group._group_actions:
                if subaction != action and subaction not in excluded:
                    excluded.append(subaction)
    return parser, action if (
        action.option_strings or action.choices
    ) and action.nargs != 0 else None


def complete_actions(
    actions: list[argparse.Action],
    excluded: list[argparse.Action],
    args: list[str],
    arg: str,
) -> None:
    """Complete actions."""
    completed: list[tuple[str, str]] = []
    for action in sorted(actions, key=lambda x: not x.required):
        if action in excluded:
            continue

        if isinstance(action, argparse._SubParsersAction):
            completed += [
                (k, v.description)
                for k, v in action.choices.items()
                if not v.description.startswith('SUPPRESS.')
            ]

        elif action.option_strings:
            completed += [(x, action.help or '') for x in action.option_strings]

        elif action.choices:
            completed += [(x, '') for x in action.choices]

        else:
            completed += [(x, '') for x in custom_complete(action, args, arg)]

        if not arg and action.required:
            break

    have_positinal = any(not x.startswith('-') for x, _ in completed)
    for option, descr in sorted(completed):
        if have_positinal and not arg and option.startswith('-'):
            continue
        if option.startswith(arg):
            print(f'string,{option},{descr}')


def complete_action(action: argparse.Action, args: list[str], arg: str) -> None:
    """Complete action."""
    completed: list[tuple[str, str]] = []

    if action.type == Path:
        print('file,,')

    if action.choices:
        if isinstance(action.choices, dict):
            completed += action.choices.items()
        else:
            completed += [(x, '') for x in action.choices]

    completed += [(x, '') for x in custom_complete(action, args, arg)]

    for option, descr in sorted(completed):
        if not arg and option.startswith('-'):
            continue
        if option.startswith(arg):
            print(f'string,{option},{descr}')


def generate_completion() -> None:
    """Generate snippet for sourcing from shell."""
    import shlex

    lexer = shlex.shlex(os.getenv('COMP_LINE'), posix=True)
    pos = int(os.getenv('COMP_POINT', '0'))
    lexer.whitespace_split = True
    lexer.commenters = ''
    args = []
    cword = None

    next(lexer)
    laststate = lexer.state  # type: ignore[attr-defined]
    try:
        for index, token in enumerate(lexer):
            args.append(token)
            laststate = lexer.state  # type: ignore[attr-defined]
            if cword is None and lexer.instream.tell() > pos:  # type: ignore[attr-defined]
                cword = index
    except ValueError:
        args.append(lexer.token)
        laststate = 'error'
    if laststate == ' ':
        args.append('')
    if cword is None:
        cword = len(args) - 1

    parser = PARSER
    action = None
    excluded: list[argparse.Action] = []

    # interpret_args(parser, action, excluded, args, cword)
    for arg in args[:cword]:
        # print('string,try', arg)
        if action:
            try:
                consume_action(action, arg)
                action = None
            except ValueError:
                sys.exit(0)
            continue

        try:
            parser, action = consume_parser(parser, excluded, arg)
        except ValueError:
            sys.exit(0)

    arg = args[cword]

    if action:
        complete_action(action, args, arg)
        sys.exit(0)

    complete_actions(parser._actions, excluded, args, arg)
    sys.exit(0)


def main() -> NoReturn:  # pragma: no cover
    """CLI entrypoint."""
    if 'COMP_LINE' in os.environ:
        generate_completion()
        sys.exit(0)

    if '_COMPLETER' in os.environ:
        generate_source_completion()
        sys.exit(0)
    args = PARSER.parse_args().__dict__

    from rosbags.convert.commands import command

    runargs = {
        k: v for k, v in args.items()
        if k in {
            'srcs',
            'dst',
            'dst_storage',
            'dst_version',
            'compress',
            'compress_mode',
            'src_typestore',
            'src_typestore_ref',
            'dst_typestore',
            'dst_typestore_ref',
            'exclude_topics',
            'include_topics',
            'exclude_msgtypes',
            'include_msgtypes',
        }
    }  # fmt: skip
    sys.exit(command(**runargs))


if __name__ == '__main__':
    main()