1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
|
# -*- coding: utf-8 -*-
"""
Drop-in replacement for Python Sniffer object.
Author: Gertjan van den Burg
"""
from enum import Enum
from io import StringIO
from typing import Dict
from typing import Iterable
from typing import Optional
from typing import Union
from .consistency import ConsistencyDetector
from .dialect import SimpleDialect
from .exceptions import NoDetectionResult
from .normal_form import detect_dialect_normal
from .read import reader
class DetectionMethod(str, Enum):
"""Possible detection methods
Valid options are `"auto"` (the default for :class:`Detector.detect`),
`"normal"`, or `"consistency"`. The `"auto"` option first attempts to
detect the dialect using normal-form detection, and uses the consistency
measure if normal-form detection is inconclusive. The `"normal"` method
uses normal-form detection excllusively, and the `"consistency"` method
uses the consistency measure exclusively.
"""
AUTO = "auto"
NORMAL = "normal"
CONSISTENCY = "consistency"
class Detector:
"""
Detect the Dialect of CSV files with normal forms or the data consistency
measure. This class provides a drop-in replacement for the Python dialect
Sniffer from the standard library.
Note
----
We call the object ``Detector`` just to mark the difference in the
implementation and avoid naming issues. You can import it as ``from ccsv
import Sniffer`` nonetheless.
"""
def sniff(
self,
sample: str,
delimiters: Optional[Iterable[str]] = None,
verbose: bool = False,
) -> Optional[SimpleDialect]:
# Compatibility method for Python
return self.detect(sample, delimiters=delimiters, verbose=verbose)
def detect(
self,
sample: str,
delimiters: Optional[Iterable[str]] = None,
verbose: bool = False,
method: Union[DetectionMethod, str] = DetectionMethod.AUTO,
skip: bool = True,
) -> Optional[SimpleDialect]:
"""Detect the dialect of a CSV file
This method detects the dialect of the CSV file using the specified
detection method.
Parameters
----------
sample : str
A sample of text from the CSV file. For best results and if time
allows, use the entire contents of the CSV file as the sample.
delimiters : Optional[Iterable[str]]
Set of delimiters to consider for dialect detection. The potential
dialects will be constructed by analyzing the sample and these
delimiters. If omitted, the set of potential delimiters will be
constructed from the sample.
verbose : bool
Enable verbose mode.
method : Union[DetectionMethod, str]
The method to use for dialect detection. Possible values are
:class:`DetectionMethod` instances or strings that can be cast to
as such an enum.
skip : bool
Whether to skip potential dialects that have too low a pattern
score in the consistency detection. See
:func:`ConsistencyDetector.compute_consistency_scores` for more
details.
Returns
-------
dialect : Optional[SimpleDialect]
The detected dialect. Can be `None` if dialect detection was
inconclusive.
"""
method = DetectionMethod(method) if isinstance(method, str) else method
if delimiters is not None:
delimiters = list(delimiters)
if method == DetectionMethod.NORMAL or method == DetectionMethod.AUTO:
if verbose:
print("Running normal form detection ...", flush=True)
dialect = detect_dialect_normal(
sample, delimiters=delimiters, verbose=verbose
)
if dialect is not None:
self.method_ = DetectionMethod.NORMAL
return dialect
self.method_ = DetectionMethod.CONSISTENCY
consistency_detector = ConsistencyDetector(skip=skip, verbose=verbose)
if verbose:
print("Running data consistency measure ...", flush=True)
return consistency_detector.detect(sample, delimiters=delimiters)
def has_header(self, sample: str, max_rows_to_check: int = 20) -> bool:
"""Detect if a file has a header from a sample.
This function is copied from CPython! The only change we've made is to
use our dialect detection method.
"""
# Creates a dictionary of types of data in each column. If any
# column is of a single type (say, integers), *except* for the first
# row, then the first row is presumed to be labels. If the type
# can't be determined, it is assumed to be a string in which case
# the length of the string is the determining factor: if all of the
# rows except for the first are the same length, it's a header.
# Finally, a 'vote' is taken at the end for each column, adding or
# subtracting from the likelihood of the first row being a header.
dialect = self.sniff(sample)
if dialect is None:
raise NoDetectionResult
rdr = reader(StringIO(sample), dialect)
header = next(rdr) # assume first row is header
columns = len(header)
columnTypes: Dict[int, Optional[Union[int, type]]] = {}
for i in range(columns):
columnTypes[i] = None
thisType: Union[int, type]
checked = 0
for row in rdr:
# arbitrary number of rows to check, to keep it sane
if checked > max_rows_to_check:
break
checked += 1
if len(row) != columns:
continue # skip rows that have irregular number of columns
for col in list(columnTypes.keys()):
for thisType in [int, float, complex]:
try:
thisType(row[col])
break
except (ValueError, OverflowError):
pass
else:
# fallback to length of string
thisType = len(row[col])
if thisType != columnTypes[col]:
if columnTypes[col] is None: # add new column type
columnTypes[col] = thisType
else:
# type is inconsistent, remove column from
# consideration
del columnTypes[col]
# finally, compare results against first row and "vote"
# on whether it's a header
hasHeader = 0
for col, colType in columnTypes.items():
if isinstance(colType, int): # it's a length
if len(header[col]) != colType:
hasHeader += 1
else:
hasHeader -= 1
else: # attempt typecast
if colType is None:
hasHeader += 1
continue
try:
colType(header[col])
except (ValueError, TypeError):
hasHeader += 1
else:
hasHeader -= 1
return hasHeader > 0
|
