XML文件解析

[exogeny]

分割XML文件

import tqdm

XML_PREFIX = [
  '<?xml version="1.0" encoding="UTF-8"?>',
  '<proteinAtlas xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://v23.proteinatlas.org/download/proteinatlas.xsd" schemaVersion="2.6">',
]

XML_POSTFIX = [
  '<copyright>Copyrighted by the Human Protein Atlas, http://www.proteinatlas.org/about/licence</copyright>',
  '</proteinAtlas>',
]

if __name__ == '__main__':
  ensg_id = None
  content = []
  with open('proteinatlas.xml', 'r') as f:
    for line in tqdm.tqdm(f):
      if line.strip().startswith('<entry'):
        # <entry version="23" url="http://v23.proteinatlas.org/ENSG00000000003">
        index = line.find('ENSG')
        ensg_id = line[index:index+15]
        content.clear()
        content.extend(XML_PREFIX)
      content.append(line.strip('\n'))
      if line.strip().startswith('</entry'):
        content.extend(XML_POSTFIX)
        with open(f'protein_entries/{ensg_id}.xml', 'w') as xmlf:
          xmlf.write('\n'.join(content))

解析XML文件

import os
import tqdm
import xml.etree.ElementTree as ET


def split_image_url(image_url):
  if image_url is None:
    return None, None

  image_id, filename = image_url.split('/')[-2:]
  file_prefix = filename.replace('_blue_red_green.jpg', '')
  return image_id, file_prefix


def read_cell_expression(cell_expression, location_status):
  subassays = cell_expression.findall('./subAssay')
  for subassay in subassays:
    verification = subassay.find('./verification')
    verification = verification.text if verification is not None else ''

    subassay_type = subassay.get('type')
    for data in subassay.findall('./data'):
      cell_line = data.find('./cellLine').text
      locations = data.findall('./location[@GOId]')
      locations = [location.get('GOId') for location in locations]
      main_locations, extra_locations = [], []
      for location in locations:
        if location_status.get(location, None) == 'main':
          main_locations.append(location)
        else:
          extra_locations.append(location)
      main_locations = ','.join(main_locations)
      extra_locations = ','.join(extra_locations)

      for image in data.findall('./assayImage/image[@imageType="sampleImage"]'):
        image_url = image.find('imageUrl')
        if image_url is not None:
          image_url = image_url.text
        image_id, file_prefix = split_image_url(image_url)
        if image_id is not None:
          yield image_id, file_prefix, subassay_type, verification, cell_line, main_locations, extra_locations


def read_entry(entry):
  image_infos = []
  root = ET.fromstring(entry)

  # read locations
  location_status = {}
  cell_expression = root.find('./entry/cellExpression[@source="HPA"][@technology="ICC/IF"]')
  if cell_expression is not None:
    for location in cell_expression.findall('./data/location[@status][@GOId]'):
      location_status[location.get('GOId')] = location.get('status')

  antibodies = root.findall('./entry/antibody')
  for antibody in antibodies:
    cell_expressions = antibody.findall('./cellExpression[@source="HPA"][@technology="ICC/IF"]')
    for cell_expression in cell_expressions:
      for image_info in read_cell_expression(cell_expression, location_status):
        image_infos.append(image_info)
  return image_infos


if __name__ == '__main__':
  ensgs = os.listdir('proteins_xml')
  ensgs = [ensg.replace('.xml', '') for ensg in ensgs]

  with open('image_file_info.csv', 'w') as csvf:
    for ensg in tqdm.tqdm(ensgs):
      with open(f'proteins_xml/{ensg}.xml', 'r') as f:
        image_infos = read_entry(f.read())
        if len(image_infos) == 0:
          continue

        image_infos = ['\t'.join(image_info) for image_info in image_infos]
        image_infos = [f'{ensg}\t{image_info}' for image_info in image_infos]
        info_text = '\n'.join(image_infos)
        csvf.write(f'{info_text}\n')