code: def coerce_dtypes(df, dtypes):
      """Coerce dataframe to dtypes safely
  
      Operates in place
  
      Parameters
      ----------
      df: Pandas DataFrame
      dtypes: dict like {'x': float}
      """
      bad_dtypes = []
      bad_dates = []
      errors = []
      for c in df.columns:
          if c in dtypes and df.dtypes[c] != dtypes[c]:
              actual = df.dtypes[c]
              desired = dtypes[c]
              if is_float_dtype(actual) and is_integer_dtype(desired):
                  bad_dtypes.append((c, actual, desired))
              elif is_object_dtype(actual) and is_datetime64_any_dtype(
                  desired
              ):
                  # This can only occur when parse_dates is specified, but an
                  # invalid date is encountered. Pandas then silently falls back
                  # to object dtype. Since `object_array.astype(datetime)` will
                  # silently overflow, error here and report.
                  bad_dates.append(c)
              else:
                  try:
                      df[c] = df[c].astype(dtypes[c])
                  except Exception as e:
                      bad_dtypes.append((c, actual, desired))
                      errors.append((c, e))
  
      if bad_dtypes:
          if errors:
              ex = "\n".join(
                  f"- {c}\n  {e!r}"
                  for c, e in sorted(errors, key=lambda x: str(x[0]))
              )
              exceptions = (
                  "The following columns also raised exceptions on "
                  "conversion:\n\n%s\n\n"
              ) % ex
              extra = ""
          else:
              exceptions = ""
              # All mismatches are int->float, also suggest `assume_missing=True`
              extra = (
                  "\n\nAlternatively, provide `assume_missing=True` "
                  "to interpret\n"
                  "all unspecified integer columns as floats."
              )
  
          bad_dtypes = sorted(bad_dtypes, key=lambda x: str(x[0]))
          table = asciitable(
              ["Column", "Found", "Expected"], bad_dtypes
          )
          dtype_kw = "dtype={%s}" % ",\n       ".join(
              f"{k!r}: '{v}'" for (k, v, _) in bad_dtypes
          )
  
          dtype_msg = (
              "{table}\n\n"
              "{exceptions}"
              "Usually this is due to dask's dtype inference failing, and\n"
              "*may* be fixed by specifying dtypes manually by adding:\n\n"
              "{dtype_kw}\n\n"
              "to the call to `read_csv`/`read_table`."
              "{extra}"
          ).format(
              table=table,
              exceptions=exceptions,
              dtype_kw=dtype_kw,
              extra=extra,
          )
      else:
          dtype_msg = None
  
      if bad_dates:
          also = " also " if bad_dtypes else " "
          cols = "\n".join("- %s" % c for c in bad_dates)
          date_msg = (
              "The following columns{also}failed to properly parse as dates:\n\n"
              "{cols}\n\n"
              "This is usually due to an invalid value in that column. To\n"
              "diagnose and fix it's recommended to drop these columns from the\n"
              "`parse_dates` keyword, and manually convert them to dates later\n"
              "using `dd.to_datetime`."
          ).format(also=also, cols=cols)
      else:
          date_msg = None
  
      if bad_dtypes or bad_dates:
          rule = "\n\n%s\n\n" % ("-" * 61)
          msg = (
              "Mismatched dtypes found in `pd.read_csv`/`pd.read_table`.\n\n%s"
              % (rule.join(filter(None, [dtype_msg, date_msg])))
          )
          raise ValueError(msg)
  
imports: from __future__ import annotations
  import os
  from collections.abc import Mapping
  from io import BytesIO
  from warnings import catch_warnings, simplefilter, warn
  import numpy
  import pandas
  from fsspec.compression import compr
  from fsspec.core import get_fs_token_paths
  from fsspec.core import open
  from fsspec.core import open_files
  from fsspec.utils import infer_compression
  from pandas.api.types import CategoricalDtype, is_datetime64_any_dtype, is_float_dtype, is_integer_dtype, is_object_dtype
  from dask.base import tokenize
  from dask.bytes import read_bytes
  from dask.core import flatten
  from dask.dataframe.backends import dataframe_creation_dispatch
  from dask.dataframe.io.io import from_map
  from dask.dataframe.io.utils import DataFrameIOFunction
  from dask.dataframe.utils import clear_known_categories
  from dask.delayed import delayed
  from dask.utils import asciitable, parse_bytes
  from dask.dataframe.core import _Frame
label_desc: Use image node_id 2 for calling a global function with example usage: coerce_dtypes(df, dtypes) without return types