abydos.distance._aline.ALINE.alignments()

Complexity

Conditions

71

Size

Total Lines	398
Code Lines	280

Duplication

Lines	78
Ratio	19.6 %

Code Coverage

Tests	175
CRAP Score	71

Importance

Changes

0

# Copyright 2019-2020 by Christopher C. Little.
    def alignments(
        self, src: str, tar: str, score_only: bool = False
    ) -> Union[float, List[Tuple[float, str, str]]]:
        """Return the ALINE alignments of two strings.

        Parameters
        ----------
        src : str
            Source string for comparison
        tar : str
            Target string for comparison
        score_only : bool
            Return the score only, not the alignments

        Returns
        -------
        list(tuple(float, str, str) or float
            ALINE alignments and their scores or the top score

        Examples
        --------
        >>> cmp = ALINE()
        >>> cmp.alignments('cat', 'hat')
        [(50.0, 'c ‖ a t ‖', 'h ‖ a t ‖')]
        >>> cmp.alignments('niall', 'neil')
        [(90.0, '‖ n i a ll ‖', '‖ n e i l  ‖')]
        >>> cmp.alignments('aluminum', 'catalan')
        [(81.5, '‖ a l u m ‖ inum', 'cat ‖ a l a n ‖')]
        >>> cmp.alignments('atcg', 'tagc')
        [(65.0, '‖ a t c ‖ g', 't ‖ a g c ‖'), (65.0, 'a ‖ tc - g ‖',
        '‖ t  a g ‖ c')]


        .. versionadded:: 0.4.0
        .. versionchanged:: 0.4.1
            Renamed from .alignment to .alignments

        """

        def _sig_skip(*args: Any) -> float:
            return self._c_skip

        def _sig_sub(seg1: Dict[str, float], seg2: Dict[str, float]) -> float:
            return (
                self._c_sub
                - _delta(seg1, seg2)
                - _sig_vwl(seg1)
                - _sig_vwl(seg2)
            )

        def _sig_exp(
            seg1: Dict[str, float],
            seg2a: Dict[str, float],
            seg2b: Dict[str, float],
        ) -> float:
            return (
                self._c_exp
                - _delta(seg1, seg2a)
                - _delta(seg1, seg2b)
                - _sig_vwl(seg1)
                - max(_sig_vwl(seg2a), _sig_vwl(seg2b))
            )

        def _sig_vwl(seg1: Dict[str, float]) -> float:
            return (
                0.0
                if seg1['manner'] > self.feature_weights['high vowel']
                else self._c_vwl
            )

        def _delta(seg1: Dict[str, float], seg2: Dict[str, float]) -> float:
            features = (
                self.c_features
                if max(seg1['manner'], seg2['manner'])
                > self.feature_weights['high vowel']
                else self.v_features
            )
            diff = 0.0
            for f in features:
                diff += (
                    abs(seg1.get(f, 0.0) - seg2.get(f, 0.0)) * self.salience[f]
                )
            return diff

        def _retrieve(
            i: int, j: int, score: float, out: List[Tuple[str, str]]
        ) -> None:
            def _record(score: float, out: List[Tuple[str, str]]) -> None:
                out.append(('‖', '‖'))
                for i1 in range(i - 1, -1, -1):
                    out.append((src_tok[i1], ''))
                for j1 in range(j - 1, -1, -1):
                    out.append(('', tar_tok[j1]))
                if self._mode == 'global':
                    score += (i + j) * _sig_skip('')

                out = out[::-1]

                src_alignment = []
                tar_alignment = []

                out.append(('‖', '‖'))
                part = 0
                s_segment = ''  # type: Union[str, List[str]]
                t_segment = ''  # type: Union[str, List[str]]
                for ss, ts in out:
                    if ss == '‖':
                        if part % 2 == 0:
                            src_alignment.append(s_segment)
                            tar_alignment.append(t_segment)
                            s_segment = []
                            t_segment = []
                        else:
                            src_alignment.append(' '.join(s_segment))
                            tar_alignment.append(' '.join(t_segment))
                            s_segment = ''
                            t_segment = ''
                        part += 1
                    else:
                        if part % 2 == 0:
                            s_segment = cast(str, s_segment) + ss
                            t_segment = cast(str, t_segment) + ts
                        else:
                            cast(List[str], s_segment).append(
                                ss + ' ' * (len(ts) - len(ss))
                            )
                            cast(List[str], t_segment).append(
                                ts + ' ' * (len(ss) - len(ts))
                            )

                src_alignment_str = ' ‖ '.join(
                    cast(List[str], src_alignment)
                ).strip()
                tar_alignment_str = ' ‖ '.join(
                    cast(List[str], tar_alignment)
                ).strip()

                alignments.append(
                    (score, src_alignment_str, tar_alignment_str)
                )
                return

            if s_mat[i, j] == 0:
                _record(score, out)
                return
            else:
                if (
                    i > 0
                    and j > 0
                    and s_mat[i - 1, j - 1]
                    + _sig_sub(src_feat_wt[i - 1], tar_feat_wt[j - 1])
                    + score
                    >= threshold
                ):
                    loc_out = deepcopy(out)
                    loc_out.append((src_tok[i - 1], tar_tok[j - 1]))
                    _retrieve(
                        i - 1,
                        j - 1,
                        score
                        + _sig_sub(src_feat_wt[i - 1], tar_feat_wt[j - 1]),
                        loc_out,
                    )
                    loc_out.pop()

                if (

This code seems to be duplicated in your project.

                    j > 0
                    and s_mat[i, j - 1] + _sig_skip(tar_tok[j - 1]) + score
                    >= threshold
                ):
                    loc_out = deepcopy(out)
                    loc_out.append(('-', tar_tok[j - 1]))
                    _retrieve(
                        i, j - 1, score + _sig_skip(tar_tok[j - 1]), loc_out
                    )
                    loc_out.pop()

                if (

This code seems to be duplicated in your project.

                    i > 0
                    and j > 1
                    and s_mat[i - 1, j - 2]
                    + _sig_exp(
                        src_feat_wt[i - 1],
                        tar_feat_wt[j - 2],
                        tar_feat_wt[j - 1],
                    )
                    + score
                    >= threshold
                ):
                    loc_out = deepcopy(out)
                    loc_out.append(
                        (src_tok[i - 1], tar_tok[j - 2] + tar_tok[j - 1],)
                    )
                    _retrieve(
                        i - 1,
                        j - 2,
                        score
                        + _sig_exp(
                            src_feat_wt[i - 1],
                            tar_feat_wt[j - 2],
                            tar_feat_wt[j - 1],
                        ),
                        loc_out,
                    )
                    loc_out.pop()

                if (

This code seems to be duplicated in your project.

                    i > 0
                    and s_mat[i - 1, j] + _sig_skip(src_tok[i - 1]) + score
                    >= threshold
                ):
                    loc_out = deepcopy(out)
                    loc_out.append((src_tok[i - 1], '-'))
                    _retrieve(
                        i - 1, j, score + _sig_skip(src_tok[i - 1]), loc_out
                    )
                    loc_out.pop()

                if (

This code seems to be duplicated in your project.

                    i > 1
                    and j > 0
                    and s_mat[i - 2, j - 1]
                    + _sig_exp(
                        tar_feat_wt[j - 1],
                        src_feat_wt[i - 2],
                        src_feat_wt[i - 1],
                    )
                    + score
                    >= threshold
                ):
                    loc_out = deepcopy(out)
                    loc_out.append(
                        (src_tok[i - 2] + src_tok[i - 1], tar_tok[j - 1],)
                    )
                    _retrieve(
                        i - 2,
                        j - 1,
                        score
                        + _sig_exp(
                            tar_feat_wt[j - 1],
                            src_feat_wt[i - 2],
                            src_feat_wt[i - 1],
                        ),
                        loc_out,
                    )
                    loc_out.pop()

        sg_max = 0.0

        src_tok = []  # type: List[str]
        src_feat = []  # type: List[Dict[str, str]]
        tar_tok = []  # type: List[str]
        tar_feat = []  # type: List[Dict[str, str]]

        for ch in src:
            if ch in self._phones:
                src_tok.append(ch)
                src_feat.append(dict(self._phones[ch]))
        for ch in tar:
            if ch in self._phones:
                tar_tok.append(ch)
                tar_feat.append(dict(self._phones[ch]))

        for i in range(1, len(src_feat)):
            if 'supplemental' in src_feat[i]:
                j = i - 1
                while j > -1:
                    if 'supplemental' not in src_feat[j]:
                        src_tok[j] += src_tok[i]
                        for key, value in src_feat[i].items():
                            if key != 'supplemental':
                                src_feat[j][key] = value
                        j = 0
                    j -= 1

        zipped = [
            fb for fb in zip(src_feat, src_tok) if 'supplemental' not in fb[0]
        ]
        if zipped:
            src_feat, src_tok = zip(*zipped)  # type: ignore
        else:
            src_feat, src_tok = [], []

        src_feat_wt = []  # type: List[Dict[str, float]]
        for f_dict in src_feat:
            src_feat_wt.append(
                {
                    key: self.feature_weights[f_dict[key]]
                    for key in f_dict.keys()
                }
            )

        src_len = len(src_tok)

        for i in range(1, len(tar_feat)):
            if 'supplemental' in tar_feat[i]:
                j = i - 1
                while j > -1:
                    if 'supplemental' not in tar_feat[j]:
                        tar_tok[j] += tar_tok[i]
                        for key, value in tar_feat[i].items():
                            if key != 'supplemental':
                                tar_feat[j][key] = value
                        j = 0
                    j -= 1

        zipped = [
            fb for fb in zip(tar_feat, tar_tok) if 'supplemental' not in fb[0]
        ]
        if zipped:
            tar_feat, tar_tok = zip(*zipped)  # type: ignore
        else:
            tar_feat, tar_tok = [], []

        tar_feat_wt = []  # type: List[Dict[str, float]]
        for f_dict in tar_feat:
            tar_feat_wt.append(
                {
                    key: self.feature_weights[f_dict[key]]
                    for key in f_dict.keys()
                }
            )

        tar_len = len(tar_tok)

        s_mat = zeros((src_len + 1, tar_len + 1), dtype=float_)

        if self._mode == 'global':
            for i in range(1, src_len + 1):
                s_mat[i, 0] = s_mat[i - 1, 0] + _sig_skip(src_tok[i - 1])
            for j in range(1, tar_len + 1):
                s_mat[0, j] = s_mat[0, j - 1] + _sig_skip(tar_tok[j - 1])

        for i in range(1, src_len + 1):
            for j in range(1, tar_len + 1):
                s_mat[i, j] = max(
                    s_mat[i - 1, j] + _sig_skip(src_feat_wt[i - 1]),
                    s_mat[i, j - 1] + _sig_skip(tar_feat_wt[j - 1]),
                    s_mat[i - 1, j - 1]
                    + _sig_sub(src_feat_wt[i - 1], tar_feat_wt[j - 1]),
                    s_mat[i - 1, j - 2]
                    + _sig_exp(
                        src_feat_wt[i - 1],
                        tar_feat_wt[j - 2],
                        tar_feat_wt[j - 1],
                    )
                    if j > 1
                    else -inf,
                    s_mat[i - 2, j - 1]
                    + _sig_exp(
                        tar_feat_wt[j - 1],
                        src_feat_wt[i - 2],
                        src_feat_wt[i - 1],
                    )
                    if i > 1
                    else -inf,
                    0 if self._mode in {'local', 'half-local'} else -inf,
                )

                if s_mat[i, j] > sg_max:
                    if self._mode == 'semi-global':
                        if i == src_len or j == tar_len:
                            sg_max = s_mat[i, j]
                    else:
                        sg_max = s_mat[i, j]

        if self._mode in {'global', 'half-local'}:
            dp_score = s_mat[src_len, tar_len]
        else:
            dp_score = s_mat.max()

        if score_only:
            return cast(float, dp_score)

        threshold = (1 - self._epsilon) * dp_score

        alignments = []  # type: List[Tuple[float, str, str]]

        for i in range(1, src_len + 1):
            for j in range(1, tar_len + 1):
                if self._mode in {'global', 'half-local'} and (
                    i < src_len or j < tar_len
                ):
                    continue
                if self._mode == 'semi-global' and (
                    i < src_len and j < tar_len
                ):
                    continue
                if s_mat[i, j] >= threshold:
                    out = []
                    for j1 in range(tar_len - 1, j - 1, -1):
                        out.append(('', tar_tok[j1]))
                    for i1 in range(src_len - 1, i - 1, -1):
                        out.append((src_tok[i1], ''))
                    out.append(('‖', '‖'))
                    _retrieve(i, j, 0, out)

        return sorted(alignments, key=lambda _: _[0], reverse=True)