From f37eb63996bff9756cca947ce7daff4172bf45a1 Mon Sep 17 00:00:00 2001 From: ahmedk <karim.ahmed@xfel.eu> Date: Wed, 12 Jun 2024 15:32:33 +0200 Subject: [PATCH] refactor(calcat_bits): small refactors(delete unused functions and apply pep8) --- utils/calcat_bits.py | 54 ++++++++++++++++---------------------------- 1 file changed, 19 insertions(+), 35 deletions(-) diff --git a/utils/calcat_bits.py b/utils/calcat_bits.py index 7e1ad59..920ecf4 100644 --- a/utils/calcat_bits.py +++ b/utils/calcat_bits.py @@ -1,19 +1,24 @@ import functools -import json import pathlib import sys from typing import List -from .config import calcat_secrets import numpy as np import streamlit as st -from calibration_client.modules import (Calibration, CalibrationConstant, - CalibrationConstantVersion, Condition, - Detector, DetectorType, Parameter, - PhysicalDetectorUnit) +from calibration_client.modules import ( + Calibration, + CalibrationClient, + CalibrationConstant, + CalibrationConstantVersion, + Condition, + Detector, DetectorType, Parameter, + PhysicalDetectorUnit, +) from utils import logger, utils +from .config import calcat_secrets + logging = logger.setup_logger(st) # behave differently in notebook vs streamlit app @@ -34,8 +39,9 @@ class CalcatWrapper: def __init__( self, ): + logging.info("Initializing CalCat wrapper...") - from calibration_client import CalibrationClient + # TODO configure using test or production calcat database. self.base_url = calcat_secrets.base_url self.cal_client = CalibrationClient( client_id=calcat_secrets.client_id, @@ -49,8 +55,6 @@ class CalcatWrapper: session_token=None, ) self.caldb_store = pathlib.Path(calcat_secrets.caldb_store_path) - # caldb_root = pathlib.Path("/gpfs/exfel/d/cal/caldb_store/xfel/cal") - # calcat_caldb_root = pathlib.Path("/gpfs/exfel/d/cal/caldb_store") def get_all_over_pages(self, fun, *args, **kwargs): first_page = fun(self.cal_client, *args, **kwargs) @@ -129,7 +133,7 @@ class CalcatWrapper: Returns: list: list of dictionaries containing detector type ID, - name, and more info. + name, and more info. e.g.[{ 'id': 1, 'name': 'LPD-Type', @@ -139,17 +143,6 @@ class CalcatWrapper: logging.info("Fetching all detector types") return _self.get_all_over_pages(DetectorType.get_all_by_name, None) - @functools.cached_property - def _detector_type_lowercase_to_calcat_case(self): - return { - det_type["name"].lower(): det_type["name"] - for det_type in self.get_detector_types() - } - - def calcat_case_detector_type(self, detector_type_name): - return self._detector_type_lowercase_to_calcat_case[ - detector_type_name.lower()] - @cache_fun def get_constant_types(_self): logging.info("Fetching all constant types") @@ -172,7 +165,7 @@ class CalcatWrapper: {'instrument_id': 1, 'detector_id': 1}, {'instrument_id': 7, 'detector_id': 1}] }, - ] + ] """ logging.info("Fetching all detectors") @@ -196,6 +189,7 @@ class CalcatWrapper: _self.cal_client.get_api_url(f"conditions/{condition_id}") ).json() + @cache_fun def get_recent_constant_versions_for_pdu( self, pdu_id, count=500, calibration_id=None, @@ -278,6 +272,7 @@ class CalcatWrapper: def get_detector_by_id(_self, detector_id): return Detector.get_by_id(_self.cal_client, detector_id)["data"] + @cache_fun def get_pdus_current_in_detector(self, detector_id, snapshot_at=None): # note: as not iterating over pages here, # will break if one detector contains > 100 modules @@ -287,18 +282,6 @@ class CalcatWrapper: pdu_snapshot_at=snapshot_at, ) - def get_pdu_ids_current_in_detector(self, detector_id, snapshot_at=None): - # note: as not iterating over pages here, - # will break if one detector contains > 100 modules - return [ - pdu["id"] - for pdu in PhysicalDetectorUnit.get_all_by_detector( - self.cal_client, - detector_id, - pdu_snapshot_at=snapshot_at, - )["data"] - ] - def get_pdu_names_current_in_detector(self, detector_id, snapshot_at=None): # note: as not iterating over pages here, # will break if one detector contains > 100 modules @@ -346,7 +329,8 @@ class CalcatWrapper: begin_at_after=start, ) return [ - ccv for ccv in ccvs if ccv["calibration_constant"]["condition_id"] in possible_conditions] + ccv for ccv in ccvs if ccv[ + "calibration_constant"]["condition_id"] in possible_conditions] @cache_fun def get_possible_conditions_detector( -- GitLab