1import importlib
2import os
3import platform
4import warnings
5from base64 import b64decode, b64encode
6from collections.abc import Iterable
7from copy import deepcopy
8from datetime import date
9from itertools import chain
10from uuid import uuid4
11
12import dill
13import numpy as np
14import pandas as pd
15import pymongo
16
17from omegaml.backends.tracking.base import TrackingProvider
18from omegaml.documents import Metadata
19from omegaml.util import _raise, batched, ensure_index, ensurelist, signature, tryOr
20
21
[docs]
22class NoTrackTracker(TrackingProvider):
23 """A default tracker that does not record anything"""
24
25 def start(self, run=None):
26 pass
27
28 def stop(self):
29 pass
30
31 def log_artifact(self, obj, name, step=None, **extra):
32 pass
33
34 def log_metric(self, key, value, step=None, **extra):
35 pass
36
37 def log_param(self, key, value, step=None, **extra):
38 pass
39
40 def log_event(self, event, key, value, step=None, **extra):
41 pass
42
43 def log_events(self, event, key, values, step=None, dt=None, **extra):
44 pass
45
46 def log_extra(self, **kwargs):
47 pass
48
49 def log_data(self, **kwargs):
50 pass
51
52 def data(self, experiment=None, run=None, event=None, step=None, key=None, raw=False, **query):
53 pass
54
55
[docs]
56class OmegaSimpleTracker(TrackingProvider):
57 """A tracking provider that logs to an omegaml dataset
58
59 Usage::
60
61 with om.runtime.experiment(provider='default') as exp:
62 ...
63 exp.log_metric('accuracy', .78)
64
65 .. versionchanged:: 0.17
66 any extra
67
68 """
69
70 _provider = 'simple'
71 _experiment = None
72 _startdt = None
73 _stopdt = None
74 _autotrack = False
75
76 _ensure_active = lambda self, r: (
77 r if r is not None else _raise(ValueError('no active run, call .start() or .use() '))
78 )
79
80 def __init__(self, *args, **kwargs):
81 super().__init__(*args, **kwargs)
82 self.log_buffer = []
83 self.max_buffer = 10
84 self._initialize_dataset()
85
[docs]
86 def active_run(self, run=None):
87 """set the lastest run as the active run
88
89 Args:
90 run (int|str): optional or unique task id, if None the
91 latest active run will be set, or a new run is created if
92 no active run exists.
93
94 Returns:
95 current run (int)
96 """
97 if run is None:
98 latest = self._latest_run
99 latest_is_active = latest is not None and self.status(run=latest) == 'STARTED'
100 self._run = latest if latest_is_active else self.start(run=None)
101 else:
102 self._run = run
103 self._experiment = self._experiment or uuid4().hex
104 return self._run
105
[docs]
106 def use(self, run=None):
107 """reuse the latest run instead of starting a new one
108
109 semantic sugar for self.active_run()
110
111 Returns:
112 self
113 """
114 self.active_run(run=run)
115 return self
116
117 @property
118 def autotrack(self):
119 return self._autotrack
120
121 @autotrack.setter
122 def autotrack(self, value):
123 self._autotrack = value
124
125 @property
126 def _latest_run(self):
127 cursor = self.data(event='start', run='*', lazy=True)
128 data = list(cursor.sort('data.run', -1).limit(1)) if cursor else None
129 run = data[-1].get('data', {}).get('run') if data is not None and len(data) > 0 else None
130 return run
131
[docs]
132 def status(self, run=None):
133 """status of a run
134
135 Args:
136 run (int): the run number, defaults to the currently active run
137
138 Returns:
139 status in 'STARTED', 'STOPPED'
140 """
141 self._run = run or self._run or self._latest_run
142 data = self.data(event=['start', 'stop'], run=self._run, raw=True)
143 no_runs = data is None or len(data) == 0
144 has_stop = not no_runs and sum(1 for row in data if row.get('event') == 'stop')
145 return 'PENDING' if no_runs else 'STOPPED' if has_stop else 'STARTED'
146
[docs]
147 def start(self, run=None, immediate=True):
148 """start a new run
149
150 This starts a new run and logs the start event
151 """
152 self._run = run or (self._latest_run or 0) + 1
153 self._startdt = datetime.utcnow()
154 data = self._common_log_data('start', key=None, value=None, step=None, dt=self._startdt)
155 self._write_log(data, immediate=immediate)
156 return self._run
157
[docs]
158 def stop(self, flush=True):
159 """stop the current run
160
161 This stops the current run and records the stop event
162 """
163 self._stopdt = datetime.utcnow()
164 data = self._common_log_data('stop', key=None, value=None, step=None, dt=self._stopdt)
165 self._write_log(data)
166 self.flush() if flush else None
167
168 def flush(self):
169 # passing list of list, as_many=True => collection.insert_many() for speed
170 if self.log_buffer:
171 self._store.put(self.log_buffer, self._data_name, noversion=True, as_many=True)
172 self.log_buffer.clear()
173
[docs]
174 def clear(self, force=False):
175 """clear all data
176
177 All data is removed from the experiment's dataset. This is not recoverable.
178
179 Args:
180 force (bool): if True, clears all data, otherwise raises an error
181
182 Caution:
183 * this will clear all data and is not recoverable
184
185 Raises:
186 AssertionError: if force is not True
187
188 .. versionadded:: 0.16.2
189
190 """
191 assert force, (
192 "clear() requires force=True to prevent accidental data loss. This will clear all experiment data and is not recoverable."
193 )
194 self._store.drop(self._data_name, force=True)
195 self._initialize_dataset(force=True)
196
197 def _common_log_data(self, event, key, value, step=None, dt=None, run=None, **extra):
198 if isinstance(value, dict):
199 # shortcut to resolve PassthroughDataset actual values
200 # -- enables storing the actual values of a dataset passed as a PassthroughDataset
201 # TODO: should this be the responsibility of SimpleTracker?
202 if isinstance(value.get('args'), (list, tuple)):
203 value['args'] = [getattr(arg, '_passthrough_data', arg) for arg in value['args']]
204 if isinstance(value.get('kwargs'), dict):
205 value['kwargs'] = {k: getattr(v, '_passthrough_data', v) for k, v in value['kwargs'].items()}
206 data = {
207 'experiment': self._experiment,
208 'run': run or self._ensure_active(self._run),
209 'step': step,
210 'event': event,
211 'key': key or event,
212 'value': value,
213 'dt': dt or datetime.utcnow(),
214 'node': os.environ.get('HOSTNAME', platform.node()),
215 'userid': self.userid,
216 }
217 # add **extra, check for duplicate keys to avoid overwriting
218 dupl_keys = set(data.keys()) & set(extra.keys())
219 if dupl_keys:
220 raise ValueError(f'duplicate extra keys : {dupl_keys}')
221 data.update(extra)
222 data.update(self._extra_log) if self._extra_log else None
223 return data
224
225 def _write_log(self, data, immediate=False):
226 self.log_buffer.append(data)
227 if immediate or len(self.log_buffer) > self.max_buffer:
228 self.flush()
229
[docs]
230 def log_artifact(self, obj, name, step=None, dt=None, event=None, key=None, **extra):
231 """log any object to the current run
232
233 Usage::
234
235 # log an artifact
236 exp.log_artifact(mydict, 'somedata')
237
238 # retrieve back
239 mydict_ = exp.restore_artifact('somedata')
240
241 Args:
242 obj (obj): any object to log
243 name (str): the name of artifact
244 step (int): the step, if any
245 **extra: any extra data to log
246
247 Notes:
248 * bool, str, int, float, list, dict are stored as ``format=type``
249 * Metadata is stored as ``format=metadata``
250 * objects supported by ``om.models`` are stored as ``format=model``
251 * objects supported by ``om.datasets`` are stored as ``format=dataset``
252 * all other objects are pickled and stored as ``format=pickle``
253 """
254 event = event or 'artifact'
255 key = key or name
256 if isinstance(obj, (bool, str, int, float, list, dict)):
257 format = 'type'
258 rawdata = obj
259 elif isinstance(obj, Metadata):
260 format = 'metadata'
261 rawdata = obj.to_json()
262 elif self._model_store.get_backend_byobj(obj) is not None:
263 objname = uuid4().hex
264 meta = self._model_store.put(obj, f'.experiments/.artefacts/{objname}')
265 format = 'model'
266 rawdata = meta.name
267 elif self._store.get_backend_byobj(obj) is not None:
268 objname = uuid4().hex
269 meta = self._store.put(obj, f'.experiments/.artefacts/{objname}')
270 format = 'dataset'
271 rawdata = meta.name
272 else:
273 try:
274 rawdata = b64encode(dill.dumps(obj)).decode('utf8')
275 format = 'pickle'
276 except TypeError:
277 rawdata = repr(obj)
278 format = 'repr'
279 value = {'name': name, 'data': rawdata, 'format': format}
280 data = self._common_log_data(event, key, value, step=step, dt=dt, name=name, **extra)
281 self._write_log(data)
282
[docs]
283 def log_event(self, event, key, value, step=None, dt=None, **extra):
284 """log a system event
285
286 Args:
287 event (str): the event name
288 key (str): the key to use, defaults to 'event'
289 value (str|float|int|bool|dict): the parameter value
290 step (int): the step
291 dt (datetime): the datetime
292 **extra: any other values to store with event
293
294 Returns:
295 None
296
297 Notes:
298 * events generated by tracking include 'start', 'stop', 'param', 'metric',
299 'data', 'system'
300 * events generated by the runtime include 'task_call', 'task_failure', 'task_retry',
301 'task_success'
302 * events generated by drift monitoring include 'snapshot', 'drift', 'monitor', 'drift'
303 * events generated by the model backends include 'fit', 'predict', 'complete', 'embed'
304 * to ensure your own events are distinct from the above or future system events, prefix
305 your events with 'user_' or an application code (e.g. 'myapp_')
306
307 .. versionchanged:: 0.17.0
308 added documentation on events throughout the system
309 """
310 data = self._common_log_data(event, key, value, step=step, dt=dt, **extra)
311 self._write_log(data)
312
[docs]
313 def log_events(self, event, key, values, step=None, dt=None, **extra):
314 """log a series of events
315
316 This is a convenience method to log multiple values for the same event.
317 All values will be logged with the same commong log data, i.e. the same
318 datetime, step, and any extra values.
319
320 Args:
321 event (str): the event name
322 key (str): the key for the event
323 values (list): a list of values to log
324 step (int): the step, if any
325 dt (datetime): the datetime, defaults to now
326 **extra: any other values to store with event
327
328 .. versionadded:: 0.18.0
329
330 """
331 data = self._common_log_data(event, key, None, step=step, dt=dt, **extra)
332 for value in values:
333 data['value'] = value
334 self._write_log(dict(data))
335
[docs]
336 def log_param(self, key, value, step=None, dt=None, **extra):
337 """log an experiment parameter
338
339 Args:
340 key (str): the parameter name
341 value (str|float|int|bool|dict): the parameter value
342 step (int): the step
343 **extra: any other values to store with event
344
345 Notes:
346 * logged as ``event=param``
347 """
348 data = self._common_log_data('param', key, value, step=step, dt=dt, **extra)
349 self._write_log(data)
350
[docs]
351 def log_metric(self, key, value, step=None, dt=None, **extra):
352 """log a metric value
353
354 Args:
355 key (str): the metric name
356 value (str|float|int|bool|dict): the metric value
357 step (int): the step
358 **extra: any other values to store with event
359
360 Notes:
361 * logged as ``event=metric``
362 """
363 data = self._common_log_data('metric', key, value, step=step, dt=dt, **extra)
364 self._write_log(data)
365
[docs]
366 def log_data(self, key, value, step=None, dt=None, event=None, **extra):
367 """log x/y data for model predictions
368
369 This is semantic sugar for log_artifact() using the 'data' event.
370
371 Args:
372 key (str): the name of the artifact
373 value (any): the x/y data
374 step (int): the step
375 dt (datetime): the datetime
376 event (str): the event, defaults to 'data'
377 **extra: any other values to store with event
378
379 Returns:
380 None
381 """
382 event = event or 'data'
383 self.log_artifact(value, key, step=step, dt=dt, key=key, event=event, **extra)
384
[docs]
385 def log_system(self, key=None, value=None, step=None, dt=None, **extra):
386 """log system data
387
388 Args:
389 key (str): the key to use, defaults to 'system'
390 value (str|float|int|bool|dict): the parameter value
391 step (int): the step
392 **extra: any other values to store with event
393
394 Notes:
395 * logged as ``event=system``
396 * logs platform, python version and list of installed packages
397 """
398 key = key or 'system'
399 value = value or {
400 'platform': platform.uname()._asdict(),
401 'python': '-'.join((platform.python_implementation(), platform.python_version())),
402 'packages': ['=='.join((d.metadata['Name'], d.version)) for d in importlib.metadata.distributions()],
403 }
404 data = self._common_log_data('system', key, value, step=step, dt=dt, **extra)
405 self._write_log(data)
406
425
[docs]
426 def data(
427 self,
428 experiment=None,
429 run=None,
430 event=None,
431 step=None,
432 key=None,
433 raw=False,
434 lazy=False,
435 since=None,
436 end=None,
437 batchsize=None,
438 slice=None,
439 **extra,
440 ):
441 """build a dataframe of all stored data
442
443 Args:
444 experiment (str|list): the name of the experiment, defaults to its current value
445 run (int|list|str|slice): the run(s) to get data back, defaults to current run, use 'all' for all,
446 1-indexed since first run, or -1 indexed from latest run, can combine both. If run < 0
447 would go before the first run, run 1 will be returned. A slice(start, stop) can be used
448 to specify a range of runs.
449 event (str|list): the event(s) to include
450 step (int|list): the step(s) to include
451 key (str|list): the key(s) to include
452 raw (bool): if True returns the raw data instead of a DataFrame
453 lazy (bool): if True returns the Cursor instead of data, ignores raw
454 since (datetime|timedelta|str): only return data since this date. If both since and run are specified,
455 only matches since the given date are returned. If since is a string it must be parseable
456 by pd.to_datime, or be given in the format '<n><unit:[smhdwMqy]>' for relative times, or a timedelta object. See
457 dtrelative() for details on relative times.
458 end (datetime): only return data until this date
459 batchsize (int): if specified, returns a generator yielding data in batches of batchsize,
460 note that raw is respected, i.e. raw=False yields a DataFrame for every batch, raw=True
461 yields a list of dicts
462 slice (tuple): if specified, returns a slice of the data, e.g. slice=(10, 25) returns rows 10-25,
463 the slice is applied after all other filters
464
465 Returns:
466 For lazy == False:
467 * data (DataFrame) if raw == False
468 * data (list of dicts) if raw == True
469 * None if no data exists
470
471 For lazy == True, no batchsize, regardless of raw:
472 * data (Cursor) for any value of raw
473
474 For lazy == True, with batchsize:
475 * data(generator of list[dict]) if raw = True
476 * data(generator of DataFrame) if raw = False
477
478 .. versionchanged:: 0.16.2
479 run supports negative indexing
480
481 .. versionchanged:: 0.17
482 added batchsize
483
484 .. versionchanged:: 0.17
485 enabled the use of run='*' to retrieve all runs, equivalent of run='all'
486
487 .. versionchanged:: 0.17
488 enabled data(run=, start=, end=, since=), accepting range queries on run, dt and event#
489 """
490 from functools import cache
491
492 experiment = experiment or self._experiment
493 # -- flush all buffers before querying
494 self.flush()
495 # -- build filter
496 if since is None:
497 run = run if run is not None else self._run
498 run = ensurelist(run) if not isinstance(run, str) and isinstance(run, Iterable) else run
499 # actual run
500 # -- run is 1-indexed, so we need to adjust for -1 indexing
501 # e.g. -1 means the latest run, -2 the run before that
502 # e.g. latest_run = 5, run=-1 means 5, run=-2 means 4 etc.
503 # -- run can be a list, in which case we adjust run < 0 for each element
504 # -- run can never be less than 1 (1-indexed), even if run << 0
505 last_run = cache(
506 lambda: int(self._latest_run or 0)
507 ) # PERF/consistency: memoize the last run per each .data() call
508 relative_run = lambda r: max(1, 1 + last_run() + r)
509 if isinstance(run, list) and any(r < 0 for r in run):
510 run = [(r if r >= 0 else relative_run(r)) for r in run]
511 elif isinstance(run, int) and run < 0:
512 run = relative_run(run)
513 filter = self._build_data_filter(experiment, run, event, step, key, since, end, extra)
514
515 def read_data(cursor):
516 # remove mongodb weakly referenced objects
517 data = pd.DataFrame.from_records(deepcopy(data) for data in cursor)
518 if 'dt' in data.columns:
519 data['dt'] = pd.to_datetime(data['dt'], errors='coerce')
520 data.sort_values('dt', inplace=True)
521 return data
522
523 def read_data_batched(cursor, batchsize, slice):
524 from builtins import slice as t_slice
525
526 if cursor is None:
527 yield None
528 return
529 if slice:
530 slice = (slice.start or 0, slice.stop or 0) if isinstance(slice, t_slice) else slice
531 cursor.skip(slice[0])
532 cursor.limit(slice[1] - slice[0])
533 batchsize = batchsize or (slice[1] - slice[0])
534 for rows in batched(cursor, batchsize):
535 data = (r.get('data') for r in rows)
536 yield read_data(data) if not raw else list(data)
537
538 if batchsize or slice:
539 data = self._store.get(self._data_name, filter=filter, lazy=True, trusted=signature(filter))
540 data = read_data_batched(data, batchsize, slice)
541 if slice and not batchsize:
542 # try to resolve just one iteration
543 data, _ = tryOr(lambda: (next(data), data.close()), (None, None))
544 else:
545 data = self._store.get(self._data_name, filter=filter, lazy=lazy, trusted=signature(filter))
546 data = read_data(data) if data is not None and not lazy and not raw else data
547 return data
548
549 def _build_data_filter(self, experiment, run, event, step, key, since, end, extra):
550 # build a filter for the data query, suitable for OmegaStore.get()
551 filter = {}
552 valid = lambda s: s is not None and str(s).lower() not in ('all', '*')
553 # SEC: ensure all values are basic types, to prevent operator injection
554 valid_types = (str, int, float, list, tuple, date, datetime)
555 op = lambda s: (
556 {'$in': ensurelist(s)} if isinstance(s, (list, tuple, np.ndarray)) else ensure_type(s, valid_types)
557 )
558 ensure_type = lambda v, t: v if isinstance(v, t) else str(v)
559 if valid(experiment):
560 filter['data.experiment'] = op(experiment)
561 if valid(run):
562 if isinstance(run, slice):
563 filter['data.run'] = {'$gte': run.start, '$lte': run.stop}
564 else:
565 filter['data.run'] = op(run)
566 if valid(event):
567 filter['data.event'] = op(event)
568 if valid(step):
569 filter['data.step'] = op(step)
570 if valid(key):
571 filter['data.key'] = op(key)
572 if valid(since):
573 dtnow = getattr(self, '_since_dtnow', datetime.utcnow())
574 if isinstance(since, str):
575 since = tryOr(lambda: pd.to_datetime(since), lambda: dtrelative('-' + since, now=dtnow))
576 elif isinstance(since, timedelta):
577 since = dtnow - since
578 elif isinstance(since, datetime):
579 since = since
580 else:
581 raise ValueError(
582 f'invalid since value: {since}, must be datetime, timedelta or string in format "<n><unit:[smhdwMqy]>"'
583 )
584 filter['data.dt'] = {'$gte': str(since.isoformat())}
585 if valid(end):
586 dtnow = getattr(self, '_since_dtnow', datetime.utcnow())
587 if isinstance(end, str):
588 end = tryOr(lambda: pd.to_datetime(end), lambda: dtrelative('+' + end, now=dtnow))
589 elif isinstance(end, timedelta):
590 end = dtnow + end
591 elif isinstance(end, datetime):
592 end = end
593 else:
594 raise ValueError(
595 f'invalid end value: {end}, must be datetime, timedelta or string in format "<n><unit:[smhdwMqy]>"'
596 )
597 filter['data.dt'] = filter.setdefault('data.dt', {})
598 filter['data.dt']['$lte'] = str(end.isoformat())
599 for k, v in extra.items():
600 if valid(v):
601 fk = f'data.{k}'
602 filter[fk] = op(v)
603 return filter
604
605 @property
606 def dataset(self):
607 return self._data_name
608
609 @property
610 def stats(self):
611 from omegaml.backends.tracking.statistics import ExperimentStatistics
612
613 return ExperimentStatistics(self)
614
615 def summary(self, **kwargs):
616 return self.stats.summary(**kwargs)
617
618 def _initialize_dataset(self, force=False):
619 # create indexes when the dataset is first created
620 if not force and self._store.exists(self._data_name):
621 return
622 coll = self._store.collection(self._data_name)
623 meta = self._store.metadata(self._data_name)
624 idxs = [
625 # fast retrieval by run, event, key
626 {
627 'data.run': pymongo.ASCENDING,
628 'data.event': pymongo.ASCENDING,
629 'data.key': pymongo.ASCENDING,
630 'data.experiment': pymongo.ASCENDING,
631 },
632 # fast retrieval by dt, event, key
633 {
634 'data.dt': pymongo.ASCENDING,
635 'data.event': pymongo.ASCENDING,
636 'data.key': pymongo.ASCENDING,
637 'data.experiment': pymongo.ASCENDING,
638 },
639 {'data.dt': pymongo.ASCENDING, 'data.event': pymongo.ASCENDING, 'data.experiment': pymongo.ASCENDING},
640 # fast retrieval by event, key, userid across all runs
641 {
642 'data.event': pymongo.ASCENDING,
643 'data.key': pymongo.ASCENDING,
644 'data.userid': pymongo.ASCENDING,
645 'data.experiment': pymongo.ASCENDING,
646 },
647 # optimize self._latest_run - this is O(1) retrieval of max(run)
648 {
649 'data.run': pymongo.DESCENDING,
650 'data.event': pymongo.ASCENDING,
651 'data.experiment': pymongo.ASCENDING,
652 'create_index_kwargs': dict(partialFilterExpression={"data.event": "start"}),
653 },
654 ]
655 # add additional indexes from metadata
656 idxs.extend(meta.attributes.get('indexes', [])) if meta else None
657 # TODO use DatasetIndexesMixin
658 # self._store.ensure_indexes(self._data_name, idx_specs=idxs, replace=force)
659 for specs in idxs:
660 kwargs = specs.pop('create_index_kwargs', {})
661 ensure_index(coll, specs, **kwargs)
662 # self._store.put(coll, self._data_name)
663
[docs]
664 def restore_artifact(self, *args, **kwargs):
665 """restore a specific logged artifact
666
667 .. versionchanged:: 0.17
668 deprecated, use exp.restore_artifacts() instead
669 """
670 warnings.warn('deprecated, use exp.restore_artifacts() instead', DeprecationWarning)
671 restored = self.restore_artifacts(*args, **kwargs)
672 return restored[-1] if restored else None
673
[docs]
674 def restore_artifacts(
675 self, key=None, experiment=None, run=None, since=None, step=None, value=None, event=None, name=None
676 ):
677 """restore logged artifacts
678
679 Args:
680 key (str): the name of the artifact as provided in log_artifact
681 run (int): the run for which to query, defaults to current run
682 since (datetime): only return data since this date
683 step (int): the step for which to query, defaults to all steps in run
684 value (dict|list): dict or list of dict, this value is used instead of
685 querying data, use to retrieve an artifact from contents of ``.data()``
686
687 Returns:
688 list of restored objects
689
690 Notes:
691 * this will restore the artifact according to its type assigned
692 by ``.log_artifact()``. If the type cannot be determined, the
693 actual data is returned
694
695 Updates:
696 * since 0.17: return list of objects instead of last object
697 """
698 event = event or 'artifact'
699 name = name or '*'
700 if value is None:
701 all_data = self.data(
702 experiment=experiment, run=run, since=since, event=event, step=step, key=key, raw=True, name=name
703 )
704 else:
705 all_data = [{'value': value}] if isinstance(value, dict) else value
706 restored = []
707 all_data = all_data or []
708 for item in all_data:
709 data = item.get('value')
710 if data['format'] == 'type':
711 obj = data['data']
712 elif data['format'] == 'metadata':
713 meta = self._store._Metadata
714 obj = meta.from_json(data['data'])
715 elif data['format'] == 'dataset':
716 obj = self._store.get(data['data'])
717 elif data['format'] == 'model':
718 obj = self._model_store.get(data['data'])
719 elif data['format'] == 'pickle':
720 obj = dill.loads(b64decode((data['data']).encode('utf8')))
721 else:
722 obj = data.get('data', data)
723 restored.append(obj)
724 return restored
725
[docs]
726 def restore_data(self, key, run=None, event=None, since=None, concat=True, **extra):
727 """restore x/y data for model predictions
728
729 This is semantic sugar for restore_artifacts() using the event='data' event.
730
731 Args:
732 key (str): the name of the artifact
733 run (int): the run for which to query, defaults to current run
734 event (str): the event, defaults to 'data'
735 since (datetime): only return data since this date
736 concat (bool): if True, concatenates the data into a single object,
737 in this case all data must be of the same type. Defaults to True.
738 **extra: any other values to store with event
739
740 Returns:
741 list of restored objects
742 """
743 event = event or 'data'
744
745 def _concat(values):
746 if values is None:
747 return None
748 if len(values) and isinstance(values[0], (pd.DataFrame, pd.Series)):
749 ensure_df_or_series = lambda v: pd.Series(v) if isinstance(v, (np.ndarray, list)) else v
750 return pd.concat((ensure_df_or_series(v) for v in values), axis=0)
751 elif len(values) and isinstance(values[0], np.ndarray):
752 return np.concatenate(values, axis=0)
753 # chain seems to be the fastests approach
754 # -- https://stackoverflow.com/a/56407963/890242
755 return list(chain(*values))
756
757 restored = self.restore_artifacts(run=run, key=key, event=event, since=since, **extra)
758 return restored if not concat else _concat(restored)
759
760
761from datetime import datetime, timedelta
762
763
764def dtrelative(delta, now=None, as_delta=False):
765 """return a datetime relative to now
766
767 Args:
768 delta (str|timedelta): the relative delta, if a string, specify as '[+-]<n><unit:[smhdwMqy]>',
769 e.g. '1d' for one day, '-1d' for one day ago, '+1d' for one day from now. Special cases:
770 '-0y' means the beginning of the current year, '+0y' means the end of the current year.
771 smhdwMqy = seconds, minutes, hours, days, weeks, months, quarters, years
772 now (datetime): the reference datetime, defaults to datetime.utcnow()
773 as_delta (bool): if True, returns a timedelta object, otherwise a datetime object
774
775 Returns:
776 datetime|timedelta: the relative datetime or timedelta
777 """
778 # Parse the numeric part and the unit from the specifier
779 UNIT_MAP = {
780 's': 1, # 1 second
781 'm': 60, # 1 minute
782 'h': 60 * 60, # 1 hour
783 'd': 24 * 60 * 60, # 1 day
784 'w': 7 * 24 * 60 * 60, # 1 week
785 'n': 30 * 24 * 60 * 60, # 1 month
786 'q': 90 * 24 * 60 * 60, # 1 quarter
787 'y': 365 * 24 * 60 * 60,
788 } # 1 year
789 now = now or datetime.utcnow()
790 error_msg = f"Invalid delta {delta}. Use a string of format '<n><unit:[hdwmqy]>' or timedelta object."
791 if isinstance(delta, str):
792 try:
793 past = delta.startswith('-')
794 delta = (
795 delta
796 .replace('-', '')
797 .replace('+', '') # Remove the sign
798 .replace(' ', '') # Remove spaces
799 .replace('M', 'n') # m is ambiguous, so we use n for month
800 .lower()
801 )
802 num = int(delta[:-1]) # The numeric part
803 units = UNIT_MAP.get(delta[-1]) # The last character
804 if delta[-1] == 'y' and num == 0:
805 # special case -0y means beginning of the year, +0y means end of the year
806 dtdelta = (datetime(now.year, 1, 1) - now) if past else (datetime(now.year, 12, 31) - now)
807 else:
808 dtdelta = timedelta(seconds=num * units * (-1 if past else 1))
809 except:
810 raise ValueError(error_msg)
811 elif isinstance(delta, timedelta):
812 dtdelta = delta
813 else:
814 raise ValueError(error_msg)
815 return now + dtdelta if not as_delta else dtdelta