output.csv
Make an asymmetrical diff query against a CBETA corpus.
tacl diff -a Dhr cbeta2-10.db corpus/cbeta/ dhr-vs-rest.txt > output.csv
Make a diff query against a Pagel corpus.
tacl diff -t pagel pagel1-7.db corpus/pagel/ by-author.txt > output.csv
''' + ENCODING_EPILOG
DIFF_HELP = 'List n-grams unique to each sub-corpus.'
HIGHLIGHT_BASE_NAME_HELP = 'Name of text to display.'
HIGHLIGHT_BASE_SIGLUM_HELP = 'Siglum of text to display.'
HIGHLIGHT_DESCRIPTION = '''\
Output an HTML document showing a text with its matches visually
highlighted.'''
HIGHLIGHT_EPILOG = '''\
The scope of the supplied results may have a dramatic influence on
the amount of highlighting. Results containing 1-grams are very
likely to be almost entirely highlighted. Results may be
restricted by using the tacl report command.
examples:
tacl highlight corpus/stripped/ intersect.csv T0001 å…ƒ'''
HIGHLIGHT_HELP = 'Output a text with its matches visually highlighted.'
INTERSECT_DESCRIPTION = '''\
List n-grams common to all sub-corpora (as defined by the labels
in the specified catalogue file).'''
INTERSECT_EPILOG = '''\
examples:
Make an intersect query against a CBETA corpus.
tacl intersect cbeta2-10.db corpus/cbeta/ dhr-vs-rest.txt > output.csv
Make an intersect query against a Pagel corpus.
tacl intersect -t pagel pagel1-7.db corpus/pagel/ by-author.txt > output.csv
''' + ENCODING_EPILOG
INTERSECT_HELP = 'List n-grams common to all sub-corpora.'
NGRAMS_DESCRIPTION = 'Generate n-grams from a corpus.'
NGRAMS_EPILOG = '''\
examples:
Create a database of 2 to 10-grams from a CBETA corpus.
tacl ngrams cbeta2-10.db corpus/cbeta/ 2 10
Create a database of 1 to 7-grams from a Pagel corpus.
tacl ngrams pagel1-7.db corpus/pagel/ 1 7
'''
NGRAMS_HELP = 'Generate n-grams from a corpus.'
NGRAMS_MAXIMUM_HELP = 'Maximum size of n-gram to generate (integer).'
NGRAMS_MINIMUM_HELP = 'Minimum size of n-gram to generate (integer).'
PREPARE_DESCRIPTION = '''\
Convert CBETA TEI XML files (which may have multiple files per
text) into XML suitable for processing via the tacl strip
command.'''
PREPARE_HELP = '''\
Convert CBETA TEI XML files into an XML form suitable for
stripping.'''
PREPARE_INPUT_HELP = 'Directory containing XML files to prepare.'
PREPARE_OUTPUT_HELP = 'Directory to output prepared files to.'
REPORT_CATALOGUE_HELP = '''\
Path to the catalogue file used to generate the results'''
REPORT_DESCRIPTION = '''\
Modify a query results file by removing certain results. Outputs
the new set of results.'''
REPORT_EXTEND_HELP = '''\
Extend the results to list the highest size grams that also count
as matches, going beyond the maximum size recorded in the
database. This has no effect if the results contain only 1-grams.'''
REPORT_EPILOG = '''\
If more than one modifier is specified, they are applied in the
following order: --extend, --reduce, --reciprocal, --zero-fill,
--min/max-texts, --min/max-size, --min/max-count, --remove.
It is important to be careful with the use of --reduce. Coupled
with --max-size, many results may be discarded without trace
(since the reduce occurs first). Note too that performing "reduce"
on a set of results more than once will make the results
inaccurate!
Since this command always outputs a valid results file, its output
can be used as input for a subsequent tacl report command. To
chain commands together without creating an intermediate file,
pipe the commands together and use - instead of a filename, as:
tacl report --recriprocal results.csv | tacl report --reduce -
examples:
Extend CBETA results and set a minimum total count.
tacl report -e corpus/cbeta/ --min-count 9 output.csv > mod-output.csv
Zero-fill CBETA results.
tacl report -c dhr-vs-rest.txt -z corpus/cbeta/ output.csv > mod-output.csv
Reduce Pagel results.
tacl report --reduce -t pagel output.csv > mod-output.csv
''' + ENCODING_EPILOG
REPORT_HELP = 'Modify a query results file.'
REPORT_MINIMUM_COUNT_HELP = 'Minimum total count of n-gram to include.'
REPORT_MAXIMUM_COUNT_HELP = 'Maximum total count of n-gram to include.'
REPORT_MINIMUM_SIZE_HELP = 'Minimum size of n-grams to include.'
REPORT_MAXIMUM_SIZE_HELP = 'Maximum size of n-grams to include.'
REPORT_MINIMUM_TEXT_HELP = 'Minimum count of texts containing n-gram to include.'
REPORT_MAXIMUM_TEXT_HELP = 'Maximum count of texts containing n-gram to include.'
REPORT_RECIPROCAL_HELP = '''\
Remove n-grams that are not attested by at least one text in each
labelled set of texts. This can be useful after reducing a set of
intersection results.'''
REPORT_REDUCE_HELP = 'Remove n-grams that are contained in larger n-grams.'
REPORT_REMOVE_HELP = 'Remove labelled results.'
REPORT_RESULTS_HELP = 'Path to CSV results; use - for stdin.'
REPORT_SORT_HELP = 'Sort the results.'
REPORT_ZERO_FILL_HELP = '''\
Add rows with a count of 0 for each n-gram in each witness of a
text that has at least one witness bearing that n-gram. The
catalogue used to generate the results must also be specified with
the -c option.'''
SEARCH_DESCRIPTION = '''\
List texts containing at least one of the supplied n-grams, along
with a total count of how many occurrences of the n-grams are
present in each text, and the number of n-grams that match in each
text.
Specifying a catalogue file will not restrict the search to only
those labelled texts, but rather adds the labels to any
appropriate texts in the results.'''
SEARCH_HELP = 'List texts containing at least one of the supplied n-grams.'
SEARCH_NGRAMS_HELP = '''\
Path to file containing list of n-grams to search for, with one
n-gram per line.'''
STATISTICS_DESCRIPTION = '''
Generate summary statistics for a set of results. This gives the
counts of all tokens and matching tokens in each witness and the
percentage of the witness that is encompassed by the matches.'''
STATISTICS_HELP = 'Generate summary statistics for a set of results.'
STATISTICS_RESULTS_HELP = 'Path to CSV results.'
STRIP_DESCRIPTION = '''\
Preprocess a corpus by stripping unwanted material from each
text.'''
STRIP_EPILOG = '''\
The CBETA texts are in TEI XML that needs to have the markup and
metadata removed. If the TEI specifies textual variants, plain
text versions based on these are also created.'''
STRIP_HELP = 'Generate texts for use with TACL from a corpus of TEI XML.'
STRIP_INPUT_HELP = 'Directory containing files to strip.'
STRIP_OUTPUT_HELP = 'Directory to output stripped files to.'
SUPPLIED_DIFF_DESCRIPTION = '''\
List n-grams unique to each set of results (as defined by the
specified results files).'''
SUPPLIED_DIFF_HELP = 'List n-grams unique to each results file.'
SUPPLIED_EPILOG = '''\
The number of labels supplied must match the number of results
files. The first label is assigned to all results in the first
results file, the second label to all results in the second
results file, etc. The labels specified in the results files are
replaced with the supplied labels in the output.'''
SUPPLIED_DIFF_EPILOG = SUPPLIED_EPILOG.format('sdiff')
SUPPLIED_INTERSECT_EPILOG = SUPPLIED_EPILOG.format('sintersect')
SUPPLIED_INTERSECT_DESCRIPTION = '''\
List n-grams common to all sets of results (as defined by the
specified results files).'''
SUPPLIED_INTERSECT_HELP = 'List n-grams common to all results files.'
SUPPLIED_LABELS_HELP = 'Labels to be assigned in order to the supplied results.'
SUPPLIED_RESULTS_HELP = 'Paths to results files to be used in the query.'
TACL_DESCRIPTION = 'Analyse the text of corpora in various simple ways.'
TACL_HELPER_DESCRIPTION = '''\
Perform helpful but non-essential tacl-related functions.'''
TACL_HELPER_AGAINST_DESCRIPTION = '''\
Generate a script to compare each text of a corpus against all the
texts in another corpus.'''
TACL_HELPER_AGAINST_HELP = '''\
Generate a script to compare each text of a corpus against all the
texts in another corpus.'''
TACL_HELPER_AGAINST_A_HELP = '''\
File containing text names to compare (one per line).'''
TACL_HELPER_AGAINST_B_HELP = '''\
File containing corpus text names to be compared against (one per
line).'''
TACL_HELPER_COLLAPSE_DESCRIPTION = '''
Collapse result rows for multiple witnesses having the same count
for an n-gram. Instead of the "siglum" column, all of the
witnesses (per text) with the same n-gram count are listed, space
separated, in the "sigla" column.'''
TACL_HELPER_COLLAPSE_HELP = 'Collapse result rows for multiple witnesses having the same count for an n-gram'
TACL_HELPER_IN_DESCRIPTION = '''\
Generate a script to compare each text of a corpus with all the
other texts of that corpus.'''
TACL_HELPER_IN_HELP = '''\
Generate a script to compare each text of a corpus with all the
other texts of that corpus.'''
TACL_HELPER_IN_TEXTS_HELP = '''\
File containing text names to examine (one per line).'''
TACL_HELPER_OUTPUT = 'Output directory for script and catalogue files.'
TACL_HELPER_RESULTS_HELP = 'Path to CSV results'
VERBOSE_HELP = '''\
Display debug information; multiple -v options increase the verbosity.'''
# Error messages.
CATALOGUE_TEXT_RELABELLED_ERROR = 'Catalogue file labels "{}" more than once'
INSUFFICIENT_LABELS_QUERY_ERROR = 'Not running query with less than two defined labels'
LABEL_NOT_IN_CATALOGUE_ERROR = 'Supplied label is not present in the supplied catalogue'
SUPPLIED_ARGS_LENGTH_MISMATCH_ERROR = 'The number of labels supplied does not match the number of results files.'
# SQL statements.
ANALYSE_SQL = 'ANALYZE {}'
CREATE_INDEX_INPUT_RESULTS_SQL = 'CREATE INDEX IF NOT EXISTS ' \
'temp.InputResultsLabel ON InputResults (ngram)'
CREATE_INDEX_TEXT_SQL = 'CREATE INDEX IF NOT EXISTS TextIndexLabel ' \
'ON Text (label)'
CREATE_INDEX_TEXTHASNGRAM_SQL = 'CREATE UNIQUE INDEX IF NOT EXISTS ' \
'TextHasNGramIndex ON TextHasNGram (text, size)'
CREATE_INDEX_TEXTNGRAM_SQL = 'CREATE INDEX IF NOT EXISTS ' \
'TextNGramIndexTextNGram ON TextNGram (text, ngram)'
CREATE_TABLE_TEXT_SQL = 'CREATE TABLE IF NOT EXISTS Text (' \
'id INTEGER PRIMARY KEY ASC, ' \
'name TEXT NOT NULL, ' \
'siglum TEXT NOT NULL, ' \
'checksum TEXT NOT NULL, ' \
'token_count INTEGER NOT NULL, ' \
'label TEXT NOT NULL, ' \
'UNIQUE (name, siglum))'
CREATE_TABLE_TEXTNGRAM_SQL = 'CREATE TABLE IF NOT EXISTS TextNGram (' \
'text INTEGER NOT NULL REFERENCES Text (id), ' \
'ngram TEXT NOT NULL, ' \
'size INTEGER NOT NULL, ' \
'count INTEGER NOT NULL)'
CREATE_TABLE_TEXTHASNGRAM_SQL = 'CREATE TABLE IF NOT EXISTS TextHasNGram (' \
'text INTEGER NOT NULL REFERENCES Text (id), ' \
'size INTEGER NOT NULL, ' \
'count INTEGER NOT NULL)'
CREATE_TEMPORARY_NGRAMS_TABLE_SQL = 'CREATE TEMPORARY TABLE InputNGram (' \
'ngram TEXT)'
CREATE_TEMPORARY_RESULTS_TABLE_SQL = 'CREATE TEMPORARY TABLE InputResults (' \
'ngram TEXT NOT NULL, ' \
'size INTEGER NOT NULL, ' \
'name TEXT NOT NULL, ' \
'siglum TEXT NOT NULL, ' \
'count INTEGER NOT NULL, ' \
'label TEXT NOT NULL)'
DELETE_TEXT_HAS_NGRAMS_SQL = 'DELETE FROM TextHasNGram WHERE text = ?'
DELETE_TEXT_NGRAMS_SQL = 'DELETE FROM TextNGram WHERE text = ?'
DROP_TEMPORARY_NGRAMS_TABLE_SQL = 'DROP TABLE IF EXISTS InputNGram'
DROP_TEMPORARY_RESULTS_TABLE_SQL = 'DROP TABLE IF EXISTS InputResults'
DROP_TEXTNGRAM_INDEX_SQL = 'DROP INDEX IF EXISTS TextNGramIndexTextNGram'
INSERT_NGRAM_SQL = 'INSERT INTO TextNGram (text, ngram, size, count) ' \
'VALUES (?, ?, ?, ?)'
INSERT_TEXT_HAS_NGRAM_SQL = 'INSERT INTO TextHasNGram (text, size, count) ' \
'VALUES (?, ?, ?)'
INSERT_TEXT_SQL = 'INSERT INTO Text ' \
'(name, siglum, checksum, token_count, label) ' \
'VALUES (?, ?, ?, ?, ?)'
INSERT_TEMPORARY_NGRAM_SQL = 'INSERT INTO temp.InputNGram (ngram) VALUES (?)'
INSERT_TEMPORARY_RESULTS_SQL = 'INSERT INTO temp.InputResults ' \
'(ngram, size, name, siglum, count, label) ' \
'VALUES (?, ?, ?, ?, ?, ?)'
PRAGMA_CACHE_SIZE_SQL = 'PRAGMA cache_size={}'
PRAGMA_COUNT_CHANGES_SQL = 'PRAGMA count_changes=OFF'
PRAGMA_FOREIGN_KEYS_SQL = 'PRAGMA foreign_keys=ON'
PRAGMA_LOCKING_MODE_SQL = 'PRAGMA locking_mode=EXCLUSIVE'
PRAGMA_SYNCHRONOUS_SQL = 'PRAGMA synchronous=OFF'
PRAGMA_TEMP_STORE_SQL = 'PRAGMA temp_store=MEMORY'
SELECT_COUNTS_SQL = 'SELECT Text.name AS "text name", Text.siglum, ' \
'TextHasNGram.size, TextHasNGram.count AS "unique ngrams", ' \
'Text.token_count + 1 - TextHasNGram.size AS "total ngrams", ' \
'Text.token_count AS "total tokens", Text.label ' \
'FROM Text, TextHasNGram ' \
'WHERE Text.id = TextHasNGram.text AND Text.label IN ({}) ' \
'ORDER BY Text.name, TextHasNGram.size'
SELECT_DIFF_ASYMMETRIC_SQL = 'SELECT TextNGram.ngram, TextNGram.size, ' \
'TextNGram.count, Text.name AS "text name", Text.siglum, Text.label ' \
'FROM Text, TextNGram ' \
'WHERE Text.label = ? AND Text.id = TextNGram.text ' \
'AND TextNGram.ngram IN (' \
'SELECT TextNGram.ngram FROM Text, TextNGram ' \
'WHERE Text.id = TextNGram.text AND Text.label = ? ' \
'EXCEPT ' \
'SELECT TextNGram.ngram FROM Text, TextNGram ' \
'WHERE Text.id = TextNGram.text AND Text.label IN ({}))'
SELECT_DIFF_SQL = 'SELECT TextNGram.ngram, TextNGram.size, TextNGram.count, ' \
'Text.name AS "text name", Text.siglum, Text.label ' \
'FROM Text, TextNGram ' \
'WHERE Text.label IN ({}) AND Text.id = TextNGram.text ' \
'AND TextNGram.ngram IN (' \
'SELECT TextNGram.ngram FROM Text, TextNGram ' \
'WHERE Text.id = TextNGram.text AND Text.label IN ({}) ' \
'GROUP BY TextNGram.ngram HAVING COUNT(DISTINCT Text.label) = 1)'
SELECT_DIFF_SUPPLIED_SQL = '''SELECT ngram, size, count, name AS "text name",
siglum, label
FROM temp.InputResults
WHERE ngram IN (
SELECT ngram FROM temp.InputResults
GROUP BY ngram HAVING COUNT(DISTINCT label) = 1)'''
SELECT_HAS_NGRAMS_SQL = 'SELECT text FROM TextHasNGram ' \
'WHERE text = ? AND size = ?'
SELECT_INTERSECT_SQL = 'SELECT TextNGram.ngram, TextNGram.size, ' \
'TextNGram.count, Text.name AS "text name", Text.siglum, Text.label ' \
'FROM Text, TextNGram ' \
'WHERE Text.label IN ({}) AND Text.id = TextNGram.text ' \
'AND TextNGram.ngram IN ({})'
SELECT_INTERSECT_SUB_EXTRA_SQL = ' AND TextNGram.ngram IN ({})'
SELECT_INTERSECT_SUB_SQL = 'SELECT TextNGram.ngram ' \
'FROM Text, TextNGram ' \
'WHERE Text.label = ? AND Text.id = TextNGram.text'
SELECT_INTERSECT_SUPPLIED_SQL = '''SELECT ngram, size, count,
name AS "text name", siglum, label
FROM temp.InputResults
WHERE ngram IN (
SELECT ngram FROM temp.InputResults
GROUP BY ngram HAVING COUNT(DISTINCT label) = ?)'''
SELECT_SEARCH_SQL = 'SELECT Text.name AS "text name", Text.siglum, ' \
'SUM(TextNGram.count) AS count, ' \
"Text.label, group_concat(TextNGram.ngram, ', ') AS ngrams, " \
'count(TextNGram.ngram) AS number ' \
'FROM Text, TextNGram ' \
'WHERE Text.id = TextNGram.text ' \
'AND TextNGram.ngram IN (SELECT ngram FROM temp.InputNGram) ' \
'GROUP BY TextNGram.text'
SELECT_TEXT_TOKEN_COUNT_SQL = 'SELECT Text.token_count ' \
'FROM Text WHERE Text.name = ?'
SELECT_TEXT_SQL = 'SELECT id, checksum FROM Text WHERE name = ? AND siglum = ?'
UPDATE_LABEL_SQL = 'UPDATE Text SET label = ? WHERE name = ?'
UPDATE_LABELS_SQL = 'UPDATE Text SET label = ?'
UPDATE_TEXT_SQL = 'UPDATE Text SET checksum = ?, token_count = ? WHERE id = ?'
VACUUM_SQL = 'VACUUM'
PK \-•Gƒ]Àér ér tacl/data_store.py"""Module containing the DataStore class."""
import csv
import io
import logging
import os.path
import sqlite3
import sys
import pandas as pd
from . import constants
from .exceptions import MalformedQueryError
class DataStore:
"""Class representing the data store for text data.
It provides an interface to the underlying database, with methods
to add and query data.
"""
def __init__ (self, db_name, use_memory=True, ram=0):
self._logger = logging.getLogger(__name__)
if db_name == ':memory:':
self._db_name = db_name
else:
self._db_name = os.path.abspath(db_name)
self._conn = sqlite3.connect(self._db_name)
self._conn.row_factory = sqlite3.Row
if use_memory:
self._conn.execute(constants.PRAGMA_TEMP_STORE_SQL)
if ram:
cache_size = ram * -1000000
self._conn.execute(constants.PRAGMA_CACHE_SIZE_SQL.format(
cache_size))
self._conn.execute(constants.PRAGMA_COUNT_CHANGES_SQL)
self._conn.execute(constants.PRAGMA_FOREIGN_KEYS_SQL)
self._conn.execute(constants.PRAGMA_LOCKING_MODE_SQL)
self._conn.execute(constants.PRAGMA_SYNCHRONOUS_SQL)
def _add_indices (self):
"""Adds the database indices relating to n-grams."""
self._logger.info('Adding database indices')
self._conn.execute(constants.CREATE_INDEX_TEXTNGRAM_SQL)
self._logger.info('Indices added')
def add_ngrams (self, corpus, minimum, maximum):
"""Adds n-gram data from `corpus` to the data store.
:param corpus: corpus of texts
:type corpus: `Corpus`
:param minimum: minimum n-gram size
:type minimum: `int`
:param maximum: maximum n-gram size
:type maximum: `int`
"""
self._initialise_database()
for text in corpus.get_texts():
self._add_text_ngrams(text, minimum, maximum)
self._add_indices()
self._analyse()
def _add_temporary_ngrams (self, ngrams):
"""Adds `ngrams` to a temporary table."""
self._conn.execute(constants.DROP_TEMPORARY_NGRAMS_TABLE_SQL)
self._conn.execute(constants.CREATE_TEMPORARY_NGRAMS_TABLE_SQL)
self._conn.executemany(constants.INSERT_TEMPORARY_NGRAM_SQL,
[(ngram,) for ngram in ngrams])
def _add_temporary_results_sets (self, results_filenames, labels):
if len(labels) < 2:
raise MalformedQueryError(
constants.INSUFFICIENT_LABELS_QUERY_ERROR)
if len(results_filenames) != len(labels):
raise MalformedQueryError(
constants.SUPPLIED_ARGS_LENGTH_MISMATCH_ERROR)
self._create_temporary_results_table()
for results_filename, label in zip(results_filenames, labels):
with open(results_filename, encoding='utf-8', newline='') as fh:
self._add_temporary_results(fh, label)
self._add_temporary_results_index()
self._analyse('temp.InputResults')
def _add_temporary_results (self, results, label):
"""Adds `results` to a temporary table with `label`.
:param results: results file
:type results: `File`
:param label: label to be associated with results
:type label: `str`
"""
NGRAM, SIZE, NAME, SIGLUM, COUNT, LABEL = constants.QUERY_FIELDNAMES
reader = csv.DictReader(results)
data = [(row[NGRAM], row[SIZE], row[NAME], row[SIGLUM], row[COUNT],
label) for row in reader]
self._conn.executemany(constants.INSERT_TEMPORARY_RESULTS_SQL, data)
def _add_temporary_results_index (self):
self._logger.info('Adding index to temporary results table')
self._conn.execute(constants.CREATE_INDEX_INPUT_RESULTS_SQL)
self._logger.info('Index added')
def _add_text_ngrams (self, text, minimum, maximum):
"""Adds n-gram data from `text` to the data store.
:param text: text to get n-grams from
:type text: `Text`
:param minimum: minimum n-gram size
:type minimum: `int`
:param maximum: maximum n-gram size
:type maximum: `int`
"""
text_id = self._get_text_id(text)
self._logger.info('Adding n-grams ({} <= n <= {}) for {}'.format(
minimum, maximum, text.get_filename()))
skip_sizes = []
for size in range(minimum, maximum + 1):
if self._has_ngrams(text_id, size):
self._logger.info('{}-grams are already in the database'.format(
size))
skip_sizes.append(size)
for size, ngrams in text.get_ngrams(minimum, maximum, skip_sizes):
self._add_text_size_ngrams(text_id, size, ngrams)
def _add_text_record (self, text):
"""Adds a Text record for `text`.
:param text: text to add a record for
:type text: `Text`
"""
filename = text.get_filename()
name, siglum = text.get_names()
self._logger.info('Adding record for text {}'.format(filename))
checksum = text.get_checksum()
token_count = len(text.get_tokens())
cursor = self._conn.execute(constants.INSERT_TEXT_SQL,
[name, siglum, checksum, token_count, ''])
self._conn.commit()
return cursor.lastrowid
def _add_text_size_ngrams (self, text_id, size, ngrams):
"""Adds `ngrams`, that are of size `size`, to the data store.
The added `ngrams` are associated with `text_id`.
:param text_id: database ID of text associated with `ngrams`
:type text_id: `int`
:param size: size of n-grams
:type size: `int`
:param ngrams: n-grams to be added
:type ngrams: `collections.Counter`
"""
unique_ngrams = len(ngrams)
self._logger.info('Adding {} unique {}-grams'.format(
unique_ngrams, size))
parameters = [[text_id, ngram, size, count]
for ngram, count in ngrams.items()]
self._conn.execute(constants.INSERT_TEXT_HAS_NGRAM_SQL,
[text_id, size, unique_ngrams])
self._conn.executemany(constants.INSERT_NGRAM_SQL, parameters)
self._conn.commit()
def _analyse (self, table=''):
"""Analyses the database, or `table` if it is supplied.
:param table: optional name of table analyse
:type table: `str`
"""
self._logger.info('Starting analysis of database')
self._conn.execute(constants.ANALYSE_SQL.format(table))
self._logger.info('Analysis of database complete')
@staticmethod
def _check_diff_result (row, matches, tokenize, join):
"""Returns `row`, possibly with its count changed to 0, depending on
the status of the n-grams that compose it.
The n-gram represented in `row` can be decomposed into two
(n-1)-grams. If neither sub-n-gram is present in `matches`, do
not change the count since this is a new difference.
If both sub-n-grams are present with a positive count, do not
change the count as it is composed entirely of sub-ngrams and
therefore not filler.
Otherwise, change the count to 0 as the n-gram is filler.
:param row: result row of the n-gram to check
:type row: pandas.Series
:param matches: (n-1)-grams and their associated counts to check against
:type matches: `dict`
:param tokenize: function to tokenize a string
:type tokenize: `function`
:param join: function to join tokens into a string
:type join: `function`
:rtype: pandas.Series
"""
ngram_tokens = tokenize(row[constants.NGRAM_FIELDNAME])
sub_ngram1 = join(ngram_tokens[:-1])
sub_ngram2 = join(ngram_tokens[1:])
count = constants.COUNT_FIELDNAME
discard = False
# For performance reasons, avoid searching through matches
# unless necessary.
status1 = matches.get(sub_ngram1)
if status1 == 0:
discard = True
else:
status2 = matches.get(sub_ngram2)
if status2 == 0:
discard = True
elif (status1 is None) ^ (status2 is None):
discard = True
if discard:
row[count] = 0
return row
def counts (self, catalogue, output_fh):
"""Returns `output_fh` populated with CSV results giving
n-gram counts of the texts in `catalogue`.
:param catalogue: catalogue matching filenames to labels
:type catalogue: `Catalogue`
:param output_fh: object to output results to
:type output_fh: file-like object
:rtype: file-like object
"""
labels = list(self._set_labels(catalogue))
label_placeholders = self._get_placeholders(labels)
query = constants.SELECT_COUNTS_SQL.format(label_placeholders)
self._logger.info('Running counts query')
self._logger.debug('Query: {}\nLabels: {}'.format(query, labels))
cursor = self._conn.execute(query, labels)
return self._csv(cursor, constants.COUNTS_FIELDNAMES, output_fh)
def _create_temporary_results_table (self):
self._conn.execute(constants.DROP_TEMPORARY_RESULTS_TABLE_SQL)
self._conn.execute(constants.CREATE_TEMPORARY_RESULTS_TABLE_SQL)
def _csv (self, cursor, fieldnames, output_fh):
"""Writes the rows of `cursor` in CSV format to `output_fh`
and returns it.
:param cursor: database cursor containing data to be be output
:type cursor: `sqlite3.Cursor`
:param fieldnames: row headings
:type fieldnames: `list`
:param output_fh: file to write data to
:type output_fh: file object
:rtype: file object
"""
self._logger.info('Finished query; outputting results in CSV format')
# Specify a lineterminator to avoid an extra \r being added on
# Windows; see
# https://stackoverflow.com/questions/3191528/csv-in-python-adding-extra-carriage-return
if sys.platform in ('win32', 'cygwin') and output_fh is sys.stdout:
writer = csv.writer(output_fh, lineterminator='\n')
else:
writer = csv.writer(output_fh)
writer.writerow(fieldnames)
for row in cursor:
writer.writerow([row[fieldname] for fieldname in fieldnames])
self._logger.info('Finished outputting results')
return output_fh
def _delete_text_ngrams (self, text_id):
"""Deletes all n-grams associated with `text_id` from the data
store.
:param text_id: database ID of text
:type text_id: `int`
"""
self._conn.execute(constants.DELETE_TEXT_NGRAMS_SQL, [text_id])
self._conn.execute(constants.DELETE_TEXT_HAS_NGRAMS_SQL, [text_id])
self._conn.commit()
def diff (self, catalogue, tokenizer, output_fh):
"""Returns `output_fh` populated with CSV results giving the n-grams
that are unique to each labelled set of texts in `catalogue`.
Note that this is not the same as the symmetric difference of
these sets, except in the case where there are only two
labels.
:param catalogue: catalogue matching filenames to labels
:type catalogue: `Catalogue`
:param tokenizer: tokenizer for the n-grams
:type tokenizer: `Tokenizer`
:param output_fh: object to output results to
:type output_fh: file-like object
:rtype: file-like object
"""
labels = self._sort_labels(self._set_labels(catalogue))
if len(labels) < 2:
raise MalformedQueryError(constants.INSUFFICIENT_LABELS_QUERY_ERROR)
label_placeholders = self._get_placeholders(labels)
query = constants.SELECT_DIFF_SQL.format(label_placeholders,
label_placeholders)
parameters = labels + labels
self._logger.info('Running diff query')
self._logger.debug('Query: {}\nLabels: {}'.format(query, labels))
self._log_query_plan(query, parameters)
cursor = self._conn.execute(query, parameters)
results_fh = self._csv(cursor, constants.QUERY_FIELDNAMES,
io.StringIO(newline=''))
return self._reduce_diff_results(results_fh, tokenizer, output_fh)
def diff_asymmetric (self, catalogue, prime_label, tokenizer, output_fh):
"""Returns `output_fh` populated with CSV results giving the
difference in n-grams between the labelled sets of texts in
`catalogue`, limited to those texts labelled with
`prime_label`.
:param catalogue: catalogue matching filenames to labels
:type catalogue: `Catalogue`
:param prime_label: label to limit results to
:type prime_label: `str`
:param tokenizer: tokenizer for the n-grams
:type tokenizer: `Tokenizer`
:param output_fh: object to output results to
:type output_fh: file-like object
:rtype: file-like object
"""
labels = list(self._set_labels(catalogue))
if len(labels) < 2:
raise MalformedQueryError(constants.INSUFFICIENT_LABELS_QUERY_ERROR)
try:
labels.remove(prime_label)
except ValueError:
raise MalformedQueryError(constants.LABEL_NOT_IN_CATALOGUE_ERROR)
label_placeholders = self._get_placeholders(labels)
query = constants.SELECT_DIFF_ASYMMETRIC_SQL.format(label_placeholders)
parameters = [prime_label, prime_label] + labels
self._logger.info('Running asymmetric diff query')
self._logger.debug('Query: {}\nLabels: {}\nPrime label: {}'.format(
query, labels, prime_label))
self._log_query_plan(query, parameters)
cursor = self._conn.execute(query, parameters)
results_fh = self._csv(cursor, constants.QUERY_FIELDNAMES,
io.StringIO(newline=''))
return self._reduce_diff_results(results_fh, tokenizer, output_fh)
def diff_supplied (self, results_filenames, labels, tokenizer, output_fh):
"""Returns `output_fh` populated with CSV results giving the n-grams
that are unique to each set of texts in `results_sets`, using
the labels in `labels`.
Note that this is not the same as the symmetric difference of
these sets, except in the case where there are only two
labels.
:param results_filenames: list of results filenames to be diffed
:type results_filenames: `list` of `str`
:param labels: labels to be applied to the results_sets
:type labels: `list`
:param tokenizer: tokenizer for the n-grams
:type tokenizer: `Tokenizer`
:param output_fh: object to output results to
:type output_fh: file-like object
:rtype: file-like object
"""
self._add_temporary_results_sets(results_filenames, labels)
query = constants.SELECT_DIFF_SUPPLIED_SQL
self._logger.info('Running supplied diff query')
self._logger.debug('Query: {}'.format(query))
self._log_query_plan(query, [])
cursor = self._conn.execute(query)
results_fh = self._csv(cursor, constants.QUERY_FIELDNAMES,
io.StringIO(newline=''))
return self._reduce_diff_results(results_fh, tokenizer, output_fh)
def _drop_indices (self):
"""Drops the database indices relating to n-grams."""
self._logger.info('Dropping database indices')
self._conn.execute(constants.DROP_TEXTNGRAM_INDEX_SQL)
self._logger.info('Finished dropping database indices')
@staticmethod
def _get_intersection_subquery (labels):
# Create nested subselects.
subquery = constants.SELECT_INTERSECT_SUB_SQL
# The subqueries are nested in reverse order of 'size', so
# that the inmost select is operating on the smallest corpus,
# thereby minimising the result sets of outer queries the most.
for label in labels[1:]:
subquery = constants.SELECT_INTERSECT_SUB_SQL + \
constants.SELECT_INTERSECT_SUB_EXTRA_SQL.format(
subquery)
return subquery
@staticmethod
def _get_placeholders (items):
"""Returns a string of placeholders, one for each item in
`items`.
:param items: items to create placeholders for
:type items: `list`
:rtype: `str`
"""
return ('?,' * len(items)).strip(',')
def _get_text_id (self, text):
"""Returns the database ID of the Text record for `text`.
This may require creating such a record.
If `text`\'s checksum does not match an existing record's
checksum, the record's checksum is updated and all associated
TextNGram and TextHasNGram records are deleted.
:param text: text to add a record for
:type text: `.Text`
:rtype: `int`
"""
name, siglum = text.get_names()
text_record = self._conn.execute(constants.SELECT_TEXT_SQL,
[name, siglum]).fetchone()
if text_record is None:
text_id = self._add_text_record(text)
else:
text_id = text_record['id']
if text_record['checksum'] != text.get_checksum():
filename = text.get_filename()
self._logger.info('Text {} has changed since it was added to '
'the database'.format(filename))
self._update_text_record(text, text_id)
self._logger.info('Deleting potentially out-of-date n-grams')
self._delete_text_ngrams(text_id)
return text_id
def _has_ngrams (self, text_id, size):
"""Returns True if a text has existing records for n-grams of
size `size`.
:param text_id: database ID of text to check
:type text_id: `int`
:param size: size of n-grams
:type size: `int`
:rtype: `bool`
"""
if self._conn.execute(constants.SELECT_HAS_NGRAMS_SQL,
[text_id, size]).fetchone() is None:
return False
return True
def _initialise_database (self):
"""Creates the database schema.
This will not create tables or indices that already exist and
is safe to be called on an existing database.
"""
self._logger.info('Creating database schema, if necessary')
self._conn.execute(constants.CREATE_TABLE_TEXT_SQL)
self._conn.execute(constants.CREATE_TABLE_TEXTNGRAM_SQL)
self._conn.execute(constants.CREATE_TABLE_TEXTHASNGRAM_SQL)
self._conn.execute(constants.CREATE_INDEX_TEXTHASNGRAM_SQL)
self._conn.execute(constants.CREATE_INDEX_TEXT_SQL)
def intersection (self, catalogue, output_fh):
"""Returns `output_fh` populated with CSV results giving the
intersection in n-grams of the labelled sets of texts in
`catalogue`.
:param catalogue: catalogue matching filenames to labels
:type catalogue: `Catalogue`
:param output_fh: object to output results to
:type output_fh: file-like object
:rtype: file-like object
"""
labels = self._sort_labels(self._set_labels(catalogue))
if len(labels) < 2:
raise MalformedQueryError(constants.INSUFFICIENT_LABELS_QUERY_ERROR)
label_placeholders = self._get_placeholders(labels)
subquery = self._get_intersection_subquery(labels)
query = constants.SELECT_INTERSECT_SQL.format(label_placeholders,
subquery)
parameters = labels + labels
self._logger.info('Running intersection query')
self._logger.debug('Query: {}\nLabels: {}'.format(query, labels))
self._log_query_plan(query, parameters)
cursor = self._conn.execute(query, parameters)
return self._csv(cursor, constants.QUERY_FIELDNAMES, output_fh)
def intersection_supplied (self, results_filenames, labels, output_fh):
"""Returns `output_fh` populated with CSV results giving the n-grams
that are common to every set of texts in `results_sets`, using
the labels in `labels`.
:param results_filenames: list of results to be diffed
:type results_filenames: `list` of `str`
:param labels: labels to be applied to the results_sets
:type labels: `list`
:param output_fh: object to output results to
:type output_fh: file-like object
:rtype: file-like object
"""
self._add_temporary_results_sets(results_filenames, labels)
query = constants.SELECT_INTERSECT_SUPPLIED_SQL
parameters = [len(labels)]
self._logger.info('Running supplied intersect query')
self._logger.debug('Query: {}\nNumber of labels: {}'.format(
query, parameters[0]))
self._log_query_plan(query, parameters)
cursor = self._conn.execute(query, parameters)
return self._csv(cursor, constants.QUERY_FIELDNAMES, output_fh)
def _log_query_plan (self, query, parameters):
cursor = self._conn.execute('EXPLAIN QUERY PLAN ' + query, parameters)
query_plan = 'Query plan:\n'
for row in cursor.fetchall():
query_plan += '|'.join([str(value) for value in row]) + '\n'
self._logger.debug(query_plan)
def _reduce_diff_results (self, matches_fh, tokenizer, output_fh):
"""Returns `output_fh` populated with a reduced set of data from
`matches_fh`.
Diff results typically contain a lot of filler results that
serve only to hide real differences. If one text has a single
extra token than another, the diff between them will have
results for every n-gram containing that extra token, which is
not helpful. This method removes these filler results by
'reducing down' the results.
:param matches_fh: holder of CSV results to be reduced
:type matches_fh: `io.StringIO`
:param tokenizer: tokenizer for the n-grams
:type tokenizer: `Tokenizer`
:param output_fh: file to write results to
:type output_fh: file-like object
:rtype: file-like object
"""
self._logger.info('Removing filler results')
# For performance, perform the attribute accesses once.
tokenize = tokenizer.tokenize
join = tokenizer.joiner.join
results = []
previous_witness = (None, None)
previous_data = {}
# Calculate the index of ngram and count columns in a Pandas
# named tuple row, as used below. The +1 is due to the tuple
# having the row index as the first element.
ngram_index = constants.QUERY_FIELDNAMES.index(
constants.NGRAM_FIELDNAME) + 1
count_index = constants.QUERY_FIELDNAMES.index(
constants.COUNT_FIELDNAME) + 1
matches_fh.seek(0)
# Operate over individual witnesses and sizes, so that there
# is no possible results pollution between them.
grouped = pd.read_csv(matches_fh, encoding='utf-8',
na_filter=False).groupby(
[constants.NAME_FIELDNAME, constants.SIGLUM_FIELDNAME,
constants.SIZE_FIELDNAME])
for (text, siglum, size), group in grouped:
if (text, siglum) != previous_witness:
previous_matches = group
previous_witness = (text, siglum)
else:
self._logger.debug('Reducing down {} {}-grams for {} {}'.format(
len(group.index), size, text, siglum))
if previous_matches.empty:
reduced_count = 0
else:
previous_matches = group.apply(
self._check_diff_result, axis=1,
args=(previous_data, tokenize, join))
reduced_count = len(previous_matches[previous_matches[constants.COUNT_FIELDNAME] != 0].index)
self._logger.debug('Reduced down to {} grams'.format(
reduced_count))
# Put the previous matches into a form that is more
# performant for the lookups made in _check_diff_result.
previous_data = {}
for row in previous_matches.itertuples():
previous_data[row[ngram_index]] = row[count_index]
if not previous_matches.empty:
results.append(previous_matches[previous_matches[
constants.COUNT_FIELDNAME] != 0])
reduced_results = pd.concat(results, ignore_index=True).reindex(
columns=constants.QUERY_FIELDNAMES)
reduced_results.to_csv(output_fh, encoding='utf-8', float_format='%d',
index=False)
return output_fh
def search (self, catalogue, ngrams, output_fh):
self._set_labels(catalogue)
self._add_temporary_ngrams(ngrams)
query = constants.SELECT_SEARCH_SQL
self._logger.info('Running search query')
self._logger.debug('Query: {}\nN-grams: {}'.format(
query, ', '.join(ngrams)))
self._log_query_plan(query, [])
cursor = self._conn.execute(query)
return self._csv(cursor, constants.SEARCH_FIELDNAMES, output_fh)
def _set_labels (self, catalogue):
"""Returns a dictionary of the unique labels in `catalogue` and the
number of their associated texts, and sets the record of each
Text to the corresponding label.
Texts that do not have a label specified are set to the empty
string.
Token counts are included in the results to allow for
semi-accurate sorting based on corpora size.
:param catalogue: catalogue matching filenames to labels
:type catalogue: `Catalogue`
:rtype: `dict`
"""
self._conn.execute(constants.UPDATE_LABELS_SQL, [''])
labels = {}
for name, label in catalogue.items():
self._conn.execute(constants.UPDATE_LABEL_SQL, [label, name])
cursor = self._conn.execute(constants.SELECT_TEXT_TOKEN_COUNT_SQL,
[name])
token_count = cursor.fetchone()['token_count']
labels[label] = labels.get(label, 0) + token_count
self._conn.commit()
return labels
@staticmethod
def _sort_labels (label_data):
"""Returns the labels in `label_data` sorted in descending order
according to the 'size' (total token count) of their referent
corpora.
:param label_data: labels (with their token counts) to sort
:type: `dict`
:rtype: `list`
"""
labels = list(label_data)
labels.sort(key=label_data.get, reverse=True)
return labels
def _update_text_record (self, text, text_id):
"""Updates the record with `text_id` with `text`\'s checksum and
token count.
:param text: text to update from
:type text: `Text`
:param text_id: database ID of Text record
:type text_id: `int`
"""
checksum = text.get_checksum()
token_count = len(text.get_tokens())
self._conn.execute(constants.UPDATE_TEXT_SQL,
[checksum, token_count, text_id])
self._conn.commit()
def validate (self, corpus, catalogue):
"""Returns True if all of the files labelled in `catalogue`
are up-to-date in the database.
:param corpus: corpus of texts
:type corpus: `Corpus`
:param catalogue: catalogue matching filenames to labels
:type catalogue: `Catalogue`
:rtype: `bool`
"""
is_valid = True
for name in catalogue:
count = 0
# It is unfortunate that this creates Text objects for
# each text, since that involves reading the file.
for text in corpus.get_texts(name):
count += 1
name, siglum = text.get_names()
filename = text.get_filename()
row = self._conn.execute(constants.SELECT_TEXT_SQL,
[name, siglum]).fetchone()
if row is None:
is_valid = False
self._logger.warning(
'No record (or n-grams) exists for {} in '
'the database'.format(filename))
elif row['checksum'] != text.get_checksum():
is_valid = False
self._logger.warning(
'{} has changed since its n-grams were '
'added to the database'.format(filename))
if count == 0:
self._logger.error('Catalogue references text {} that does not '
'exist in the corpus'.format(name))
raise FileNotFoundError
return is_valid
PK YoGF˜…f f tacl/sequence.py"""Module containing the Sequence and Sequencer classes."""
import logging
import os
import re
from Bio import pairwise2
from jinja2 import Environment, PackageLoader
import pandas as pd
from . import constants
class Sequence:
def __init__ (self, alignment, substitutes):
self._alignment = alignment
self._substitutes = substitutes
def _format_alignment (self, a1, a2):
html = []
for index, char in enumerate(a1):
output = self._substitutes.get(char, char)
if a2[index] == char:
html.append('{}'.format(output))
elif char != '-':
html.append(output)
return ''.join(html)
def render (self):
"""Returns a tuple of HTML fragments rendering each element of the
sequence."""
f1 = self._format_alignment(self._alignment[0], self._alignment[1])
f2 = self._format_alignment(self._alignment[1], self._alignment[0])
return f1, f2
class Sequencer:
def __init__ (self, corpus, tokenizer, results, output_dir):
self._logger = logging.getLogger(__name__)
self._corpus = corpus
self._tokenizer = tokenizer
self._matches = pd.read_csv(results, encoding='utf-8', na_filter=False)
self._output_dir = output_dir
def _generate_sequence (self, t1, t1_span, t2, t2_span, context_length,
covered_spans):
old_length = 0
self._logger.debug('Match found; generating new sequence')
while True:
s1, span1 = self._get_text_sequence(t1, t1_span, context_length)
s2, span2 = self._get_text_sequence(t2, t2_span, context_length)
length = len(s1)
alignment = pairwise2.align.globalms(
s1, s2, constants.IDENTICAL_CHARACTER_SCORE,
constants.DIFFERENT_CHARACTER_SCORE,
constants.OPEN_GAP_PENALTY, constants.EXTEND_GAP_PENALTY)[0]
context_length = length
score = alignment[2] / length
if not alignment:
return None
elif score < constants.SCORE_THRESHOLD or length == old_length:
break
else:
self._logger.debug('Score: {}'.format(score))
old_length = length
covered_spans[0].append(span1)
covered_spans[1].append(span2)
return Sequence(alignment, self._r_substitutes)
def generate_sequences (self, minimum_size):
loader = PackageLoader('tacl', 'assets/templates')
env = Environment(loader=loader)
template = env.get_template('sequence.html')
# Get a list of the files in the matches, grouped by label
# (ordered by number of texts).
labels = list(self._matches.groupby([constants.LABEL_FIELDNAME])[constants.NAME_FIELDNAME].nunique().index)
ngrams = self._matches[self._matches[constants.SIZE_FIELDNAME] >= minimum_size].sort(constants.SIZE_FIELDNAME, ascending=False)[constants.NGRAM_FIELDNAME].unique()
for index, primary_label in enumerate(labels):
for secondary_label in labels[index+1:]:
self._generate_sequences(primary_label, secondary_label, ngrams,
template)
def _generate_sequences (self, primary_label, secondary_label, ngrams,
template):
self._substitutes = {}
self._char_code = 61440
cols = [constants.NAME_FIELDNAME, constants.SIGLUM_FIELDNAME]
primary_texts = self._matches[self._matches[constants.LABEL_FIELDNAME] == primary_label][cols].drop_duplicates()
secondary_texts = self._matches[self._matches[constants.LABEL_FIELDNAME] == secondary_label][cols].drop_duplicates()
for index, (name1, siglum1) in primary_texts.iterrows():
text1 = self._get_text(name1, siglum1)
label1 = '{}_{}'.format(name1, siglum1)
for index, (name2, siglum2) in secondary_texts.iterrows():
text2 = self._get_text(name2, siglum2)
label2 = '{}_{}'.format(name2, siglum2)
self._generate_sequences_for_texts(label1, text1, label2, text2,
ngrams, template)
def _generate_sequences_for_texts (self, l1, t1, l2, t2, ngrams, template):
self._r_substitutes = dict((v, k) for k, v in self._substitutes.items())
sequences = []
covered_spans = [[], []]
for ngram in ngrams:
# Keep track of the spans within each text that have been
# covered by a sequence, to ensure that they aren't
# reported more than once.
sequences.extend(self._generate_sequences_for_ngram(
t1, t2, ngram, covered_spans))
if sequences:
html = template.render(l1=l1, l2=l2, sequences=sequences)
os.makedirs(self._output_dir, exist_ok=True)
output_name = os.path.join(self._output_dir,
'{}-{}.html'.format(l1, l2))
with open(output_name, 'w', encoding='utf-8') as fh:
fh.write(html)
def _generate_sequences_for_ngram (self, t1, t2, ngram, covered_spans):
self._logger.debug('Generating sequences for n-gram "{}"'.format(ngram))
pattern = re.compile(re.escape(ngram))
context_length = len(ngram)
t1_spans = [match.span() for match in pattern.finditer(t1)]
t2_spans = [match.span() for match in pattern.finditer(t2)]
sequences = []
for t1_span in t1_spans:
for t2_span in t2_spans:
if self._is_inside(t1_span, t2_span, covered_spans):
self._logger.debug('Skipping match due to existing coverage')
continue
sequence = self._generate_sequence(
t1, t1_span, t2, t2_span, context_length, covered_spans)
if sequence:
sequences.append(sequence.render())
return sequences
def _get_text (self, name, siglum):
"""Returns the text identified by `name` and `siglum`, with all []
tokens replaced with a single character. Substitutions are
recorded in self._substitutes.
"""
tokens = self._corpus.get_text(name, siglum).get_tokens()
for i, token in enumerate(tokens):
if len(token) > 1:
char = chr(self._char_code)
substitute = self._substitutes.setdefault(token, char)
if substitute == char:
self._char_code += 1
tokens[i] = substitute
return self._tokenizer.joiner.join(tokens)
def _get_text_sequence (self, text, span, context_length):
start = max(0, span[0] - context_length)
end = min(len(text), span[1] + context_length)
return text[start:end], (start, end)
def _is_inside (self, span1, span2, covered_spans):
"""Returns True if both `span1` and `span2` fall within
`covered_spans`."""
if self._is_span_inside(span1, covered_spans[0]) and \
self._is_span_inside(span2, covered_spans[1]):
return True
return False
def _is_span_inside (self, span, covered_spans):
start = span[0]
end = span[1]
for c_start, c_end in covered_spans:
if start >= c_start and end <= c_end:
return True
return False
PK â.ÚF
Nl× × tacl/corpus.py"""Module containing the Corpus class."""
import glob
import logging
import os.path
from .text import Text
class Corpus:
"""A Corpus represents a collection of `Text`\s.
A Corpus is built from a directory that contains the text files
that become `Text` objects.
"""
def __init__ (self, path, tokenizer):
self._logger = logging.getLogger(__name__)
self._path = os.path.abspath(path)
self._tokenizer = tokenizer
def get_sigla (self, name):
"""Returns a list of all of the sigla for the named text.
:param name: name of text
:type name: `str`
:rtype: `list` of `str`
"""
return [os.path.splitext(os.path.basename(path))[0]
for path in glob.glob(os.path.join(self._path, name, '*.txt'))]
def get_text (self, name, siglum):
"""Returns a `Text` representing the file associated with `name` and
`siglum`.
Combined, `name` and `siglum` form the basis of a filename for
retrieving the text.
:param name: name of text
:type name: `str`
:param siglum: siglum (variant name) of text
:type siglum: `str`
:rtype: `Text`
"""
filename = os.path.join(name, siglum + '.txt')
self._logger.debug('Creating Text object from {}'.format(filename))
with open(os.path.join(self._path, filename), encoding='utf-8') as text:
content = text.read()
return Text(name, siglum, content, self._tokenizer)
def get_texts (self, name='*'):
"""Returns a generator supplying `Text` objects for each file
in the corpus.
:rtype: `generator`
"""
for filepath in glob.glob(os.path.join(self._path, name, '*.txt')):
if os.path.isfile(filepath):
name = os.path.split(os.path.split(filepath)[0])[1]
siglum = os.path.splitext(os.path.basename(filepath))[0]
yield self.get_text(name, siglum)
PK Á&ÚF³)ð ð tacl/exceptions.pyclass TACLError (Exception):
def __init__ (self, msg):
self._msg = msg
def __str__ (self):
return self._msg
class MalformedCatalogueError (TACLError):
pass
class MalformedQueryError (TACLError):
pass
PK …oG Údw w tacl/highlighter.py"""Module containing the Highlighter class."""
import logging
import re
from jinja2 import Environment, PackageLoader
from lxml import etree
import pandas as pd
from . import constants
from .text import Text
class Highlighter:
def __init__ (self, corpus, tokenizer):
self._logger = logging.getLogger(__name__)
self._corpus = corpus
self._tokenizer = tokenizer
def _annotate_tokens (self, match_obj):
match = match_obj.group(0)
root = etree.fromstring('
{}
'.format(match))
for span in root.xpath('//span'):
# The results are not guaranteed to have non-base matches
# in it, so do not rely on being able to derive base
# matches from them.
if self._match_source == self._base_filename:
if span.get('data-base-match') is None:
span.set('data-base-match', '')
else:
texts = span.get('data-texts')
if ' {} '.format(self._match_source) not in texts:
new_value = '{}{} '.format(texts, self._match_source)
span.set('data-texts', new_value)
return etree.tostring(root, encoding='unicode')[5:-6]
def _format_text (self, text):
"""Returns `text` with consecutive spaces converted to non-break
spaces, and linebreak converted into HTML br elements.
:param text: text to format
:type text: `str`
:rtype: `str`
"""
text = re.sub(r'\n', ' \n', text)
text = re.sub(r' ', ' ', text)
text = re.sub(r' ', ' ', text)
return text
def generate_base (self, matches, text_name, siglum, all=True):
"""Returns an XML document containing the text of `filename`
marked up with its n-grams in `matches`.
If `all` is True, generate results for all matches, not just
those on `filename`.
:param matches: matches data
:type matches: `pandas.DataFrame`
:param text_name: name of text to generate an XML document from
:type text_name: `str`
:param siglum: siglum of text variant to generate an XML document from
:type siglum: `str`
:rtype: `lxml.etree._Element`
"""
text = self._corpus.get_text(text_name, siglum)
filename = text.get_filename()
self._logger.debug('Generating the base XML file for {}'.format(
filename))
self._base_filename = filename
content = text.get_content().strip()
content = self._prepare_text(content)
if not all:
matches = matches[matches[constants.NAME_FIELDNAME] == filename]
content = self._highlight(content, matches)
content = self._format_text(content)
root = etree.fromstring('
{}
'.format(content))
return root
def _generate_html (self, matches, text_name, siglum, text):
loader = PackageLoader('tacl', 'assets/templates')
env = Environment(loader=loader)
text_list = self._generate_text_list(matches, text_name, siglum)
text_data = {'base_name': text_name, 'base_siglum': siglum,
'text': text, 'text_list': text_list}
template = env.get_template('highlight.html')
return template.render(text_data)
@staticmethod
def _generate_text_list (matches, base_name, base_siglum):
texts = matches[[constants.NAME_FIELDNAME,
constants.SIGLUM_FIELDNAME]].drop_duplicates()
text_list = []
for index, (name, siglum) in texts.iterrows():
if not(name == base_name and siglum == base_siglum):
text_list.append(Text.assemble_filename(name, siglum))
text_list.sort()
return text_list
def _get_regexp_pattern (self, ngram):
inter_token_pattern = r'\W*]*>'
pattern = inter_token_pattern.join(
[re.escape(token) for token in self._tokenizer.tokenize(ngram)])
return r'(]*>{})'.format(pattern)
def highlight (self, matches_filename, text_name, siglum):
"""Returns the text of `filename` as an HTML document with its matches
in `matches` highlighted.
:param results: file containing matches to highlight
:type results: `TextIOWrapper`
:param corpus: corpus of documents containing `text_filename`
:type corpus: `tacl.Corpus`
:param text_name: name of text to highlight
:type text_name: `str`
:param siglum: siglum of text to highlight
:type siglum: `str`
:rtype: `str`
"""
matches = pd.read_csv(matches_filename)
base = self.generate_base(matches, text_name, siglum, all=True)
text = etree.tostring(base, encoding='unicode', xml_declaration=False)
return self._generate_html(matches, text_name, siglum, text)
def _highlight (self, text, matches):
for row_index, row in matches.iterrows():
ngram = row[constants.NGRAM_FIELDNAME]
self._match_source = Text.assemble_filename(
row[constants.NAME_FIELDNAME], row[constants.SIGLUM_FIELDNAME])
pattern = self._get_regexp_pattern(ngram)
text = re.sub(pattern, self._annotate_tokens, text)
return text
def _prepare_text (self, text):
"""Returns `text` with each consituent token wrapped in HTML markup
for later match annotation.
:param text: text to be marked up
:type text: `str`
:rtype: `str`
"""
# Remove characters that should be escaped for XML input (but
# which cause problems when escaped, since they become
# tokens).
text = re.sub(r'[<>&]', '', text)
pattern = r'({})'.format(self._tokenizer.pattern)
replacement = r'\1'
return re.sub(pattern, replacement, text)
PK é³ôFØ&pSõ õ tacl/jitc-old.pyimport csv
import io
import logging
import os
from bokeh.embed import components
from bokeh.charts import Bar
import pandas as pd
from . import constants
from .report import Report
from .statistics_report import StatisticsReport
class JITCProcessor:
"""Generate statistics to list texts from one corpus (referred to
below as "Maybe" and defined in a catalogue file) in order of
similarity to each text in that corpus. Takes into account a
second corpus of texts (referred to below as "No" and defined in a
catalogue file) that are similar to those in the first, but not in
the way(s) that are the subject of the investigation.
Given the two corpora, Maybe and No, the script performs the
following actions:
1. For each text Y in Maybe:
1. Run an intersection between Y and No.
2. For each text M in Maybe (excluding Y):
1. Run an intersect between Y and M.
2. Drop Y results.
3. Run a supplied diff between results from [1.2.2] and
results from [1.1].
4. Get number of tokens in M.
3. Rank and list texts in Maybe in descending order of the
ratio, from [1.2.3], of matching tokens (n-gram size x count)
to total tokens [1.2.5].
4. Concatenate all results from [1.2.3] files.
"""
def __init__ (self, store, corpus, catalogue, maybe_label, tokenizer,
output_dir):
self._logger = logging.getLogger(__name__)
self._corpus = corpus
self._maybe_label = maybe_label
self._maybe_texts = [text for text, label in catalogue.items()
if label == maybe_label]
self._no_texts = [text for text, label in catalogue.items()
if label != maybe_label]
self._no_label = catalogue[self._no_texts[0]]
self._output_dir = output_dir
self._store = store
self._tokenizer = tokenizer
self._ratios = {}
def _drop_no_label_results (self, results, fh, reduce=False):
# Drop results associated with the 'no' label.
results.seek(0)
report = Report(results, self._tokenizer)
report.remove_label(self._no_label)
if reduce:
report.reduce()
results = report.csv(fh)
def process_maybe_text (self, yes_text, maybe_text, work_dir,
yn_results_path):
if maybe_text == yes_text:
return
self._logger.debug(
'Processing "maybe" text {} against "yes" text {}.'.format(
maybe_text, yes_text))
ym_results_path = os.path.join(
work_dir, 'intersect_with_' + maybe_text + '.csv')
catalogue = {yes_text: self._no_label,
maybe_text: self._maybe_label}
self._run_query(ym_results_path, self._store.intersection, [catalogue])
distinct_results_path = os.path.join(
work_dir, 'distinct_' + maybe_text + '.csv')
results = [yn_results_path, ym_results_path]
labels = [self._no_label, self._maybe_label]
self._run_query(distinct_results_path, self._store.diff_supplied,
[results, labels])
stats_path = os.path.join(work_dir, 'stats_' + maybe_text + '.csv')
if not os.path.exists(stats_path):
stats_report = StatisticsReport(self._corpus, self._tokenizer,
distinct_results_path)
stats_report.generate_statistics()
with open(stats_path, mode='w', encoding='utf-8', newline='') as fh:
stats_report.csv(fh)
with open(stats_path, encoding='utf-8', newline='') as fh:
reader = csv.DictReader(fh)
for row in reader:
siglum = row[constants.SIGLUM_FIELDNAME]
ratio = float(row[constants.PERCENTAGE_FIELDNAME])
self._ratios[yes_text][(maybe_text, siglum)] = ratio
def process_yes_text (self, yes_text, no_catalogue):
self._logger.debug('Processing "maybe" text {} as "yes".'.format(
yes_text))
self._ratios[yes_text] = {}
yes_work_dir = os.path.join(self._output_dir, yes_text)
os.makedirs(yes_work_dir, exist_ok=True)
results_path = os.path.join(yes_work_dir, 'intersect_with_no.csv')
self._run_query(results_path, self._store.intersection, [no_catalogue])
for maybe_text in self._maybe_texts:
self.process_maybe_text(yes_text, maybe_text, yes_work_dir,
results_path)
def process_yes_texts (self):
no_catalogue = {text: self._no_label for text in self._no_texts}
data = {}
graphs = {}
for yes_text in self._maybe_texts:
no_catalogue[yes_text] = self._maybe_label
self.process_yes_text(yes_text, no_catalogue)
no_catalogue.pop(yes_text)
values = [ratio for ratio in self._ratios[yes_text].values()]
index = pd.MultiIndex.from_tuples(list(self._ratios[yes_text].keys()),
names=['text', 'siglum'])
series = pd.Series(values, index=index)
data[yes_text] = series
df = pd.DataFrame(data)
# Create a chart that has two bars per text on x-axis: one for
# the percentage of that text that overlaps with the base
# text, and one for the percentage of the base text that
# overlaps with that text. A tooltip showing the values per
# witness would be good.
#
# Create a stacked bar chart that shows the percentage the
# content consisting of shared markers that aren't in the no
# corpus, shared markers that are in the no corpus, and
# unshared markers.
#texts = list(set(index.get_level_values('text')))
#ratios = []
#for text in texts:
# ratio = series[text].max()
# ratios.append(ratio)
#title = 'Shared markers with {}'.format(yes_text)
#bar = Bar(ratios, texts, stacked=False, title=title,
# xlabel='Text', ylabel='% of text sharing markers')
#graphs[yes_text + '-related'] = bar
script, divs = components(graphs)
def _run_query (self, path, query, query_args):
if os.path.exists(path):
return
output_results = io.StringIO(newline='')
query(*query_args, output_fh=output_results)
with open(path, mode='w', encoding='utf-8', newline='') as fh:
self._drop_no_label_results(output_results, fh)
PK Á&ÚF#ïãWV V tacl/stripper.py"""Module containing the Stripper class."""
import logging
import os
import re
from lxml import etree
BASE_WITNESS = 'base'
witnesses_splitter = re.compile(r'ã€|】')
STRIP_XSLT = '''
'''.format(base=BASE_WITNESS)
class Stripper:
"""Class used for preprocessing a corpus of texts by stripping out
all material that is not the textual material proper.
The intention is to keep the stripped text as close in formatting
to the original as possible, including whitespace."""
def __init__ (self, input_dir, output_dir):
self._logger = logging.getLogger(__name__)
self._input_dir = os.path.abspath(input_dir)
self._output_dir = os.path.abspath(output_dir)
self._transform = etree.XSLT(etree.XML(STRIP_XSLT))
self._texts = {}
def get_witnesses (self, source_tree):
"""Returns a list of all witnesses of variant readings in
`source_tree`.
:param source_tree: XML tree of source document
:type source_tree: `etree._ElementTree`
:rtype: `set`
"""
witnesses = set([BASE_WITNESS])
witness_values = source_tree.xpath('//app/rdg[@wit]/@wit')
for witness_value in witness_values:
for witness in witnesses_splitter.split(witness_value):
if witness:
witnesses.add(witness)
return witnesses
def _output_file (self, text_name, witnesses):
text_dir = os.path.join(self._output_dir, text_name)
try:
os.makedirs(text_dir)
except OSError as err:
logging.error('Could not create output directory: {}'.format(
err))
raise
for witness in witnesses.keys():
witness_file_path = os.path.join(
text_dir, '{}.txt'.format(witness))
with open(witness_file_path, 'wb') as output_file:
output_file.write(witnesses[witness].encode('utf-8'))
def strip_files (self):
if not os.path.exists(self._output_dir):
try:
os.makedirs(self._output_dir)
except OSError as err:
self._logger.error(
'Could not create output directory: {}'.format(err))
raise
for dirpath, dirnames, filenames in os.walk(self._input_dir):
for filename in filenames:
if os.path.splitext(filename)[1] == '.xml':
text_name, witnesses = self.strip_file(
os.path.join(dirpath, filename))
self._output_file(text_name, witnesses)
def strip_file (self, filename):
file_path = os.path.join(self._input_dir, filename)
text_name = os.path.splitext(os.path.basename(filename))[0]
stripped_file_path = os.path.join(self._output_dir, text_name)
self._logger.info('Stripping file {} into {}'.format(
file_path, stripped_file_path))
try:
tei_doc = etree.parse(file_path)
except etree.XMLSyntaxError:
logging.warning('XML file "{}" is invalid'.format(filename))
return
text_witnesses = self._texts.setdefault(stripped_file_path, {})
for witness in self.get_witnesses(tei_doc):
witness_param = "'{}'".format(witness)
text = str(self._transform(tei_doc, witness=witness_param))
text_witnesses[witness] = text
return text_name, text_witnesses
PK Á&ÚF†¾w¶P P tacl/catalogue.pyimport csv
import os
from .constants import CATALOGUE_TEXT_RELABELLED_ERROR
from .exceptions import MalformedCatalogueError
class Catalogue (dict):
def generate (self, path, label):
"""Creates default data from the corpus at `path`, marking all
texts with `label`.
:param path: path to a corpus directory
:type path: `str`
:param label: label to categorise each text as
:type label: `str`
"""
for filename in os.listdir(path):
self[filename] = label
def load (self, path):
"""Loads the data from `path` into the catalogue.
:param path: path to catalogue file
:type path: `str`
"""
fieldnames = ['text', 'label']
with open(path, 'r', encoding='utf-8', newline='') as fh:
reader = csv.DictReader(fh, delimiter=' ', fieldnames=fieldnames,
skipinitialspace=True)
for row in reader:
text, label = row['text'], row['label']
if label:
if text in self:
raise MalformedCatalogueError(
CATALOGUE_TEXT_RELABELLED_ERROR.format(text))
self[text] = label
def save (self, path):
"""Saves this catalogue's data to `path`.
:param path: file path to save catalogue data to
:type path: `str`
"""
writer = csv.writer(open(path, 'w', newline=''), delimiter=' ')
rows = list(self.items())
rows.sort(key=lambda x: x[0])
writer.writerows(rows)
PK È PGSe1•ø? ø? tacl/jitc.pyimport csv
import io
import json
import logging
import math
import os
import shutil
import pandas as pd
from pkg_resources import resource_filename, resource_listdir
from . import constants
from .report import Report
from .statistics_report import StatisticsReport
class JITCProcessor:
"""Generate statistics to list texts from one corpus (referred to
below as "Maybe" and defined in a catalogue file) in order of
similarity to each text in that corpus. Takes into account a
second corpus of texts (referred to below as "No" and defined in a
catalogue file) that are similar to those in the first, but not in
the way(s) that are the subject of the investigation.
Given the two corpora, Maybe and No, the script performs the
following actions:
1. For each text Y in Maybe:
1. Run an intersection between Y and No.
2. For each text M in Maybe (excluding Y):
1. Run an intersect between Y and M.
2. Drop Y results.
3. Run a supplied diff between results from [1.2.2] and
results from [1.1].
4. Get number of tokens in M.
3. Rank and list texts in Maybe in descending order of the
ratio, from [1.2.3], of matching tokens (n-gram size x count)
to total tokens [1.2.4].
4. Concatenate all results from [1.2.3] files.
Note that in the above, when a text is treated as Y, its different
witnesses are not treated separately. The statistics derived from
queries including it are those that treat all of its witnesses
together; eg, if two n-grams in a witness of M are found only in
two different witnesses of Y, they will both be counted as shared.
"""
def __init__ (self, store, corpus, catalogue, maybe_label, tokenizer,
output_dir):
self._logger = logging.getLogger(__name__)
self._corpus = corpus
self._maybe_label = maybe_label
self._maybe_texts = [text for text, label in catalogue.items()
if label == maybe_label]
self._no_texts = [text for text, label in catalogue.items()
if label != maybe_label]
self._no_label = catalogue[self._no_texts[0]]
self._output_dir = output_dir
self._output_data_dir = os.path.join(self._output_dir, 'data')
self._store = store
self._tokenizer = tokenizer
self._stats = {}
self._ym_intersects_dir = os.path.join(self._output_data_dir,
'ym_intersects')
def _copy_static_assets (self, output_dir):
for asset in resource_listdir(__name__, 'assets'):
filename = resource_filename(__name__, 'assets/{}'.format(
asset))
shutil.copy2(filename, output_dir)
def _create_breakdown_chart (self, data, text, output_dir):
# Create a stacked bar chart that shows the percentage of the
# content consisting of shared tokens that aren't in the no
# corpus, shared tokens that are also in the no corpus, and
# unshared tokens.
chart_data = data.loc[text].sort('shared', ascending=False)[
['shared', 'unique', 'common']]
csv_path = os.path.join(output_dir, 'breakdown_{}.csv'.format(
text))
chart_data.to_csv(csv_path)
def _create_chord_chart (self, data, output_dir):
matrix = []
chord_data = data.unstack('main_text')['shared']
for index, row_data in chord_data.fillna(value=0).iterrows():
matrix.append([value / 100 for value in row_data])
colours = generate_colours(len(self._maybe_texts))
colour_texts = [{'name': text, 'colour': colour} for text, colour
in zip(chord_data, colours)]
json_data = json.dumps({'texts': colour_texts, 'matrix': matrix})
with open(os.path.join(output_dir, 'chord_data.js'), 'w') as fh:
fh.write('var chordData = {}'.format(json_data))
def _create_matrix_chart (self, data, output_dir):
nodes = [{'name': name, 'group': 1} for name in self._maybe_texts]
weights = data.stack().unstack('related_text').max()
seen = []
links = []
for (source, target), weight in weights.iteritems():
if target not in seen and target != source:
seen.append(source)
links.append({'source': self._maybe_texts.index(source),
'target': self._maybe_texts.index(target),
'value': weight})
json_data = json.dumps({'nodes': nodes, 'links': links})
with open(os.path.join(output_dir, 'matrix_data.js'), 'w') as fh:
fh.write('var matrixData = {}'.format(json_data))
def _create_related_chart (self, data, text, output_dir):
# Create a chart that has two bars per text on x-axis: one for
# the percentage of that text that overlaps with the base
# text, and one for the percentage of the base text that
# overlaps with that text. A tooltip showing the values per
# witness would be good.
chart_data = data[text].dropna().sort('shared_related_text',
ascending=False)
csv_path = os.path.join(output_dir, 'related_{}.csv'.format(text))
chart_data.to_csv(csv_path)
def _drop_no_label_results (self, results, fh):
"""Writes `results` to `fh` minus those results associated with the
'no' label.
:param results: results to be manipulated
:type results: file-like object
:param fh: output destination
:type fh: file-like object
"""
results.seek(0)
report = Report(results, self._tokenizer)
report.remove_label(self._no_label)
report.csv(fh)
def _generate_statistics (self, out_path, results_path):
"""Write a statistics report for `results_path` to `out_path`."""
if not os.path.exists(out_path):
report = StatisticsReport(self._corpus, self._tokenizer,
results_path)
report.generate_statistics()
with open(out_path, mode='w', encoding='utf-8', newline='') as fh:
report.csv(fh)
def _get_reversed_data (self, data):
reverse_data = data.unstack('main_text')['shared']
tuples = list(zip(['shared_related_text'] * len(reverse_data.columns),
reverse_data.columns))
reverse_data.columns = pd.MultiIndex.from_tuples(
tuples, names=['text', 'main_text'])
for text in reverse_data['shared_related_text'].columns:
reverse_data['shared_base_text', text] = reverse_data[
'shared_related_text'].loc[text].tolist()
return reverse_data.swaplevel('text', 'main_text', axis=1)
def _process_maybe_text (self, yes_text, maybe_text, work_dir,
yn_results_path):
if maybe_text == yes_text:
return
self._logger.info(
'Processing "maybe" text {} against "yes" text {}.'.format(
maybe_text, yes_text))
for siglum in self._corpus.get_sigla(maybe_text):
witness = (maybe_text, siglum)
self._stats[yes_text]['common'][witness] = 0
self._stats[yes_text]['shared'][witness] = 0
self._stats[yes_text]['unique'][witness] = 100
texts = [yes_text, maybe_text]
texts.sort()
ym_results_path = os.path.join(
self._ym_intersects_dir, '{}_intersect_{}.csv'.format(*texts))
catalogue = {yes_text: self._no_label,
maybe_text: self._maybe_label}
self._run_query(ym_results_path, self._store.intersection, [catalogue],
False)
intersect_stats_path = os.path.join(
work_dir, 'stats_intersect_{}.csv'.format(maybe_text))
self._generate_statistics(intersect_stats_path, ym_results_path)
with open(intersect_stats_path, encoding='utf-8', newline='') as fh:
reader = csv.DictReader(fh)
for row in reader:
if row[constants.NAME_FIELDNAME] == maybe_text:
witness = (maybe_text, row[constants.SIGLUM_FIELDNAME])
ratio = float(row[constants.PERCENTAGE_FIELDNAME])
self._stats[yes_text]['common'][witness] = ratio
self._stats[yes_text]['unique'][witness] -= ratio
distinct_results_path = os.path.join(
work_dir, 'distinct_{}.csv'.format(maybe_text))
results = [yn_results_path, ym_results_path]
labels = [self._no_label, self._maybe_label]
self._run_query(distinct_results_path, self._store.diff_supplied,
[results, labels])
diff_stats_path = os.path.join(work_dir,
'stats_diff_{}.csv'.format(maybe_text))
self._generate_statistics(diff_stats_path, distinct_results_path)
with open(diff_stats_path, encoding='utf-8', newline='') as fh:
reader = csv.DictReader(fh)
for row in reader:
if row[constants.NAME_FIELDNAME] == maybe_text:
witness = (maybe_text, row[constants.SIGLUM_FIELDNAME])
ratio = float(row[constants.PERCENTAGE_FIELDNAME])
self._stats[yes_text]['shared'][witness] = ratio
self._stats[yes_text]['common'][witness] -= ratio
def _process_yes_text (self, yes_text, no_catalogue, output_dir):
self._logger.info('Processing "maybe" text {} as "yes".'.format(
yes_text))
self._stats[yes_text] = {'common': {}, 'shared': {}, 'unique': {}}
yes_work_dir = os.path.join(output_dir, yes_text)
os.makedirs(yes_work_dir, exist_ok=True)
results_path = os.path.join(yes_work_dir, 'intersect_with_no.csv')
self._run_query(results_path, self._store.intersection, [no_catalogue])
for maybe_text in self._maybe_texts:
self._process_maybe_text(yes_text, maybe_text, yes_work_dir,
results_path)
def process (self):
no_catalogue = {text: self._no_label for text in self._no_texts}
data = {}
os.makedirs(self._ym_intersects_dir, exist_ok=True)
for yes_text in self._maybe_texts:
no_catalogue[yes_text] = self._maybe_label
self._process_yes_text(yes_text, no_catalogue,
self._output_data_dir)
no_catalogue.pop(yes_text)
for scope in ('shared', 'common', 'unique'):
text_data = self._stats[yes_text][scope]
# QAZ: Check that keys() and values() will always
# return items in the same order when called
# consecutively like this.
index = pd.MultiIndex.from_tuples(
list(text_data.keys()), names=['related_text', 'siglum'])
data[(yes_text, scope)] = pd.Series(list(text_data.values()),
index=index)
full_data = pd.DataFrame(data)
full_data.columns.names = ['main_text', 'scope']
full_data = full_data.stack('main_text').swaplevel(
'main_text', 'siglum').swaplevel('related_text', 'main_text')
grouped = full_data.groupby(level=['main_text', 'related_text'],
axis=0, group_keys=False)
max_data = grouped.apply(lambda x: x.loc[x['shared'].idxmax()])
reverse_data = self._get_reversed_data(max_data)
report_data_dir = os.path.join(self._output_dir, 'report_data')
os.makedirs(report_data_dir, exist_ok=True)
report_assets_dir = os.path.join(self._output_dir, 'report_assets')
os.makedirs(report_assets_dir, exist_ok=True)
# Matrix chart.
self._create_matrix_chart(reverse_data, report_data_dir)
# Chord chart.
self._create_chord_chart(max_data, report_data_dir)
# Individual text bar charts.
texts = []
scripts = []
toc = []
export_data = full_data.unstack('main_text').swaplevel(
'main_text', 'scope', axis=1)
export_data.index.names = ['related text', 'siglum']
for index, text in enumerate(self._maybe_texts):
self._create_breakdown_chart(max_data, text, report_data_dir)
self._create_related_chart(reverse_data, text, report_data_dir)
table = export_data[text].dropna().to_html()
toc.append(TOC_HTML.format(index=index, text=text))
texts.append(TEXT_HTML.format(index=index, table=table, text=text))
scripts.append(SCRIPT_HTML.format(index=index, sep=os.sep,
text=text))
with open(os.path.join(self._output_dir, 'report.html'), 'w') as fh:
fh.write(CHART_HTML.format(scripts='\n'.join(scripts),
sep=os.sep,
texts='\n'.join(texts),
toc='\n'.join(toc)))
self._copy_static_assets(report_assets_dir)
def _run_query (self, path, query, query_args, drop_no=True):
if os.path.exists(path):
return
output_results = io.StringIO(newline='')
query(*query_args, output_fh=output_results)
with open(path, mode='w', encoding='utf-8', newline='') as fh:
if drop_no:
self._drop_no_label_results(output_results, fh)
else:
fh.write(output_results.getvalue())
def rgb_colour (h, f):
"""Convert a colour specified by h-value and f-value to an RGB string."""
v = 1
p = 0
if h == 0:
colour = v, f, p
elif h == 1:
colour = 1 - f, v, p
elif h == 2:
colour = p, v, f
elif h == 3:
colour = p, 1 - f, v
elif h == 4:
colour = f, p, v
elif h == 5:
colour = v, p, 1 - f
return 'rgb({}, {}, {})'.format(*[round(value * 255) for value in colour])
def generate_colours (n):
"""Return a list of distinct colours, each of which is represented as
an RGB string suitable for use in CSS."""
hues = [360 / n * i for i in range(n)]
hs = (math.floor(hue / 60) % 6 for hue in hues)
fs = (hue / 60 - math.floor(hue / 60) for hue in hues)
return [rgb_colour(h, f) for h, f in zip(hs, fs)]
CHART_HTML = '''
JitC Report