Completed
Push — master ( 4b6b67...211fa1 )
by Jasper
9s
created

JsonFile.getSeries()   B

Complexity

Conditions 6

Size

Total Lines 19

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 6
dl 0
loc 19
rs 8
c 0
b 0
f 0
1
import os
2
from operator import itemgetter
3
from niprov.dependencies import Dependencies
4
5
6
class JsonFile(object):
7
    """Stores provenance in a local text file encoded as json.
8
    """
9
10
    def __init__(self, dependencies=Dependencies()):
11
        self.filesys = dependencies.getFilesystem()
12
        self.json = dependencies.getSerializer()
13
        self.factory = dependencies.getFileFactory()
14
        self.pictureCache = dependencies.getPictureCache()
15
        url = dependencies.getConfiguration().database_url
16
        self.datafile = os.path.expanduser(url)
17
18
    def serializeAndWrite(self, images):
19
        jsonstr = self.json.serializeList(images)
20
        self.filesys.write(self.datafile, jsonstr)
21
22
    def add(self, image):
23
        """Add the provenance for one file to storage.
24
25
        Args:
26
            image (:class:`.BaseFile`): Image file to store.
27
        """
28
        current = self.all()
29
        current.append(image)
30
        self.serializeAndWrite(current)
31
        self.pictureCache.saveToDisk(for_=image)
32
33
    def update(self, image):
34
        """Save changed provenance for this file..
35
36
        Args:
37
            image (:class:`.BaseFile`): Image file that has changed.
38
        """
39
        current = self.all()
40
        for r in range(len(current)):
41
            if current[r].location.toString() == image.location.toString():
42
                current[r] = image
43
        self.serializeAndWrite(current)
44
45
    def all(self):
46
        """Retrieve all known provenance from storage.
47
48
        Returns:
49
            list: List of provenance for known files.
50
        """
51
        try:
52
            jsonstr = self.filesys.read(self.datafile)
53
        except IOError:
54
            return []
55
        return self.json.deserializeList(jsonstr)
56
57
    def knowsByLocation(self, locationString):
58
        """Whether the file at this path has provenance associated with it.
59
60
        Returns:
61
            bool: True if provenance is available for that path.
62
        """
63
        try:
64
            self.byLocation(locationString)
65
        except IndexError:
66
            return False
67
        return True
68
69
    def knows(self, image):
70
        """Whether this file has provenance associated with it.
71
72
        Returns:
73
            bool: True if provenance is available for this image.
74
        """
75
        try:
76
            self.byLocation(image.path)
77
        except IndexError:
78
            return False
79
        return True
80
81
    def knowsSeries(self, image):
82
        """Whether the series that this file is part of has provenance 
83
        associated with it.
84
85
        Args:
86
            image (:class:`.BaseFile`): File for which the series is sought.
87
88
        Returns:
89
            bool: True if provenance is available for this series.
90
        """
91
        try:
92
            self.getSeries(image)
93
        except IndexError:
94
            return False
95
        return True
96
97
    def byLocation(self, locationString):
98
        """Get the provenance for a file at the given location. 
99
100
        In the case of a dicom series, this returns the provenance for the 
101
        series.
102
103
        Args:
104
            locationString (str): Location of the image file.
105
106
        Returns:
107
            dict: Provenance for one image file.
108
        """
109
        for image in self.all():
110
            if image.location.toString() == locationString:
111
                return image
112
            elif 'filesInSeries' in image.provenance and (
113
                locationString in image.provenance['filesInSeries']):
114
                return image
115
        else:
116
            raise IndexError('No file with that path known.')
117
118
    def byLocations(self, listOfLocations):
119
        return [f for f in self.all() if f.location.toString() in listOfLocations]
120
121
    def getSeries(self, image):
122
        """Get the object that carries provenance for the series that the image 
123
        passed is in. 
124
125
        Args:
126
            image (:class:`.DicomFile`): File that is part of a series.
127
128
        Returns:
129
            :class:`.DicomFile`: Image object that caries provenance for the series.
130
        """
131
        if image.getSeriesId() is None:
132
            raise IndexError('Image has no series id.')
133
        seriesId = image.getSeriesId()
134
        for image in self.all():
135
            if 'seriesuid' in image.provenance and (
136
                image.provenance['seriesuid'] == seriesId):
137
                return image
138
        else:
139
            raise IndexError('No provenance record for that series.')
140
141
    def updateApproval(self, fpath, approvalStatus):
142
        img = self.byLocation(fpath)
143
        img.provenance['approval'] = approvalStatus
144
        self.update(img)
145
146
    def latest(self, n=20):
147
        def dateTimeAdded(img):
148
            return img.provenance.get('added')
149
        sortedImages = sorted(self.all(), key=dateTimeAdded, reverse=True)
150
        return sortedImages[:n]
151
152
    def statistics(self):
153
        stats = {}
154
        images = self.all()
155
        stats['count'] = len(images)
156
        sizes = [img.provenance['size'] for img in images if 'size' in img.provenance]
157
        stats['totalsize'] = sum(sizes)
158
        return stats
159
160
    def byId(self, uid):
161
        for image in self.all():
162
            if image.provenance['id'] == uid:
163
                return image
164
        else:
165
            raise IndexError('No file with that path known.')
166
167
    def byParents(self, listOfParentLocations):
168
        return [f for f in self.all() if set(listOfParentLocations).intersection(
169
            f.provenance.get('parents',[]))]
170
171
    def inquire(self, query):
172
        field = query.getFields()[0]
173
        matches = []
174
        for image in self.all():
175
            if field.name in image.provenance:
176
                if field.all:
177
                    if not image.provenance[field.name] in matches:
178
                        matches.append(image.provenance[field.name])
179
                else:
180
                    if image.provenance[field.name] == field.value:
181
                        matches.append(image)
182
        return matches
183
184
    def search(self, text):
185
        fields = ['location','user','subject','project','protocol',
186
                  'transformation','technique','modality']
187
        matches = []
188
        for image in self.all():
189
            score = 0
190
            for word in text.split():
191
                for field in fields:
192
                    if field in image.provenance:
193
                        score += image.provenance[field].count(word)
194
            if score > 0:
195
                matches.append((image, score))
196
        sortedResults = sorted(matches, key=itemgetter(1), reverse=True)
197
        return [i for i, s in sortedResults[:20]]
198
        
199
200