def get_dicom_schema()

in pathology/transformation_pipeline/ingestion_lib/dicom_gen/wsi_to_dicom/metadata_storage_client.py [0:0]
52 lines of code
6 McCabe index (conditional complexity)

  def get_dicom_schema(self, schema_search: Dict[str, Any]) -> Dict[str, Any]:
    """Returns file path containing dicom mapping schema.

    Args:
      schema_search: Parameters to search schemas.

    Returns:
      File path

    Raises:
      MetadataSchemaExceptionError: Unable to find or read DICOM schema
    """
    schema_search = self._norm_schema(schema_search)
    schema_files_found = []
    for metadata in self._csv_metadata_cache:
      schema_file = metadata.filename
      if not is_schema(schema_file):
        continue
      schema_files_found.append(schema_file)

      try:
        with open(schema_file, 'rt') as infile:
          schema = json.load(infile)
      except json.JSONDecodeError as exp:
        msg = 'Error occurred reading metadata file'
        cloud_logging_client.warning(
            msg,
            {
                'filename': schema_file,
                'Metadata_storage_bucket': self._metadata_ingest_storage_bucket,
            },
            exp,
        )
        continue
      schema_def_key = _get_key(schema, 'DICOMSCHEMADEF')
      if not schema_def_key:
        msg = 'Schema missing DICOMSchemaDef file ignored.'
        cloud_logging_client.warning(
            msg,
            {
                'filename': schema_file,
                'Metadata_storage_bucket': self._metadata_ingest_storage_bucket,
            },
        )
        continue
      scheme_def = self._norm_schema(schema[schema_def_key])
      if self._do_schemas_match(metadata.filename, scheme_def, schema_search):
        cloud_logging_client.info(
            f'DICOM schema found: {metadata.filename}',
            {
                'schema_search': str(schema_search),
                'schema_filename': metadata.filename,
            },
        )
        del schema[schema_def_key]
        return schema

    cloud_logging_client.error(
        'DICOM schema not found',
        {
            'schema_search': str(schema_search),
            'schema_files_tested': str(schema_files_found),
        },
    )
    raise MetadataSchemaExceptionError('Schema not found')