Skip to content
Snippets Groups Projects

Sanitize cal_tools.tools.get_creation_date

Merged Cyril Danilevski requested to merge clean/get_dir_creation_date into master
+ 16
15
@@ -2,8 +2,9 @@ from collections import OrderedDict
import datetime
from glob import glob
import json
from os import environ, listdir, path, stat
from os.path import isfile, splitext
from os import environ, listdir, path
from os.path import isfile
from pathlib import Path
from queue import Queue
import re
from time import sleep
@@ -230,7 +231,7 @@ def get_run_info(proposal, run):
def get_dir_creation_date(directory: str, run: int,
tsdir: Optional[bool] = False,
verbosity: Optional[int] = 0):
verbosity: Optional[int] = 0) -> Optional[datetime.datetime]:
"""
Return run starting time from the MDC.
If not succeeded, return modification time of oldest file.h5
@@ -241,35 +242,35 @@ def get_dir_creation_date(directory: str, run: int,
:param tsdir: to get modification time of [directory]/[run]04.
:param verbosity: Level of verbosity (0 - silent)
:return: (datetime) modification time
"""
directory = Path(directory)
proposal = int(directory.parent.name[1:])
try:
path_list = list(filter(None, directory.strip('/').split('/')))
proposal = int(path_list[-2][1:])
run_info = get_run_info(proposal, run)
return dateutil.parser.parse(run_info['begin_at'])
except Exception as e:
if verbosity > 0:
print(e)
directory = directory / 'r{:04d}'.format(run)
ntries = 100
while ntries > 0:
try:
if tsdir:
creation_time = stat("{}/r{:04d}".format(directory, run)).st_mtime
creation_time = directory.stat().st_mtime
else:
rfiles = glob("{}/r{:04d}/*.h5".format(directory, run))
rfiles.sort(key=path.getmtime)
creation_time = stat(rfiles[0]).st_mtime
creation_time = datetime.datetime.fromtimestamp(creation_time)
return creation_time
except: # catch stale file handle errors etc and try again
creation_time = min(f.stat().st_mtime
for f in directory.glob('*.h5'))
return datetime.datetime.fromtimestamp(creation_time)
except IOError: # catch stale file handle errors etc and try again
ntries -= 1
def save_const_to_h5(device, constant, condition, data,
file_loc, creation_time, out_folder):
file_loc, creation_time, out_folder):
"""
Save constant in h5 file with its metadata
(e.g. db_module, condition, creation_time)
@@ -280,7 +281,7 @@ def save_const_to_h5(device, constant, condition, data,
:type constant: iCalibrationDB.know_constants object
:param condition: Calibration condition
:type condition: iCalibrationDB.know_detector_conditions object
:param data: Constant data to save
:param data: Constant data to save
:type data: ndarray
:param file_loc: Location of raw data "proposal:{} runs:{} {} {}"
:type file_loc: str
Loading