def _parse()

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

• 33 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:
            # input feature is already lookuped
            feat = input_data[self.name]
            if self.is_sparse:
                parsed_feat = _parse_fg_encoded_sparse_feature_impl(
                    self.name, feat, **self._fg_encoded_kwargs
                )
            else:
                parsed_feat = _parse_fg_encoded_dense_feature_impl(
                    self.name, feat, **self._fg_encoded_kwargs
                )
        elif self.fg_mode == FgMode.FG_NORMAL:
            inputs = copy.copy(self.inputs)
            input_feats = [input_data[inputs.pop(0)].cast(pa.string()).tolist()]
            if not self._wildcard_pkey:
                input_feats.append(input_data[inputs.pop(0)].cast(pa.string()).tolist())
            else:
                input_feats.append([])
            if not self._wildcard_skey:
                input_feats.append(input_data[inputs.pop(0)].cast(pa.string()).tolist())
            else:
                input_feats.append([])
            if self.is_sparse:
                values, lengths = self._fg_op.to_bucketized_jagged_tensor(*input_feats)
                parsed_feat = SparseData(name=self.name, values=values, lengths=lengths)
            else:
                values = self._fg_op.transform(*input_feats)
                parsed_feat = DenseData(name=self.name, values=values)
        else:
            raise ValueError(
                f"fg_mode: {self.fg_mode} is not supported without fg handler."
            )
        return parsed_feat