class_docstr: RawMff class.
docstr: Init the RawMff class.
docstr_tok: ["Init","the","RawMff","class","."]
code: def __init__(
      self,
      input_fname,
      eog=None,
      misc=None,
      include=None,
      exclude=None,
      preload=False,
      channel_naming="E%d",
      verbose=None,
  ):
      """Init the RawMff class."""
      input_fname = str(
          _check_fname(
              input_fname,
              "read",
              True,
              "input_fname",
              need_dir=True,
          )
      )
      logger.info("Reading EGI MFF Header from %s..." % input_fname)
      egi_info = _read_header(input_fname)
      if eog is None:
          eog = []
      if misc is None:
          misc = np.where(np.array(egi_info["chan_type"]) != "eeg")[
              0
          ].tolist()
  
      logger.info("    Reading events ...")
      egi_events, egi_info = _read_events(input_fname, egi_info)
      cals = _get_eeg_calibration_info(input_fname, egi_info)
      logger.info("    Assembling measurement info ...")
      if egi_info["n_events"] > 0:
          event_codes = list(egi_info["event_codes"])
          if include is None:
              exclude_list = (
                  ["sync", "TREV"] if exclude is None else exclude
              )
              exclude_inds = [
                  i
                  for i, k in enumerate(event_codes)
                  if k in exclude_list
              ]
              more_excludes = []
              if exclude is None:
                  for ii, event in enumerate(egi_events):
                      if event.sum() <= 1 and event_codes[ii]:
                          more_excludes.append(ii)
              if len(exclude_inds) + len(more_excludes) == len(
                  event_codes
              ):
                  warn(
                      "Did not find any event code with more than one "
                      "event.",
                      RuntimeWarning,
                  )
              else:
                  exclude_inds.extend(more_excludes)
  
              exclude_inds.sort()
              include_ = [
                  i
                  for i in np.arange(egi_info["n_events"])
                  if i not in exclude_inds
              ]
              include_names = [
                  k for i, k in enumerate(event_codes) if i in include_
              ]
          else:
              include_ = [
                  i for i, k in enumerate(event_codes) if k in include
              ]
              include_names = include
  
          for kk, v in [
              ("include", include_names),
              ("exclude", exclude),
          ]:
              if isinstance(v, list):
                  for k in v:
                      if k not in event_codes:
                          raise ValueError(
                              f"Could not find event named {repr(k)}"
                          )
              elif v is not None:
                  raise ValueError(
                      "`%s` must be None or of type list" % kk
                  )
          logger.info('    Synthesizing trigger channel "STI 014" ...')
          logger.info(
              "    Excluding events {%s} ..."
              % ", ".join(
                  [
                      k
                      for i, k in enumerate(event_codes)
                      if i not in include_
                  ]
              )
          )
          if all(ch.startswith("D") for ch in include_names):
              # support the DIN format DIN1, DIN2, ..., DIN9, DI10, DI11, ... DI99,
              # D100, D101, ..., D255 that we get when sending 0-255 triggers on a
              # parallel port.
              events_ids = list()
              for ch in include_names:
                  while not ch[0].isnumeric():
                      ch = ch[1:]
                  events_ids.append(int(ch))
          else:
              events_ids = np.arange(len(include_)) + 1
          egi_info["new_trigger"] = _combine_triggers(
              egi_events[include_], remapping=events_ids
          )
          self.event_id = dict(
              zip(
                  [e for e in event_codes if e in include_names],
                  events_ids,
              )
          )
          if egi_info["new_trigger"] is not None:
              egi_events = np.vstack(
                  [egi_events, egi_info["new_trigger"]]
              )
          assert egi_events.shape[1] == egi_info["last_samps"][-1]
      else:
          # No events
          self.event_id = None
          egi_info["new_trigger"] = None
          event_codes = []
  
      meas_dt_utc = egi_info["meas_dt_local"].astimezone(
          datetime.timezone.utc
      )
      info = _empty_info(egi_info["sfreq"])
      info["meas_date"] = _ensure_meas_date_none_or_dt(meas_dt_utc)
      info["utc_offset"] = egi_info["utc_offset"]
      info["device_info"] = dict(type=egi_info["device"])
  
      # read in the montage, if it exists
      ch_names, mon = _read_locs(input_fname, egi_info, channel_naming)
      # Second: Stim
      ch_names.extend(list(egi_info["event_codes"]))
      if egi_info["new_trigger"] is not None:
          ch_names.append("STI 014")  # channel for combined events
      cals = np.concatenate(
          [
              cals,
              np.repeat(1, len(event_codes) + 1 + len(misc) + len(eog)),
          ]
      )
  
      # Third: PNS
      ch_names.extend(egi_info["pns_names"])
      cals = np.concatenate(
          [cals, np.repeat(1, len(egi_info["pns_names"]))]
      )
  
      # Actually create channels as EEG, then update stim and PNS
      ch_coil = FIFF.FIFFV_COIL_EEG
      ch_kind = FIFF.FIFFV_EEG_CH
      chs = _create_chs(
          ch_names, cals, ch_coil, ch_kind, eog, (), (), misc
      )
  
      sti_ch_idx = [
          i
          for i, name in enumerate(ch_names)
          if name.startswith("STI") or name in event_codes
      ]
      for idx in sti_ch_idx:
          chs[idx].update(
              {
                  "unit_mul": FIFF.FIFF_UNITM_NONE,
                  "cal": cals[idx],
                  "kind": FIFF.FIFFV_STIM_CH,
                  "coil_type": FIFF.FIFFV_COIL_NONE,
                  "unit": FIFF.FIFF_UNIT_NONE,
              }
          )
      chs = _add_pns_channel_info(chs, egi_info, ch_names)
      info["chs"] = chs
      info._unlocked = False
      info._update_redundant()
  
      if mon is not None:
          info.set_montage(mon, on_missing="ignore")
  
      ref_idx = np.flatnonzero(np.isin(mon.ch_names, REFERENCE_NAMES))
      if len(ref_idx):
          ref_idx = ref_idx.item()
          ref_coords = info["chs"][int(ref_idx)]["loc"][:3]
          for chan in info["chs"]:
              if chan["kind"] == FIFF.FIFFV_EEG_CH:
                  chan["loc"][3:6] = ref_coords
  
      file_bin = op.join(input_fname, egi_info["eeg_fname"])
      egi_info["egi_events"] = egi_events
  
      # Check how many channels to read are from EEG
      keys = ("eeg", "sti", "pns")
      idx = dict()
      idx["eeg"] = np.where(
          [ch["kind"] == FIFF.FIFFV_EEG_CH for ch in chs]
      )[0]
      idx["sti"] = np.where(
          [ch["kind"] == FIFF.FIFFV_STIM_CH for ch in chs]
      )[0]
      idx["pns"] = np.where(
          [
              ch["kind"]
              in (
                  FIFF.FIFFV_ECG_CH,
                  FIFF.FIFFV_EMG_CH,
                  FIFF.FIFFV_BIO_CH,
              )
              for ch in chs
          ]
      )[0]
      # By construction this should always be true, but check anyway
      if not np.array_equal(
          np.concatenate([idx[key] for key in keys]),
          np.arange(len(chs)),
      ):
          raise ValueError(
              "Currently interlacing EEG and PNS channels"
              "is not supported"
          )
      egi_info["kind_bounds"] = [0]
      for key in keys:
          egi_info["kind_bounds"].append(len(idx[key]))
      egi_info["kind_bounds"] = np.cumsum(egi_info["kind_bounds"])
      assert egi_info["kind_bounds"][0] == 0
      assert egi_info["kind_bounds"][-1] == info["nchan"]
      first_samps = [0]
      last_samps = [egi_info["last_samps"][-1] - 1]
  
      annot = dict(onset=list(), duration=list(), description=list())
      if len(idx["pns"]):
          # PNS Data is present and should be read:
          egi_info["pns_filepath"] = op.join(
              input_fname, egi_info["pns_fname"]
          )
          # Check for PNS bug immediately
          pns_samples = np.sum(
              egi_info["pns_sample_blocks"]["samples_block"]
          )
          eeg_samples = np.sum(egi_info["samples_block"])
          if pns_samples == eeg_samples - 1:
              warn("This file has the EGI PSG sample bug")
              annot["onset"].append(last_samps[-1] / egi_info["sfreq"])
              annot["duration"].append(1 / egi_info["sfreq"])
              annot["description"].append("BAD_EGI_PSG")
          elif pns_samples != eeg_samples:
              raise RuntimeError(
                  "PNS samples (%d) did not match EEG samples (%d)"
                  % (pns_samples, eeg_samples)
              )
  
      self._filenames = [file_bin]
      self._raw_extras = [egi_info]
  
      super().__init__(
          info,
          preload=preload,
          orig_format="single",
          filenames=[file_bin],
          first_samps=first_samps,
          last_samps=last_samps,
          raw_extras=[egi_info],
          verbose=verbose,
      )
  
      # Annotate acquisition skips
      for first, prev_last in zip(
          egi_info["first_samps"][1:], egi_info["last_samps"][:-1]
      ):
          gap = first - prev_last
          assert gap >= 0
          if gap:
              annot["onset"].append(
                  (prev_last - 0.5) / egi_info["sfreq"]
              )
              annot["duration"].append(gap / egi_info["sfreq"])
              annot["description"].append("BAD_ACQ_SKIP")
  
      if len(annot["onset"]):
          self.set_annotations(Annotations(**annot, orig_time=None))
  
imports: import datetime
  import math
  import os.path
  import re
  from collections import OrderedDict
  from pathlib import Path
  import numpy
  from ..._fiff.constants import FIFF
  from ..._fiff.meas_info import _empty_info, _ensure_meas_date_none_or_dt, create_info
  from ..._fiff.proj import setup_proj
  from ..._fiff.utils import _create_chs, _mult_cal_one
  from ...annotations import Annotations
  from ...channels.montage import make_dig_montage
  from ...evoked import EvokedArray
  from ...utils import _check_fname, _check_option, _soft_import, logger, verbose, warn
  from ..base import BaseRaw
  from .events import _combine_triggers, _read_events
  from .general import _block_r, _extract, _get_blocks, _get_ep_info, _get_gains, _get_signalfname
label_desc: Use image node_id 1 to create a new RawMff object from inherited base classes: BaseRaw with example: obj = RawMff(input_fname, eog, misc, include, exclude, preload, channel_naming, verbose)