source: OpenWorkouts-current/ow/models/workout.py @ 78af3d1

currentfeature/docs
Last change on this file since 78af3d1 was 78af3d1, checked in by Borja Lopez <borja@…>, 5 years ago

Fix permissions. From now on users can see (and edit, delete, etc) their own data

  • Property mode set to 100644
File size: 14.9 KB
Line 
1import os
2from datetime import datetime, timedelta, timezone
3from decimal import Decimal
4
5import pytz
6import gpxpy
7from repoze.folder import Folder
8from pyramid.security import Allow, Deny, Everyone, ALL_PERMISSIONS
9
10from ow.utilities import (
11    GPXMinidomParser,
12    copy_blob,
13    create_blob,
14    mps_to_kmph,
15    save_map_screenshot,
16    timedelta_to_hms
17)
18
19from ow.fit import Fit
20
21
22class Workout(Folder):
23
24    __parent__ = __name__ = None
25
26    def __acl__(self):
27        """
28        If the workout is owned by a given user, only that user have access to
29        it (for now). If not, everybody can view it, only admins can edit it.
30        """
31        uid = self.__parent__.uid
32        permissions = [
33            (Allow, str(uid), 'view'),
34            (Allow, str(uid), 'edit'),
35            (Allow, str(uid), 'delete'),
36            (Deny, Everyone, ALL_PERMISSIONS)
37        ]
38        return permissions
39
40    def __init__(self, **kw):
41        super(Workout, self).__init__()
42        # we do store datetime objects with UTC timezone
43        self.start = kw.get('start', datetime.now(timezone.utc))
44        self.sport = kw.get('sport', 'unknown')  # string
45        self.title = kw.get('title', '')  # unicode string
46        self.notes = kw.get('notes', '')  # unicode string
47        self.duration = kw.get('duration', None)  # a timedelta object
48        self.distance = kw.get('distance', None)  # kilometers, Decimal
49        self.speed = kw.get('speed', {})
50        self.hr_min = kw.get('hr_min', None)  # bpm, Decimal
51        self.hr_max = kw.get('hr_max', None)  # bpm, Decimal
52        self.hr_avg = kw.get('hr_avg', None)  # bpm, Decimal
53        self.uphill = kw.get('uphill', None)
54        self.downhill = kw.get('downhill', None)
55        self.cad_min = kw.get('cad_min', None)
56        self.cad_max = kw.get('cad_max', None)
57        self.cad_avg = kw.get('cad_avg', None)
58        self.atemp_min = kw.get('atemp_min', None)
59        self.atemp_max = kw.get('atemp_max', None)
60        self.atemp_avg = kw.get('atemp_avg', None)
61        self.tracking_file = kw.get('tracking_file', None)  # Blob
62        self.tracking_filetype = ''  # unicode string
63        # attr to store ANT fit files. For now this file is used to
64        # generate a gpx-encoded tracking file we then use through
65        # the whole app
66        self.fit_file = kw.get('fit_file', None)  # Blob
67
68    @property
69    def workout_id(self):
70        return self.__name__
71
72    @property
73    def owner(self):
74        return self.__parent__
75
76    @property
77    def end(self):
78        if not self.duration:
79            return None
80        return self.start + self.duration
81
82    @property
83    def start_date(self):
84        return self.start.strftime('%d/%m/%Y')
85
86    @property
87    def start_time(self):
88        return self.start.strftime('%H:%M')
89
90    def start_in_timezone(self, timezone):
91        """
92        Return a string representation of the start date and time,
93        localized into the given timezone
94        """
95        _start = self.start.astimezone(pytz.timezone(timezone))
96        return _start.strftime('%d/%m/%Y %H:%M (%Z)')
97
98    def end_in_timezone(self, timezone):
99        """
100        Return a string representation of the end date and time,
101        localized into the given timezone
102        """
103        _end = self.end.astimezone(pytz.timezone(timezone))
104        return _end.strftime('%d/%m/%Y %H:%M (%Z)')
105
106    def split_duration(self):
107        return timedelta_to_hms(self.duration)
108
109    @property
110    def duration_hours(self):
111        return str(self.split_duration()[0]).zfill(2)
112
113    @property
114    def duration_minutes(self):
115        return str(self.split_duration()[1]).zfill(2)
116
117    @property
118    def duration_seconds(self):
119        return str(self.split_duration()[2]).zfill(2)
120
121    @property
122    def _duration(self):
123        return ':'.join(
124            [self.duration_hours, self.duration_minutes, self.duration_seconds]
125        )
126
127    @property
128    def rounded_distance(self):
129        """
130        Return rounded value for distance, '-' if the workout has no distance
131        data (weight lifting, martial arts, table tennis, etc)
132        """
133        if self.distance:
134            return round(self.distance, 2)
135        return '-'
136
137    @property
138    def has_hr(self):
139        """
140        True if this workout has heart rate data, False otherwise
141        """
142        data = [self.hr_min, self.hr_max, self.hr_avg]
143        return data.count(None) == 0
144
145    @property
146    def hr(self):
147        """
148        Return a dict with rounded values for hr min, max and average,
149        return None if there is no heart rate data for this workout
150        """
151        if self.has_hr:
152            return {'min': round(self.hr_min),
153                    'max': round(self.hr_max),
154                    'avg': round(self.hr_avg)}
155        return None
156
157    @property
158    def has_cad(self):
159        """
160        True if this workout has cadence data, False otherwise
161        """
162        data = [self.cad_min, self.cad_max, self.cad_avg]
163        return data.count(None) == 0
164
165    @property
166    def cad(self):
167        """
168        Return a dict with rounded values for cadence min, max and average
169        return None if there is no cadence data for this workout
170        """
171        if self.has_cad:
172            return {'min': round(self.cad_min),
173                    'max': round(self.cad_max),
174                    'avg': round(self.cad_avg)}
175        return None
176
177    @property
178    def has_atemp(self):
179        """
180        True if this workout has temperature data, False otherwise
181        """
182        data = [self.atemp_min, self.atemp_max, self.atemp_avg]
183        return data.count(None) == 0
184
185    @property
186    def atemp(self):
187        """
188        Return a dict with rounded values for temperature min, max and average
189        return None if there is no temperature data for this workout
190        """
191        if self.has_atemp:
192            return {'min': round(self.atemp_min),
193                    'max': round(self.atemp_max),
194                    'avg': round(self.atemp_avg)}
195        return None
196
197    @property
198    def tracking_file_path(self):
199        """
200        Get the path to the blob file attached as a tracking file.
201
202        First check if the file was not committed to the db yet (new workout
203        not saved yet) and use the path to the temporary file on the fs.
204        If none is found there, go for the real blob file in the blobs
205        directory
206        """
207        path = None
208        if self.tracking_file:
209            path = self.tracking_file._uncommitted()
210            if path is None:
211                path = self.tracking_file.committed()
212        return path
213
214    @property
215    def fit_file_path(self):
216        """
217        Get the path to the blob file attached as a fit file.
218
219        First check if the file was not committed to the db yet (new workout
220        not saved yet) and use the path to the temporary file on the fs.
221        If none is found there, go for the real blob file in the blobs
222        directory
223        """
224        path = None
225        if self.fit_file:
226            path = self.fit_file._uncommitted()
227            if path is None:
228                path = self.fit_file.committed()
229        return path
230
231    def load_from_file(self):
232        """
233        Check which kind of tracking file we have for this workout, then call
234        the proper method to load info from the tracking file
235        """
236        if self.tracking_filetype == 'gpx':
237            self.load_from_gpx()
238        elif self.tracking_filetype == 'fit':
239            self.load_from_fit()
240
241    def load_from_gpx(self):
242        """
243        Load some information from an attached gpx file. Return True if data
244        had been succesfully loaded, False otherwise
245        """
246        with self.tracking_file.open() as gpx_file:
247            gpx_contents = gpx_file.read()
248            gpx_contents = gpx_contents.decode('utf-8')
249            gpx = gpxpy.parse(gpx_contents)
250            if gpx.tracks:
251                track = gpx.tracks[0]
252                # Start time comes in UTC/GMT/ZULU
253                time_bounds = track.get_time_bounds()
254                self.start = time_bounds.start_time
255                # ensure this datetime start object is timezone-aware
256                self.start = self.start.replace(tzinfo=timezone.utc)
257                # get_duration returns seconds
258                self.duration = timedelta(seconds=track.get_duration())
259                # length_3d returns meters
260                self.distance = Decimal(track.length_3d()) / Decimal(1000.00)
261                ud = track.get_uphill_downhill()
262                self.uphill = Decimal(ud.uphill)
263                self.downhill = Decimal(ud.downhill)
264                # If the user did not provide us with a title, and the gpx has
265                # one, use that
266                if not self.title and track.name:
267                    self.title = track.name
268
269                # Hack to calculate some values from the GPX 1.1 extensions,
270                # using our own parser (gpxpy does not support those yet)
271                tracks = self.parse_gpx()
272                hr = []
273                cad = []
274                atemp = []
275                for t in tracks:
276                    hr += [
277                        d['hr'] for d in tracks[t] if d['hr'] is not None]
278                    cad += [
279                        d['cad'] for d in tracks[t] if d['cad'] is not None]
280                    atemp += [
281                        d['atemp'] for d in tracks[t]
282                        if d['atemp'] is not None]
283
284                if hr:
285                    self.hr_min = Decimal(min(hr))
286                    self.hr_avg = Decimal(sum(hr)) / Decimal(len(hr))
287                    self.hr_max = Decimal(max(hr))
288
289                if cad:
290                    self.cad_min = Decimal(min(cad))
291                    self.cad_avg = Decimal(sum(cad)) / Decimal(len(cad))
292                    self.cad_max = Decimal(max(cad))
293
294                if atemp:
295                    self.atemp_min = Decimal(min(atemp))
296                    self.atemp_avg = Decimal(sum(atemp)) / Decimal(len(atemp))
297                    self.atemp_max = Decimal(max(atemp))
298
299                return True
300
301        return False
302
303    def parse_gpx(self):
304        """
305        Parse the gpx using minidom.
306
307        This method is needed as a workaround to get HR/CAD/TEMP values from
308        gpx 1.1 extensions (gpxpy does not handle them correctly so far)
309        """
310        if not self.has_gpx:
311            # No gpx, nothing to return
312            return {}
313
314        # Get the path to the blob file, first check if the file was not
315        # committed to the db yet (new workout not saved yet) and use the
316        # path to the temporary file on the fs. If none is found there, go
317        # for the final blob file
318        gpx_path = self.tracking_file._p_blob_uncommitted
319        if gpx_path is None:
320            gpx_path = self.tracking_file._p_blob_committed
321
322        # Create a parser, load the gpx and parse the tracks
323        parser = GPXMinidomParser(gpx_path)
324        parser.load_gpx()
325        parser.parse_tracks()
326        return parser.tracks
327
328    def load_from_fit(self):
329        """
330        Try to load data from an ANT-compatible .fit file (if any has been
331        added to this workout).
332
333        "Load data" means:
334
335        1. Copy over the uploaded fit file to self.fit_file, so we can keep
336           that copy around for future use
337
338        2. generate a gpx object from the fit file
339
340        3. save the gpx object as the tracking_file, which then will be used
341           by the current code to display and gather data to be displayed/shown
342           to the user.
343
344        4. Grab some basic info from the fit file and store it in the Workout
345        """
346
347        # we can call load_from_fit afterwards for updates. In such case, check
348        # if the tracking file is a fit file uploaded to override the previous
349        # one. If not, just reuse the existing fit file
350        if self.tracking_filetype == 'fit':
351            # backup the fit file
352            self.fit_file = copy_blob(self.tracking_file)
353
354        # create an instance of our Fit class
355        fit = Fit(self.fit_file_path)
356        fit.load()
357
358        # fit -> gpx and store that as the main tracking file
359        self.tracking_file = create_blob(fit.gpx, 'gpx')
360        self.tracking_filetype = 'gpx'
361
362        # grab the needed data from the fit file, update the workout
363        self.sport = fit.data['sport']
364        self.start = fit.data['start']
365        # ensure this datetime start object is timezone-aware
366        self.start = self.start.replace(tzinfo=timezone.utc)
367        # duration comes in seconds, store a timedelta
368        self.duration = timedelta(seconds=fit.data['duration'])
369        if fit.data['distance']:
370            # distance comes in meters
371            self.distance = Decimal(fit.data['distance']) / Decimal(1000.00)
372        if fit.data['uphill']:
373            self.uphill = Decimal(fit.data['uphill'])
374        if fit.data['downhill']:
375            self.downhill = Decimal(fit.data['downhill'])
376        # If the user did not provide us with a title, build one from the
377        # info in the fit file
378        if not self.title:
379            self.title = fit.name
380
381        if fit.data['max_speed']:
382            self.speed['max'] = mps_to_kmph(fit.data['max_speed'])
383
384        if fit.data['avg_speed']:
385            self.speed['avg'] = mps_to_kmph(fit.data['avg_speed'])
386
387        if fit.data['avg_hr']:
388            self.hr_avg = Decimal(fit.data['avg_hr'])
389            self.hr_min = Decimal(fit.data['min_hr'])
390            self.hr_max = Decimal(fit.data['max_hr'])
391
392        if fit.data['avg_cad']:
393            self.cad_avg = Decimal(fit.data['avg_cad'])
394            self.cad_min = Decimal(fit.data['min_cad'])
395            self.cad_max = Decimal(fit.data['max_cad'])
396
397        if fit.data['avg_atemp']:
398            self.atemp_avg = Decimal(fit.data['avg_atemp'])
399            self.atemp_min = Decimal(fit.data['min_atemp'])
400            self.atemp_max = Decimal(fit.data['max_atemp'])
401
402        return True
403
404    @property
405    def has_tracking_file(self):
406        return self.tracking_file is not None
407
408    @property
409    def has_gpx(self):
410        return self.has_tracking_file and self.tracking_filetype == 'gpx'
411
412    @property
413    def has_fit(self):
414        return self.fit_file is not None
415
416    @property
417    def map_screenshot(self):
418        """
419        Return the static path to the screenshot image of the map for
420        this workout (works only for workouts with gps tracking)
421        """
422        if not self.has_gpx:
423            return None
424
425        current_path = os.path.abspath(os.path.dirname(__file__))
426        screenshot_path = os.path.join(
427            current_path, '../static/maps',
428            str(self.owner.uid), str(self.workout_id)) + '.png'
429
430        if not os.path.exists(screenshot_path):
431            # screenshot does not exist, generate it
432            save_map_screenshot(self)
433
434        # the value returned is relative to the static files served
435        # by the app, so we can use request.static_url() with it
436        static_path = os.path.join('static/maps', str(self.owner.uid),
437                                   str(self.workout_id))
438        return 'ow:' + static_path + '.png'
Note: See TracBrowser for help on using the repository browser.