code: def build_vocab_main(opts):
      """Apply transforms to samples of specified data and build vocab from it.
  
      Transforms that need vocab will be disabled in this.
      Built vocab is saved in plain text format as following and can be pass as
      `-src_vocab` (and `-tgt_vocab`) when training:
      ```
      <tok_0>\t<count_0>
      <tok_1>\t<count_1>
      ```
      """
  
      ArgumentParser.validate_prepare_opts(opts, build_vocab_only=True)
      assert (
          opts.n_sample == -1 or opts.n_sample > 1
      ), f"Illegal argument n_sample={opts.n_sample}."
  
      logger = init_logger()
      set_random_seed(opts.seed, False)
      transforms_cls = get_transforms_cls(opts._all_transform)
  
      if opts.learn_subwords:
          logger.info(
              f"Ingesting {opts.src_subword_type} model from corpus"
          )
          learner = make_learner(
              opts.src_subword_type, opts.learn_subwords_size
          )
          if opts.src_subword_model is not None:
              tok_path = opts.src_subword_model
          else:
              data_dir = os.path.split(opts.save_data)[0]
              if not os.path.exists(data_dir):
                  os.makedirs(data_dir)
              tok_path = os.path.join(
                  data_dir, f"{opts.src_subword_type}.model"
              )
          save_opts = copy.deepcopy(opts)
          opts.src_subword_type = "none"
          opts.tgt_subword_type = "none"
          opts.src_onmttok_kwargs["joiner_annotate"] = False
          opts.tgt_onmttok_kwargs["joiner_annotate"] = False
          transforms = make_transforms(opts, transforms_cls, None)
          ingest_tokens(opts, transforms, opts.n_sample, learner, 1, 0)
          logger.info(f"Learning {tok_path} model, patience")
          learner.learn(tok_path)
          opts = save_opts
  
      transforms = make_transforms(opts, transforms_cls, None)
  
      logger.info(f"Counter vocab from {opts.n_sample} samples.")
      src_counter, tgt_counter, src_feats_counter = build_vocab(
          opts, transforms, n_sample=opts.n_sample
      )
  
      logger.info(f"Counters src: {len(src_counter)}")
      logger.info(f"Counters tgt: {len(tgt_counter)}")
      for i, feat_counter in enumerate(src_feats_counter):
          logger.info(f"Counters src feat_{i}: {len(feat_counter)}")
  
      def save_counter(counter, save_path):
          check_path(
              save_path, exist_ok=opts.overwrite, log=logger.warning
          )
          with open(save_path, "w", encoding="utf8") as fo:
              for tok, count in counter.most_common():
                  fo.write(tok + "\t" + str(count) + "\n")
  
      if opts.share_vocab:
          src_counter += tgt_counter
          tgt_counter = src_counter
          logger.info(f"Counters after share:{len(src_counter)}")
          save_counter(src_counter, opts.src_vocab)
      else:
          save_counter(src_counter, opts.src_vocab)
          save_counter(tgt_counter, opts.tgt_vocab)
  
      for i, c in enumerate(src_feats_counter):
          save_counter(c, f"{opts.src_vocab}_feat{i}")
  
imports: import os
  import copy
  import multiprocessing
  import pyonmttok
  from functools import partial
  from onmt.utils.logging import init_logger, logger
  from onmt.utils.misc import set_random_seed, check_path
  from onmt.utils.parse import ArgumentParser
  from onmt.opts import data_prepare_opts
  from onmt.inputters.text_corpus import build_corpora_iters, get_corpora
  from onmt.inputters.text_utils import process, append_features_to_text
  from onmt.transforms import make_transforms, get_transforms_cls
  from onmt.constants import CorpusName, CorpusTask
  from collections import Counter
label_desc: Use image node_id 7 for calling a global function with example usage: build_vocab_main(opts) without return types