def _parse()

in tzrec/features/sequence_feature.py [0:0]

• 48 lines of code
• 7 McCabe index (conditional complexity)

    def _parse(self, input_data: Dict[str, pa.Array]) -> ParsedData:
        """Parse input data for the feature impl.

        Args:
            input_data (dict): raw input feature data.

        Return:
            parsed feature data.
        """
        if self.fg_mode == FgMode.FG_NONE:
            feat = input_data[self.name]
            if self.is_sparse:
                parsed_feat = _parse_fg_encoded_sequence_sparse_feature_impl(
                    self.name,
                    feat,
                    sequence_delim=self.sequence_delim,
                    **self._fg_encoded_kwargs,
                )
            else:
                parsed_feat = _parse_fg_encoded_sequence_dense_feature_impl(
                    self.name,
                    feat,
                    sequence_delim=self.sequence_delim,
                    value_dim=self.config.value_dim,
                    **self._fg_encoded_kwargs,
                )
        elif self.fg_mode == FgMode.FG_NORMAL:
            input_feats = []
            for name in self.inputs:
                x = input_data[name]
                if pa.types.is_list(x.type):
                    x = x.fill_null([])
                x = x.tolist()
                input_feats.append(x)
            if self._fg_op.is_sparse:
                values, key_lengths, seq_lengths = (
                    self._fg_op.to_bucketized_jagged_tensor(input_feats)
                )
                parsed_feat = SequenceSparseData(
                    name=self.name,
                    values=values,
                    key_lengths=key_lengths,
                    seq_lengths=seq_lengths,
                )
            else:
                values, lengths = self._fg_op.to_jagged_tensor(input_feats)
                parsed_feat = SequenceDenseData(
                    name=self.name,
                    values=values,
                    seq_lengths=lengths,
                )
        else:
            raise ValueError(
                f"fg_mode: {self.fg_mode} is not supported without fg handler."
            )
        return parsed_feat