def _validate_and_init_schema_dicom_tag()

in pathology/transformation_pipeline/ingestion_lib/dicom_gen/dicom_schema_util.py [0:0]

• 117 lines of code
• 28 McCabe index (conditional complexity)

  def _validate_and_init_schema_dicom_tag(self):
    """Validates DICOM tag schema definition and further inits tag.

    Raises:
      DICOMSchemaTagError: error in schema tag def
    """
    tagkw = self.get_schema_key_val(_SchemaTagKeywords.KEYWORD)
    try:
      if SchemaDICOMTag.get_kw_address(tagkw) != self._address:
        raise DICOMSchemaTagError(
            self,
            (
                'Schema tag keyword does not resolve to '
                f' specified address; keyword: {tagkw};'
                f' keyword: {tagkw} != {self._address}'
            ),
        )
    except DICOMSchemaError as exp:
      raise DICOMSchemaTagError(
          self,
          (
              'Schema tag keyword does not resolve to '
              f' address; keyword: {tagkw};\n'
              f'exception: {traceback.format_exc()}'
          ),
      ) from exp
    tag_param = []
    for test_param in [
        _SchemaTagKeywords.META,
        _SchemaTagKeywords.STATIC_VALUE,
        _SchemaTagKeywords.SQ,
        _SchemaTagKeywords.SEQ,
    ]:
      sq = self.get_schema_key_val(test_param)
      if sq is not None:
        tag_param.append(test_param)
    if not tag_param:
      raise DICOMSchemaTagError(
          self, 'Tag is missing expected value def: META, STATIC_VALUE, or SQ'
      )
    if len(tag_param) > 1:
      raise DICOMSchemaTagError(
          self,
          (
              'Tag definition has multiple [META, '
              'STATIC_VALUE, or SQ] value defs. '
              f'Tag defines: {str(tag_param)}'
          ),
      )
    self._schema_tag_value_source = tag_param[0]

    self._meta_join = self.get_schema_key_val(_SchemaTagKeywords.META_JOIN)
    if self._meta_join is not None and not isinstance(self._meta_join, str):
      raise DICOMSchemaTagError(
          self, 'Invalid meta_join value. Must be a string.'
      )

    value_limit = self.get_schema_key_val(_SchemaTagKeywords.VALUE_CHAR_LIMIT)
    if value_limit is not None:
      if isinstance(
          value_limit, str
      ) and SchemaDICOMTag._valid_value_char_limit(value_limit):
        self._value_char_limit = value_limit
      else:
        raise DICOMSchemaTagError(
            self, 'Invalid value_limit. Must be a valid string.'
        )

    conditional_on = self.get_schema_key_val(_SchemaTagKeywords.CONDITIONAL_ON)
    if conditional_on is not None:
      if isinstance(conditional_on, str):
        m_name = norm_column_name(conditional_on)
        self._conditional_on.add(m_name)
      else:
        for m_name in conditional_on:
          m_name = norm_column_name(m_name)
          self._conditional_on.add(m_name)
    write_uninit = self.get_schema_key_val(_SchemaTagKeywords.WRITE_EMPTY)
    if write_uninit is not None:
      write_uninit = norm_column_name(write_uninit)
      if write_uninit not in (
          _SchemaTagKeywords.TRUE,
          _SchemaTagKeywords.FALSE,
      ):
        raise DICOMSchemaTagError(
            self,
            (
                'WRITE_EMPTY expects "TRUE" or "FALSE"'
                f' values; WRITE_EMPTY: {write_uninit}'
            ),
        )
      self._write_uninitialized = write_uninit == _SchemaTagKeywords.TRUE

    required = self.get_schema_key_val(_SchemaTagKeywords.REQUIRED)
    if required is not None:
      required = norm_column_name(str(required))
      if required not in (_SchemaTagKeywords.TRUE, _SchemaTagKeywords.FALSE):
        raise DICOMSchemaTagError(
            self,
            (
                'REQUIRED expects "TRUE" or "FALSE"'
                f' values; WRITE_EMPTY: {required}'
            ),
        )
      self._tag_value_definition_required = required == _SchemaTagKeywords.TRUE
    else:
      self._tag_value_definition_required = (
          ingest_flags.REQUIRE_TYPE1_DICOM_TAG_METADATA_IS_DEFINED_FLG.value
          and SchemaDICOMTag._has_vals_in(self._table_req, [_DICOM_TYPE_1_TAG])
      )

    if SchemaDICOMTag._has_vals_in(self._table_req, [_DICOM_TYPE_2_TAG]):
      self._write_uninitialized = True

    if not self.is_seq and self.is_tag_schema_define_seq():
      raise DICOMSchemaTagError(self, 'Tags which are not SQ VR cannot be SQ.')
    elif self.is_seq and not self.is_tag_schema_define_seq():
      raise DICOMSchemaTagError(
          self, 'SQ VR type tags must be initialized as SQ'
      )
    if (
        self.is_tag_schema_define_seq()
        and not isinstance(self.seq_metadata, list)
        and not isinstance(self.seq_metadata, dict)
    ):
      raise DICOMSchemaTagError(
          self, '"SQ" should be set to a list of datasets.'
      )