/usr/lib/python3/dist-packages/cligj/features.py is in python3-cligj 0.4.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | from itertools import chain
import json
import re
import click
def normalize_feature_inputs(ctx, param, features_like):
""" Click callback which accepts the following values:
* Path to file(s), each containing single FeatureCollection or Feature
* Coordinate pair(s) of the form "[0, 0]" or "0, 0" or "0 0"
* if not specified or '-', process STDIN stream containing
- line-delimited features
- ASCII Record Separator (0x1e) delimited features
- FeatureCollection or Feature object
and yields GeoJSON Features.
"""
if len(features_like) == 0:
features_like = ('-',)
for flike in features_like:
try:
# It's a file/stream with GeoJSON
src = iter(click.open_file(flike, mode='r'))
for feature in iter_features(src):
yield feature
except IOError:
# It's a coordinate string
coords = list(coords_from_query(flike))
feature = {
'type': 'Feature',
'properties': {},
'geometry': {
'type': 'Point',
'coordinates': coords}}
yield feature
def iter_features(src):
"""Yield features from a src that may be either a GeoJSON feature
text sequence or GeoJSON feature collection."""
first_line = next(src)
# If input is RS-delimited JSON sequence.
if first_line.startswith(u'\x1e'):
buffer = first_line.strip(u'\x1e')
for line in src:
if line.startswith(u'\x1e'):
if buffer:
feat = json.loads(buffer)
yield feat
buffer = line.strip(u'\x1e')
else:
buffer += line
else:
feat = json.loads(buffer)
yield feat
else:
try:
feat = json.loads(first_line)
assert feat['type'] == 'Feature'
yield feat
for line in src:
feat = json.loads(line)
yield feat
except (TypeError, KeyError, AssertionError, ValueError):
text = "".join(chain([first_line], src))
feats = json.loads(text)
if feats['type'] == 'Feature':
yield feats
elif feats['type'] == 'FeatureCollection':
for feat in feats['features']:
yield feat
def iter_query(query):
"""Accept a filename, stream, or string.
Returns an iterator over lines of the query."""
try:
itr = click.open_file(query).readlines()
except IOError:
itr = [query]
return itr
def coords_from_query(query):
"""Transform a query line into a (lng, lat) pair of coordinates."""
try:
coords = json.loads(query)
except ValueError:
vals = re.split(r"\,*\s*", query.strip())
coords = [float(v) for v in vals]
return tuple(coords[:2])
def normalize_feature_objects(feature_objs):
"""Takes an iterable of GeoJSON-like Feature mappings or
an iterable of objects with a geo interface and
normalizes it to the former."""
for obj in feature_objs:
if hasattr(obj, "__geo_interface__") and \
'type' in obj.__geo_interface__.keys() and \
obj.__geo_interface__['type'] == 'Feature':
yield obj.__geo_interface__
elif isinstance(obj, dict) and 'type' in obj and \
obj['type'] == 'Feature':
yield obj
else:
raise ValueError("Did not recognize object {0}"
"as GeoJSON Feature".format(obj))
|