Source code for textblob.formats
# -*- coding: utf-8 -*-
"""File formats for training and testing data.
Includes a registry of valid file formats. New file formats can be added to the
registry like so: ::
from textblob import formats
class PipeDelimitedFormat(formats.DelimitedFormat):
delimiter = '|'
formats.register('psv', PipeDelimitedFormat)
Once a format has been registered, classifiers will be able to read data files with
that format. ::
from textblob.classifiers import NaiveBayesAnalyzer
with open('training_data.psv', 'r') as fp:
cl = NaiveBayesAnalyzer(fp, format='psv')
"""
from __future__ import absolute_import
import json
from collections import OrderedDict
from textblob.compat import PY2, csv
from textblob.utils import is_filelike
DEFAULT_ENCODING = 'utf-8'
[docs]class CSV(DelimitedFormat):
"""CSV format. Assumes each row is of the form ``text,label``.
::
Today is a good day,pos
I hate this car.,pos
"""
delimiter = ","
[docs]class TSV(DelimitedFormat):
"""TSV format. Assumes each row is of the form ``text\tlabel``.
"""
delimiter = "\t"
[docs]class JSON(BaseFormat):
"""JSON format.
Assumes that JSON is formatted as an array of objects with ``text`` and
``label`` properties.
::
[
{"text": "Today is a good day.", "label": "pos"},
{"text": "I hate this car.", "label": "neg"}
]
"""
def __init__(self, fp, **kwargs):
BaseFormat.__init__(self, fp, **kwargs)
self.dict = json.load(fp)
[docs] def to_iterable(self):
"""Return an iterable object from the JSON data."""
return [(d['text'], d['label']) for d in self.dict]
[docs] @classmethod
def detect(cls, stream):
"""Return True if stream is valid JSON."""
try:
json.loads(stream)
return True
except ValueError:
return False
_registry = OrderedDict([
('csv', CSV),
('json', JSON),
('tsv', TSV),
])
[docs]def detect(fp, max_read=1024):
"""Attempt to detect a file's format, trying each of the supported
formats. Return the format class that was detected. If no format is
detected, return ``None``.
"""
if not is_filelike(fp):
return None
for Format in _registry.values():
if Format.detect(fp.read(max_read)):
fp.seek(0)
return Format
fp.seek(0)
return None
[docs]def get_registry():
"""Return a dictionary of registered formats."""
return _registry
[docs]def register(name, format_class):
"""Register a new format.
:param str name: The name that will be used to refer to the format, e.g. 'csv'
:param type format_class: The format class to register.
"""
get_registry()[name] = format_class