cldfbench_siniticbetween.py

import pathlib
import re
import unicodedata
import xml.etree.ElementTree as ET
from collections import Counter, namedtuple
from itertools import chain

from cldfbench import Dataset as BaseDataset, CLDFSpec
from simplepybtex.database import parse_file, Person

ID_CHARS = 'abcdefghijklmnopqrstuvwxyz0123456789-_'


def slug(s):
    return ''.join(
        c
        for c in unicodedata.normalize('NFKD', s).lower()
        if c in ID_CHARS)


def clean_csv(csv_rows):
    for row in csv_rows:
        row = {
            k.strip(): v.strip()
            for k, v in row.items()
            if k.strip and v.strip()}
        if row:
            yield row


def parse_coords(text):
    x, y, z = text.split(',')
    assert z == '0'
    return float(x), float(y)


def read_coordinates_from_kml(f):
    tree = ET.parse(f)
    root = tree.getroot()
    return {
        desc.text.strip(): parse_coords(coord.text.strip())
        for placemark in root.iter('{http://www.opengis.net/kml/2.2}Placemark')
        if (desc := placemark.find('{http://www.opengis.net/kml/2.2}description')) is not None
        and (point := placemark.find('{http://www.opengis.net/kml/2.2}Point')) is not None
        and (coord := point.find('{http://www.opengis.net/kml/2.2}coordinates')) is not None}


def normalise_name(name):
    # strip periods
    return slug(name.rstrip('.'))


def make_hashable_person(person):
    names = chain(
        person.first_names or (),
        person.middle_names or (),
        person.prelast_names or (),
        person.last_names or (),
        person.lineage_names or ())
    return tuple(filter(None, map(normalise_name, names)))


Citation = namedtuple('Citation', 'authors year')


def make_citations(entries):
    citations = {}
    for bibkey, entry in entries:
        authors = entry.persons.get('author') or ()
        editors = entry.persons.get('editor') or ()
        people = authors or editors
        citation = Citation(
            tuple(map(make_hashable_person, people)),
            entry.fields.get('year'))
        assert citation not in citations, citation
        citations[citation] = bibkey
    return citations


def parse_source_string(source_string):
    if source_string.startswith('Phukan, J. N. An introductory'):
        return Citation((('j', 'n', 'phukan'),), None)

    source_string = source_string.replace(
        'Mao, Zongwu, Chaoji Meng & Zongze Zheng',
        'Mao, Zongwu & Chaoji Meng & Zongze Zheng')
    source_string = source_string.replace(
        'Shintani, Tadahiko & Zhao Yang',
        'Shintani, Tadahiko & Zhao, Yang')
    source_string = source_string.replace(
        'Li, Rulong, Xiaoying Hou, Tiansong Lin',
        'Li, Rulong & Xiaoying Hou & Tiansong Lin')
    source_string = source_string.replace(
        'Zhang, Junru, Min Liang, Jueya Ouyang, Yiqing Zheng, Xulian Li',
        'Zhang, Junru & Min Liang & Jueya Ouyang & Yiqing Zheng & Xulian Li')
    source_string = source_string.replace(
        'Liang, Min, Junru Zhang & Yunbing Li',
        'Liang, Min & Junru Zhang & Yunbing Li')
    source_string = source_string.replace(
        'Zhou, Zhizhi, Qixiang Yan & Guoqing Chen',
        'Zhou, Zhizhi & Qixiang Yan & Guoqing Chen')
    source_string = source_string.replace(
        'Chen, Xiangmu, Jingliu Wang & Yongliang Lai',
        'Chen, Xiangmu & Jingliu Wang & Yongliang Lai')
    source_string = source_string.replace(
        'Li, Daoyong, Xizhen Nie & Efeng Qiu',
        'Li, Daoyong & Xizhen Nie & Efeng Qiu')
    source_string = source_string.replace(
        'Deepadung, Sujaritlak, Ampikaand Rattanapitak & Supakit Buakaw',
        'Deepadung, Sujaritlak & Ampika Rattanapitak & Supakit Buakaw')
    source_string = source_string.replace(
        'Ouyang, Jueya, Fang Cheng & Cuirong Yu',
        'Ouyang, Jueya & Fang Cheng & Cuirong Yu')
    source_string = source_string.replace(
        'Kruspe, Nicole, Niclas Burenhult & Ewelina Wnuk',
        'Kruspe, Nicole & Niclas Burenhult & Ewelina Wnuk')
    source_string = source_string.replace(
        'Kruspe, Nicole; Burenhult, Niclas & Wnuk, Ewelina',
        'Kruspe, Nicole & Burenhult, Niclas & Wnuk, Ewelina')
    source_string = source_string.replace('Yager, Joanneand', 'Yager, Joanne')

    m = re.match(r'(.*?)(?:\s+\(ed.\))?\.\s+(\d\d\d\d[^.]*)\.', source_string)
    assert m, source_string
    authors_string, year = m.groups()
    authors = tuple(
        make_hashable_person(Person(author))
        for author in re.split(r'\s*&\s*', authors_string))
    return Citation(authors, year)


def make_language(
    csv_language, coordinates, citations, languoids, unique_glottocodes,
):
    glottocode = csv_language['Glottocode']
    languoid = languoids[glottocode]
    lon, lat = coordinates[csv_language['Code']]
    source_string = csv_language['Reference']

    citation = parse_source_string(source_string)
    lang = {
        'ID': slug(csv_language['Code']),
        'Name': csv_language['Datapoint'],
        'Family': languoid.family.name,
        'Latitude': lat,
        'Longitude': lon,
        'Source': [citations[citation]],
    }

    if languoid.macroareas:
        lang['Macroarea'] = languoid.macroareas[0].name

    # FIXME(johannes): how do we actually deal with non-unique glottocodes
    if glottocode in unique_glottocodes:
        if (iso := languoid.iso_code):
            lang['ISO639P3code'] = iso
        lang['Glottocode'] = glottocode

    return lang


def make_languages(csv_languages, coordinates, citations, glottolog):
    all_glottocodes = Counter(lang['Glottocode'] for lang in csv_languages)
    unique_glottocodes = {gc for gc, count in all_glottocodes.items() if count == 1}
    languoids = {lg.id: lg for lg in glottolog.languoids(ids=all_glottocodes)}
    return {
        lang['Code']: make_language(
            lang, coordinates, citations, languoids, unique_glottocodes)
        for lang in csv_languages}


def make_values(data, languages, parameters, codes):
    return [
        {
            'Language_ID': (lang_id := languages[row['Code']]['ID']),
            'Parameter_ID': (param_id := param['ID']),
            'Code_ID': (code := codes[param_id, row[colname]])['ID'],
            'Value': code['Name'],
            'ID': f'{lang_id}-{param_id}',
        }
        for row in data
        for colname, param in parameters.items()]


class Dataset(BaseDataset):
    dir = pathlib.Path(__file__).parent
    id = 'siniticbetween'

    def cldf_specs(self):  # A dataset must declare all CLDF sets it creates.
        return CLDFSpec(
            module='StructureDataset',
            dir=self.cldf_dir,
            metadata_fname='cldf-metadata.json')

    def cmd_download(self, _args):
        """
        Download files to the raw/ directory. You can use helpers methods of `self.raw_dir`, e.g.

        >>> self.raw_dir.download(url, fname)
        """
        csv_dir = self.raw_dir / 'csv-export'
        for excel_file in self.raw_dir.glob('*.xlsx'):
            self.raw_dir.xlsx2csv(excel_file.name, outdir=csv_dir)

    def cmd_readme(self, _args):
        section_header = (
            'Sinitic between Altaic and Mainland Southeast Asia\n'
            '==================================================\n'
            '\n')
        section_content = self.raw_dir.read('intro.md')
        return f'{section_header}\n{section_content}'

    def cmd_makecldf(self, args):
        """
        Convert the raw data to a CLDF dataset.

        >>> args.writer.objects['LanguageTable'].append(...)
        """

        with open(self.raw_dir / 'language-map.kml', encoding='utf-8') as f:
            coordinates = read_coordinates_from_kml(f)

        sinitic_sources = parse_file(self.raw_dir / 'sinitic.bib', 'bibtex')
        altaic_sources = parse_file(self.raw_dir / 'altaic-and-msea.bib', 'bibtex')
        citations = make_citations(chain(
            sinitic_sources.entries.items(),
            altaic_sources.entries.items()))

        csv_dir = self.raw_dir / 'csv-export'
        languages = make_languages(
            list(clean_csv(chain(
                csv_dir.read_csv('sinitic.1.csv', dicts=True),
                csv_dir.read_csv('altaic-and-msea.1.csv', dicts=True)))),
            coordinates,
            citations,
            args.glottolog.api)

        data = list(clean_csv(csv_dir.read_csv('data.Value.csv', dicts=True)))

        parameters = {
            param['Original_Name']: param
            for param in clean_csv(self.etc_dir.read_csv(
                'parameters.csv', dicts=True))}
        codes = {
            (code['Parameter_ID'], code['Original_Name']): code
            for code in clean_csv(self.etc_dir.read_csv(
                'codes.csv', dicts=True))}

        values = make_values(data, languages, parameters, codes)

        # write cldf

        args.writer.cldf.add_component(
            'LanguageTable',
            'Family',
            'http://cldf.clld.org/v1.0/terms.rdf#source')
        args.writer.cldf.add_component('ParameterTable')
        args.writer.cldf.add_component('CodeTable', 'Map_Icon')

        args.writer.objects['ValueTable'] = values
        args.writer.objects['ParameterTable'] = parameters.values()
        args.writer.objects['CodeTable'] = codes.values()
        args.writer.objects['LanguageTable'] = languages.values()

        args.writer.cldf.add_sources(sinitic_sources)
        args.writer.cldf.add_sources(altaic_sources)