class_docstr: States of the upstream tis for a specific ti.
  
  This is used to determine whether the specific ti can run in this iteration.
class_docstr_tok: ["States","of","the","upstream","tis","for","a","specific","ti",".","This","is","used","to","determine","whether","the","specific","ti","can","run","in","this","iteration","."]
docstr: Calculate states for a task instance.
  
  ``counter`` is inclusive of ``setup_counter`` -- e.g. if there are 2 skipped upstreams, one
  of which is a setup, then counter will show 2 skipped and setup counter will show 1.
  
  :param ti: the ti that we want to calculate deps for
  :param finished_tis: all the finished tasks of the dag_run
docstr_tok: ["Calculate","states","for","a","task","instance",".","``","counter","``","is","inclusive","of","``","setup_counter","``","--","e.g",".","if","there","are","2","skipped","upstreams",",","one","of","which","is","a","setup",",","then","counter","will","show","2","skipped","and","setup","counter","will","show","1.",":","param","ti",":","the","ti","that","we","want","to","calculate","deps","for",":","param","finished_tis",":","all","the","finished","tasks","of","the","dag_run"]
code: def calculate(
      cls, finished_upstreams: Iterator[TaskInstance]
  ) -> _UpstreamTIStates:
      """Calculate states for a task instance.
  
      ``counter`` is inclusive of ``setup_counter`` -- e.g. if there are 2 skipped upstreams, one
      of which is a setup, then counter will show 2 skipped and setup counter will show 1.
  
      :param ti: the ti that we want to calculate deps for
      :param finished_tis: all the finished tasks of the dag_run
      """
      counter: dict[str, int] = Counter()
      setup_counter: dict[str, int] = Counter()
      for ti in finished_upstreams:
          curr_state = {ti.state: 1}
          counter.update(curr_state)
          if ti.task.is_setup:
              setup_counter.update(curr_state)
      return _UpstreamTIStates(
          success=counter.get(TaskInstanceState.SUCCESS, 0),
          skipped=counter.get(TaskInstanceState.SKIPPED, 0),
          failed=counter.get(TaskInstanceState.FAILED, 0),
          upstream_failed=counter.get(
              TaskInstanceState.UPSTREAM_FAILED, 0
          ),
          removed=counter.get(TaskInstanceState.REMOVED, 0),
          done=sum(counter.values()),
          success_setup=setup_counter.get(TaskInstanceState.SUCCESS, 0),
          skipped_setup=setup_counter.get(TaskInstanceState.SKIPPED, 0),
      )
  
file_path: airflow/airflow/ti_deps/deps/trigger_rule_dep.py
imports: from __future__ import annotations
  import collections.abc
  import functools
  from collections import Counter
  from typing import TYPE_CHECKING, Iterator, KeysView, NamedTuple
  from sqlalchemy import and_, func, or_, select
  from airflow.models.taskinstance import PAST_DEPENDS_MET
  from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
  from airflow.utils.state import TaskInstanceState
  from airflow.utils.trigger_rule import TriggerRule
label_desc: Use image node_id 1 for calling the _UpstreamTIStates obj's underlying member method code with example usage: obj.calculate(cls, finished_upstreams) and returns: _UpstreamTIStates