Completed
Pull Request — master (#127)
by Jasper
01:07
created

JsonFile.bySubject()   B

Complexity

Conditions 5

Size

Total Lines 12

Duplication

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