]>
git.proxmox.com Git - ceph.git/blob - ceph/src/python-common/ceph/deployment/inventory.py
570c1dbb3615e0cd41ac24e7a08ac12ee35bea1f
2 from typing
import List
, Optional
, Dict
, Any
, Union
4 pass # for type checking
6 from ceph
.utils
import datetime_now
, datetime_to_str
, str_to_datetime
11 class Devices(object):
13 A container for Device instances with reporting
16 def __init__(self
, devices
):
17 # type: (List[Device]) -> None
18 self
.devices
= devices
# type: List[Device]
20 def __eq__(self
, other
: Any
) -> bool:
21 if not isinstance(other
, Devices
):
23 if len(self
.devices
) != len(other
.devices
):
25 for d1
, d2
in zip(other
.devices
, self
.devices
):
31 # type: () -> List[dict]
32 return [d
.to_json() for d
in self
.devices
]
35 def from_json(cls
, input):
36 # type: (List[Dict[str, Any]]) -> Devices
37 return cls([Device
.from_json(i
) for i
in input])
41 return Devices(devices
=list(self
.devices
))
53 'human_readable_type',
60 sys_api
=None, # type: Optional[Dict[str, Any]]
61 available
=None, # type: Optional[bool]
62 rejected_reasons
=None, # type: Optional[List[str]]
63 lvs
=None, # type: Optional[List[str]]
64 device_id
=None, # type: Optional[str]
65 lsm_data
=None, # type: Optional[Dict[str, Dict[str, str]]]
66 created
=None, # type: Optional[datetime.datetime]
67 ceph_device
=None # type: Optional[bool]
70 self
.sys_api
= sys_api
if sys_api
is not None else {} # type: Dict[str, Any]
71 self
.available
= available
72 self
.rejected_reasons
= rejected_reasons
if rejected_reasons
is not None else []
74 self
.device_id
= device_id
75 self
.lsm_data
= lsm_data
if lsm_data
is not None else {} # type: Dict[str, Dict[str, str]]
76 self
.created
= created
if created
is not None else datetime_now()
77 self
.ceph_device
= ceph_device
79 def __eq__(self
, other
):
81 if not isinstance(other
, Device
):
83 diff
= [k
for k
in self
.report_fields
if k
!= 'created' and (getattr(self
, k
)
84 != getattr(other
, k
))]
90 k
: (getattr(self
, k
) if k
!= 'created'
91 or not isinstance(getattr(self
, k
), datetime
.datetime
)
92 else datetime_to_str(getattr(self
, k
)))
93 for k
in self
.report_fields
97 def from_json(cls
, input):
98 # type: (Dict[str, Any]) -> Device
99 if not isinstance(input, dict):
100 raise ValueError('Device: Expected dict. Got `{}...`'.format(json
.dumps(input)[:10]))
103 key
: (input.get(key
, None) if key
!= 'created'
104 or not input.get(key
, None)
105 else str_to_datetime(input.get(key
, None)))
106 for key
in Device
.report_fields
107 if key
!= 'human_readable_type'
110 if ret
.rejected_reasons
:
111 ret
.rejected_reasons
= sorted(ret
.rejected_reasons
)
115 def human_readable_type(self
):
117 if self
.sys_api
is None or 'rotational' not in self
.sys_api
:
119 return 'hdd' if self
.sys_api
["rotational"] == "1" else 'ssd'
121 def __repr__(self
) -> str:
122 device_desc
: Dict
[str, Union
[str, List
[str]]] = {
123 'path': self
.path
if self
.path
is not None else 'unknown',
124 'lvs': self
.lvs
if self
.lvs
else 'None',
125 'available': str(self
.available
),
126 'ceph_device': str(self
.ceph_device
)
128 if not self
.available
and self
.rejected_reasons
:
129 device_desc
['rejection reasons'] = self
.rejected_reasons
130 return "Device({})".format(
131 ', '.join('{}={}'.format(key
, device_desc
[key
]) for key
in device_desc
.keys())