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
|
from typing import Dict, List, Iterable, Tuple, Any, Iterator
import scipy.sparse as sparse
import numpy as np
import loompy
import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore")
import h5py
from .utils import compare_loom_spec_version
class GlobalAttributeManager(object):
def __init__(self, f: h5py.File) -> None:
setattr(self, "!f", f)
storage: Dict[str, str] = {}
setattr(self, "!storage", storage)
if "attrs" not in self.f:
for key, val in f.attrs.items():
materialized = loompy.materialize_attr_values(val)
self.__dict__["storage"][key] = materialized
else:
for key, val in f["attrs"].items():
materialized = loompy.materialize_attr_values(val[()])
self.__dict__["storage"][key] = materialized
def keys(self) -> List[str]:
return list(self.__dict__["storage"].keys())
def items(self) -> Iterable[Tuple[str, sparse.coo_matrix]]:
for key in self.keys():
yield (key, self[key])
def __len__(self) -> int:
return len(self.keys())
def __contains__(self, name: str) -> bool:
return name in self.keys()
def __iter__(self) -> Iterator[str]:
for key in self.keys():
yield key
def __getitem__(self, thing: Any) -> np.ndarray:
return self.__getattr__(thing)
def __getattr__(self, name: str) -> np.ndarray:
try:
return self.__dict__["storage"][name]
except KeyError:
if self.f is not None:
if loompy.compare_loom_spec_version(self.f, "3.0.0") < 0:
if name in self.f.attrs:
val = self.f.attrs[name]
else:
raise AttributeError(f"File has no global attribute '{name}'")
else:
if name in self.f["attrs"]:
val = self.f["attrs"][name]
else:
raise AttributeError(f"File has no global attribute '{name}'")
materialized = loompy.materialize_attr_values(val)
self.__dict__["storage"][name] = materialized
return materialized
def __setitem__(self, name: str, val: Any) -> None:
return self.__setattr__(name, val)
def __setattr__(self, name: str, val: Any) -> None:
if name.startswith("!"):
super(GlobalAttributeManager, self).__setattr__(name[1:], val)
elif "/" in name:
raise KeyError("Attribute name cannot contain slash (/)")
else:
if self.f is not None:
if loompy.compare_loom_spec_version(self.f, "3.0.0") < 0 and "attrs" not in self.f["/"]:
normalized = loompy.normalize_attr_values(val, False)
self.f.attrs[name] = normalized
self.f.flush()
val = self.f.attrs[name]
# Read it back in to ensure it's synced and normalized
normalized = loompy.materialize_attr_values(val)
self.__dict__["storage"][name] = normalized
else:
normalized = loompy.normalize_attr_values(val, True)
if name in self.f["attrs"]:
del self.f["attrs"][name]
if not np.isscalar(normalized) and normalized.dtype == np.object_:
self.ds._file.create_dataset("attrs/" + name, data=normalized, dtype=h5py.special_dtype(vlen=str))
else:
self.f["attrs"][name] = normalized
self.f.flush()
val = self.f["attrs"][name][()]
# Read it back in to ensure it's synced and normalized
normalized = loompy.materialize_attr_values(val)
self.__dict__["storage"][name] = normalized
def __delattr__(self, name: str) -> None:
if name.startswith("!"):
super(GlobalAttributeManager, self).__delattr__(name[1:])
else:
if self.f is not None:
if loompy.compare_loom_spec_version(self.f, "3.0.0") < 0:
if name in self.f.attrs:
del self.f.attrs[name]
else:
if name in self.f["attrs"]:
del self.f["attrs"][name]
del self.__dict__["storage"][name]
def get(self, name: str, default: Any = None) -> np.ndarray:
"""
Return the value for a named attribute if it exists, else default.
If default is not given, it defaults to None, so that this method never raises a KeyError.
"""
if name in self:
return self[name]
else:
return default
|
