Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/components/tomo/move_resource.py: 21%

53 statements  

« prev     ^ index     » next       coverage.py v7.6.4, created at 2024-11-14 02:13 +0000

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3import logging 

4from marshmallow import fields 

5import pint 

6 

7from daiquiri.core import marshal 

8from daiquiri.core.components import ComponentResource 

9from daiquiri.core.schema import ErrorSchema, MessageSchema 

10from daiquiri.core.logging import log 

11 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class TomoMoveResource(ComponentResource): 

17 @marshal( 

18 inp={ 

19 "sy": fields.Str(metadata={"description": "Quantity to move this motor"}), 

20 "sz": fields.Str(metadata={"description": "Quantity to move this motor"}), 

21 "sampx": fields.Str( 

22 metadata={"description": "Quantity to move of this motor"} 

23 ), 

24 "sampy": fields.Str( 

25 metadata={"description": "Quantity to move of this motor"} 

26 ), 

27 "sampu": fields.Str( 

28 metadata={"description": "Quantity to move of this motor"} 

29 ), 

30 "sampv": fields.Str( 

31 metadata={"description": "Quantity to move of this motor"} 

32 ), 

33 "relative": fields.Bool( 

34 metadata={"description": "If true, motor moves are relative"} 

35 ), 

36 }, 

37 out=[ 

38 [200, MessageSchema(), "Motor moved"], 

39 [400, ErrorSchema(), "Motor can't be moved"], 

40 [404, ErrorSchema(), "Motor not found"], 

41 ], 

42 ) 

43 def get( 

44 self, 

45 sy=None, 

46 sz=None, 

47 sampx=None, 

48 sampy=None, 

49 sampu=None, 

50 sampv=None, 

51 relative=False, 

52 **kwargs, 

53 ): 

54 """Get a list of all available detectors and their info""" 

55 if [sy, sz, sampx, sampy, sampu, sampv].count(None) == 6: 

56 return {"error": "No motors are part of the move request"}, 404 

57 

58 vsy = None 

59 vsz = None 

60 vsampx = None 

61 vsampy = None 

62 vsampu = None 

63 vsampv = None 

64 try: 

65 if sy: 

66 vsy = pint.Quantity(sy) 

67 if sampx: 

68 vsampx = pint.Quantity(sampx) 

69 if sampy: 

70 vsampy = pint.Quantity(sampy) 

71 if sz: 

72 vsz = pint.Quantity(sz) 

73 if sampu: 

74 vsampu = pint.Quantity(sampu) 

75 if sampv: 

76 vsampv = pint.Quantity(sampv) 

77 except Exception: 

78 logger.error( 

79 "One of requested motor position wrongly formatted (found '%s', '%s', '%s', '%s', '%s', '%s')", 

80 sy, 

81 sz, 

82 vsampx, 

83 sampy, 

84 sampu, 

85 sampv, 

86 exc_info=True, 

87 ) 

88 return {"error": "Problem while parsing request"}, 400 

89 

90 try: 

91 self._parent.move( 

92 sy=vsy, 

93 sz=vsz, 

94 sampx=vsampx, 

95 sampy=vsampy, 

96 sampu=vsampu, 

97 sampv=vsampv, 

98 relative=relative, 

99 ) 

100 except Exception as e: 

101 motors = [] 

102 

103 def include_if_value_not_none(name, value): 

104 if value is not None: 

105 motors.append(name) 

106 

107 include_if_value_not_none("sy", sy) 

108 include_if_value_not_none("sz", sz) 

109 include_if_value_not_none("sampx", sampy) 

110 include_if_value_not_none("sampy", sampy) 

111 include_if_value_not_none("sampu", sampu) 

112 include_if_value_not_none("sampv", sampv) 

113 motors = "/".join(motors) 

114 logger.error("Error while moving %s motors", motors, exc_info=True) 

115 log.get("user").error( 

116 f"Move motors {motors} cancelled: {e.args[0]}", type="hardware" 

117 ) 

118 return {"error": f"Move motors {motors} cancelled: {e.args[0]}"}, 400 

119 return {"message": "Motors was moved"}, 200