Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/hardware/hwr/__init__.py: 0%

1# -*- coding: utf-8 -*- 

2""" Protocol handler for HardwareRepsitory HardwareObjects""" 

3import os 

4import sys 

5import logging 

6 

7from marshmallow import ValidationError, fields 

8 

9from daiquiri.core.hardware.abstract import ProtocolHandler 

10from daiquiri.core.schema.hardware import HOConfigSchema 

11from daiquiri.core.utils import loader 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class HwrHOConfigSchema(HOConfigSchema): 

17 address = fields.Str(required=True) 

18 type = fields.Str(required=True) 

19 

20 

21class HwrHandler(ProtocolHandler): 

22 def __init__(self, *args, **kwargs): 

23 super(HwrHandler, self).__init__(args, kwargs) 

24 

25 hwr_root = os.path.abspath(os.environ.get("HWR_ROOT", "")) 

26 

27 if hwr_root: 

28 logging.info("Using HardwareRepository from %s", hwr_root) 

29 sys.path.insert(0, hwr_root) 

30 

31 try: 

32 # pylint: disable=import-error 

33 from HardwareRepository import HardwareRepository as hwr 

34 

35 logging.info("Using system HardwareRepository from %s", hwr.__file__) 

36 except (ImportError, ModuleNotFoundError): 

37 raise 

38 

39 self._hwr = hwr.getHardwareRepository(kwargs.get("path", None)) 

40 self._hwr.connect() 

41 

42 hwr_logger = logging.getLogger("HWR") 

43 hwr_logger.setLevel(logging.DEBUG) 

44 

45 def get(self, *args, **kwargs): 

46 try: 

47 HwrHOConfigSchema().load(kwargs) 

48 except ValidationError as ex: 

49 raise Exception("HWR Hardware Object definition is invalid %s" % str(ex)) 

50 

51 kwargs["obj"] = self._hwr.getHardwareObject(kwargs.get("address")) 

52 

53 try: 

54 return loader( 

55 "daiquiri.core.hardware.hwr", "", kwargs.get("type"), **kwargs 

56 ) 

57 except Exception as e: 

58 logger.error(f"Could not find class for HWR object {kwargs.get('type')}") 

59 raise e