pyfsd ¶

def all_ATC_checker(_: Optional[Client], to_client: Client) -> bool:  # noqa: N802
    """A broadcast checker which only broadcast to ATC.

    Parameters:
        to_client: The dest client.

    Returns:
        The check result (send message to to_client or not).
    """
    return to_client.is_controller

def all_pilot_checker(_: Optional[Client], to_client: Client) -> bool:
    """A broadcast checker which only broadcast to pilot.

    Parameters:
        to_client: The dest client.

    Returns:
        The check result (send message to to_client or not).
    """
    return not to_client.is_controller

def at_checker(from_client: Optional[Client], to_client: Client) -> bool:
    """A broadcast checker which checks visual range when dest startswith @.

    Parameters:
        from_client: The from client.
        to_client: The dest client.

    Returns:
        The check result (send message to to_client or not).
    """
    if from_client is None:
        raise RuntimeError("at_checker needs from_client")
    if not from_client.position_ok or not to_client.position_ok:
        return False
    distance = calc_distance(from_client.position, to_client.position)
    return distance < from_client.get_range()

def broadcast_checkers(*checkers: BroadcastChecker) -> BroadcastChecker:
    """Create a set of broadcast.

    Parameters:
        checkers: The broadcast checkers.

    Returns:
        The broadcast checker.
    """

    def checker(from_client: Optional[Client], to_client: Client) -> bool:
        return all(checker(from_client, to_client) for checker in checkers)

    return checker

def broadcast_message_checker(from_client: Optional[Client], to_client: Client) -> bool:
    """A broadcast checker which checks visual range while broadcasting message.

    Parameters:
        from_client: The from client.
        to_client: The dest client.

    Returns:
        The check result (send message to to_client or not).
    """
    if from_client is None:
        raise RuntimeError("broadcast_message_checker needs from_client")
    if not from_client.position_ok or not to_client.position_ok:
        return False
    visual_range: int
    x: int = to_client.get_range()
    y: int = from_client.get_range()
    if (not from_client.is_controller) and (not to_client.is_controller):
        visual_range = x + y
    else:
        visual_range = max(y, x)
    distance = calc_distance(from_client.position, to_client.position)
    return distance < visual_range

def broadcast_position_checker(
    from_client: Optional[Client],
    to_client: Client,
) -> bool:
    """A broadcast checker which checks visual range while broadcasting position.

    Parameters:
        from_client: The from client.
        to_client: The dest client.

    Returns:
        The check result (send message to to_client or not).
    """
    if from_client is None:
        raise RuntimeError("broadcast_position_checker needs from_client")
    if not from_client.position_ok or not to_client.position_ok:
        return False
    visual_range: int
    x: int = to_client.get_range()
    y: int = from_client.get_range()
    if to_client.is_controller:
        visual_range = to_client.visual_range
    elif not from_client.is_controller:
        visual_range = x + y
    else:
        visual_range = max(x, y)
    distance = calc_distance(from_client.position, to_client.position)
    return distance < visual_range

def create_broadcast_range_checker(visual_range: int) -> BroadcastChecker:
    """Create a broadcast checker which checks visual range.

    Parameters:
        visual_range: Visual range.

    Returns:
        The broadcast checker.
    """

    def checker(from_client: Optional[Client], to_client: Client) -> bool:
        if from_client is None:
            raise RuntimeError("broadcast_range_checker needs from_client")
        if not from_client.position_ok or not to_client.position_ok:
            return False
        distance = calc_distance(from_client.position, to_client.position)
        return distance < visual_range

    return checker

def is_multicast(callsign: str) -> bool:
    """Determine if dest callsign is multicast sign.

    Parameters:
        callsign: The dest callsign.

    Returns:
        Is multicast or not.
    """
    return callsign in {"*", "*A"} or (callsign == "*P" or callsign.startswith("@"))

def assert_dict(
    dict_obj: dict,
    structure: DictStructure,
    *,
    name: str = "dict",
    allow_extra_keys: bool = False,
) -> None:
    """Wrapper of check_dict, which raises once an error is generated.

    Check type of a dict accord TypedDict.

    Args:
        dict_obj: The dict to be checked.
        structure: Expected type.
        name: Name of the dict.
        allow_extra_keys: Allow extra keys in dict_obj or not.

    Raises:
        VerifyTypeError: When found type error.
        VerifyKeyError: When found a type error about key.
        TypeError: When an unsupported/invalid type passed.
    """
    try:
        error = next(
            iter(
                check_dict(
                    dict_obj,
                    structure,
                    name=name,
                    allow_extra_keys=allow_extra_keys,
                )
            )
        )
    except StopIteration:
        pass
    else:
        raise error

def assert_simple_type(
    obj: object,
    typ: TypeHint,
    name: str = "object",
) -> None:
    """Wrapper of check_simple_type, but raise first error.

    Simple runtime type checker, supports Union, Literal, List, Dict.

    Args:
        obj: The object to be verified.
        typ: The expected type. Union, Literal, List, Dict or runtime checkable type
        name: Name of the object.

    Raises:
        VerifyTypeError: When a type error detected.
        TypeError: When an unsupported type is specified.
    """
    try:
        error = next(iter(check_simple_type(obj, typ, name)))
    except StopIteration:
        pass
    else:
        raise error

def check_dict(
    dict_obj: dict,
    structure: DictStructure,
    *,
    name: str = "dict",
    allow_extra_keys: bool = False,
) -> Iterable[Union[VerifyTypeError, VerifyKeyError]]:
    """Check type of a dict accord TypedDict.

    Args:
        dict_obj: The dict to be checked.
        structure: Expected type.
        name: Name of the dict.
        allow_extra_keys: Allow extra keys in dict_obj or not.

    Yields:
        Detected type error, in VerifyTypeError / VerifyKeyError

    Raises:
        TypeError: When an unsupported/invalid type passed.

    Examples:
        >>> class AType(TypedDict):
        ...     a: int
        ...
        >>> list(check_dict({ "a": 114514 }, AType, allow_extra_keys=False))
        []
        >>> list(check_dict({ "a": "" }, AType, allow_extra_keys=False, name="mything"))
        [VerifyTypeError("mything['a']", <class 'int'>, '')]
        >>> list(check_dict({}, AType))
        [VerifyKeyError('dict', 'a', 'missing')]
        >>> list(check_dict(
        ...     { "a": 114514, "b": 1919810 },
        ...     AType,
        ...     allow_extra_keys=True
        ... ))
        []
        >>> list(check_dict(
        ...     { "a": 114514, "b": 1919810 },
        ...     AType,
        ...     allow_extra_keys=False
        ... ))
        [VerifyKeyError('dict', 'b', 'extra')]
    """

    def deal_dict_not_required(
        dic: Mapping,
    ) -> Iterable[tuple[Hashable, Union[TypeHint, DictStructure]]]:
        for key, typ in dic.items():
            if get_origin(typ) in (NotRequired, NotRequired_ext):
                yield key, get_args(typ)[0]
            else:
                yield key, typ

    left_keys = list(dict_obj.keys())
    required_keys = tuple(lookup_required(structure))
    # New get_type_hints will change NotRequired[...] into ...
    for key, type_ in (
        new_get_type_hints(structure).items()
        if is_typeddict(structure)
        else deal_dict_not_required(structure)  # type: ignore[arg-type]
    ):
        try:
            value = dict_obj[key]
        except KeyError:
            if key in required_keys:
                yield VerifyKeyError(name, key, "missing")
            continue
        else:
            if not allow_extra_keys:
                left_keys.remove(key)
        if is_typeddict(type_) or isinstance(type_, dict):
            if not isinstance(value, dict):
                yield VerifyTypeError(f"{name}[{key!r}]", type_, value)
            else:
                yield from check_dict(
                    value,
                    type_,  # type: ignore[arg-type]
                    name=f"{name}[{key!r}]",
                    allow_extra_keys=allow_extra_keys,
                )
        else:
            yield from check_simple_type(value, type_, name=f"{name}[{key!r}]")
    if not allow_extra_keys and left_keys:
        for left_key in left_keys:
            yield VerifyKeyError(name, left_key, "extra")

def check_simple_type(
    obj: object,
    typ: TypeHint,
    name: str = "object",
) -> Iterable[VerifyTypeError]:
    """Simple runtime type checker, supports Union, Literal, List, Dict.

    Args:
        obj: The object to be verified.
        typ: The expected type. Union, Literal, List, Dict or runtime checkable type
        name: Name of the object.

    Yields:
        When a type error was detected.

    Raises:
        TypeError: When an unsupported type is specified.
    """
    if type_origin := get_origin(typ):  # elif (t_o is not None)
        if type_origin is Union:
            for sub_type in get_args(typ):
                if is_empty_iterable(check_simple_type(obj, sub_type, name=name)):
                    return
            yield VerifyTypeError(name, typ, obj)
        elif type_origin is Literal:
            if obj not in get_args(typ):
                yield VerifyTypeError(name, typ, obj)
        elif type_origin is list:
            if not isinstance(obj, list):
                yield VerifyTypeError(name, typ, obj)
                return
            for i, val in enumerate(obj):
                yield from check_simple_type(
                    val,
                    get_args(typ)[0],
                    name=f"{name}[{i}]",
                )
        elif type_origin is dict:
            if not isinstance(obj, dict):
                yield VerifyTypeError(name, typ, obj)
                return
            key_type, value_type = get_args(typ)
            for key, value in obj.items():
                # TODO: Better description of key
                yield from check_simple_type(
                    key,
                    key_type,
                    name=f"{name}[{key!r}]",
                )
                yield from check_simple_type(
                    value,
                    value_type,
                    name=f"{name}[{key!r}]",
                )
        else:
            raise TypeError(f"Unsupported type: {type_origin!r}")
    elif isinstance(typ, type):
        if not isinstance(obj, typ):
            yield VerifyTypeError(name, typ, obj)
    else:
        raise TypeError(f"Invalid type: {typ!r}")

def explain_type(typ: TypeHint) -> str:
    """Explain a type.

    Args:
        typ: The type to be explained.

    Returns:
        Description of the type.

    Raises:
        TypeError: When an unsupported/invalid type passed.
    """
    if isinstance(typ, dict) or is_typeddict(typ):
        return "dict"
    if type_origin := get_origin(typ):  # elif (t_o is not None)
        if type_origin is Union:
            return " or ".join(explain_type(sub_type) for sub_type in get_args(typ))
        if type_origin is Literal:
            return " or ".join(repr(sub_value) for sub_value in get_args(typ))
        if type_origin in (list, dict):
            return str(typ).removeprefix(typ.__module__ + ".")
        raise TypeError(f"Unsupported type: {type_origin!r}")
    if isinstance(typ, type):
        return typ.__name__
    raise TypeError(f"Invalid type: {typ!r}")

def lookup_required(structure: DictStructure) -> Iterable[Hashable]:
    """Yields all required key in a TypedDict.

    Args:
        structure: The type structure, TypedDict or dict.

    Yields:
        Keys that are required, str normally.
    """
    if is_typeddict(structure):
        # Python < 3.9 not supported
        # ---------
        # Mypy bug, ignore it
        if not structure.__total__:  # type: ignore[union-attr]
            # Nothing is required
            return
        if NotRequired.__module__ == "typing":  # Python 3.11+, not need to Workaround
            yield from structure.__required_keys__  # type: ignore[union-attr]
            return
        # Python 3.9, 3.10
        type_hints = get_type_hints(structure)
        for may_required_keys in structure.__required_keys__:  # type: ignore[union-attr]
            if get_origin(type_hints[may_required_keys]) not in (
                NotRequired,
                NotRequired_ext,
            ):
                yield may_required_keys
    else:
        for may_required_keys, type_ in structure.items():  # type: ignore[union-attr]
            if get_origin(type_) not in (NotRequired, NotRequired_ext):
                yield may_required_keys

def break_packet(
    packet: AnyStr,
    possibly_commands: Iterable[Union[AnyStr, FSDClientCommand]],
) -> tuple[Optional[Union[AnyStr, FSDClientCommand]], tuple[AnyStr, ...]]:
    """Break a packet into command and parts.

        #APzzzzzzzzzzzz1:zzzzzzz3:zzzzzzz4
        [^][^^^^^^^^^^^] [^^^^^^] [^^^^^^]
        command parts[0] parts[1] parts[2]

    Args:
        packet: The original packet.
        possibly_commands: All possibly commands. This function will check if packet \
            starts with one of possibly commands then split it out.

    Returns:
        tuple[command or None, tuple[every_part, ...]]
    """
    packet_type = type(packet)
    command: Optional[Union[AnyStr, FSDClientCommand]] = None
    splited_packet: list[AnyStr]
    for possibly_command in possibly_commands:
        command_str: AnyStr
        if isinstance(possibly_command, FSDClientCommand):
            command_str = possibly_command.as_type(packet_type)
        else:
            command_str = possibly_command
        if packet.startswith(command_str):
            command = possibly_command
            break
    splited_packet = packet.split(  # pyright: ignore[reportAssignmentType]
        SPLIT_SIGN.as_type(packet_type)
    )
    if command is not None:
        splited_packet[0] = splited_packet[0][len(command) :]
    return (command, tuple(splited_packet))

def join_lines(*lines: AnyStr, newline: bool = True) -> AnyStr:
    r"""Join lines together.

    Args:
        lines: The lines.
        newline: Append '\r\n' to every line or not.

    Returns:
        The result.
    """
    result = CompatibleString("")
    split_sign = CompatibleString("\r\n")
    for line in lines:
        # Ignore type errors. Just let it raise.
        result += line + split_sign if newline else line  # type: ignore[assignment]
    return cast("AnyStr", result)

def make_packet(*parts: Union[AnyStr, FSDClientCommand]) -> AnyStr:
    """Join parts together and add split sign between every two parts."""
    result = CompatibleString("")
    for part in parts:
        result += part + SPLIT_SIGN  # type: ignore[assignment]
    if isinstance(result, FSDClientCommand):
        raise TypeError("Must have str or bytes item")
    return cast("AnyStr", result[:-1])