ceph/src/spdk/ocf/tests/functional/pyocf/utils.py

   1 #
   2 # Copyright(c) 2019 Intel Corporation
   3 # SPDX-License-Identifier: BSD-3-Clause-Clear
   4 #
   5
   6 from ctypes import string_at
   7
   8
   9 def print_buffer(
  10     buf,
  11     length,
  12     offset=0,
  13     width=16,
  14     ignore=0,
  15     stop_after_count_ignored=0,
  16     print_fcn=print,
  17 ):
  18     end = int(offset) + int(length)
  19     offset = int(offset)
  20     ignored_lines = 0
  21     buf = string_at(buf, length)
  22     whole_buffer_ignored = True
  23     stop_after_count_ignored = int(stop_after_count_ignored / width)
  24
  25     for addr in range(offset, end, width):
  26         cur_line = buf[addr : min(end, addr + width)]
  27         byteline = ""
  28         asciiline = ""
  29         if not any(x != ignore for x in cur_line):
  30             if (
  31                 stop_after_count_ignored
  32                 and ignored_lines > stop_after_count_ignored
  33             ):
  34                 print_fcn(
  35                     "<{} bytes of '0x{:02X}' encountered, stopping>".format(
  36                         stop_after_count_ignored * width, ignore
  37                     )
  38                 )
  39                 return
  40             ignored_lines += 1
  41             continue
  42
  43         if ignored_lines:
  44             print_fcn(
  45                 "<{} of '0x{:02X}' bytes omitted>".format(
  46                     ignored_lines * width, ignore
  47                 )
  48             )
  49             ignored_lines = 0
  50
  51         for byte in cur_line:
  52             byte = int(byte)
  53             byteline += "{:02X} ".format(byte)
  54             if 31 < byte < 126:
  55                 char = chr(byte)
  56             else:
  57                 char = "."
  58             asciiline += char
  59
  60         print_fcn("0x{:08X}\t{}\t{}".format(addr, byteline, asciiline))
  61         whole_buffer_ignored = False
  62
  63     if whole_buffer_ignored:
  64         print_fcn("<whole buffer ignored>")
  65     elif ignored_lines:
  66         print_fcn("<'0x{:02X}' until end>".format(ignore))
  67
  68
  69 class Size:
  70     _KiB = 1024
  71     _MiB = _KiB * 1024
  72     _GiB = _MiB * 1024
  73     _TiB = _GiB * 1024
  74     _SECTOR_SIZE = 512
  75
  76     def __init__(self, b: int, sector_aligned: bool = False):
  77         if sector_aligned:
  78             self.bytes = int(
  79                 ((b + self._SECTOR_SIZE - 1) // self._SECTOR_SIZE)
  80                 * self._SECTOR_SIZE
  81             )
  82         else:
  83             self.bytes = int(b)
  84
  85     def __int__(self):
  86         return self.bytes
  87
  88     def __index__(self):
  89         return self.bytes
  90
  91     @classmethod
  92     def from_B(cls, value, sector_aligned=False):
  93         return cls(value, sector_aligned)
  94
  95     @classmethod
  96     def from_KiB(cls, value, sector_aligned=False):
  97         return cls(value * cls._KiB, sector_aligned)
  98
  99     @classmethod
 100     def from_MiB(cls, value, sector_aligned=False):
 101         return cls(value * cls._MiB, sector_aligned)
 102
 103     @classmethod
 104     def from_GiB(cls, value, sector_aligned=False):
 105         return cls(value * cls._GiB, sector_aligned)
 106
 107     @classmethod
 108     def from_TiB(cls, value, sector_aligned=False):
 109         return cls(value * cls._TiB, sector_aligned)
 110
 111     @classmethod
 112     def from_sector(cls, value):
 113         return cls(value * cls._SECTOR_SIZE)
 114
 115     @property
 116     def B(self):
 117         return self.bytes
 118
 119     @property
 120     def KiB(self):
 121         return self.bytes / self._KiB
 122
 123     @property
 124     def MiB(self):
 125         return self.bytes / self._MiB
 126
 127     @property
 128     def GiB(self):
 129         return self.bytes / self._GiB
 130
 131     @property
 132     def TiB(self):
 133         return self.bytes / self._TiB
 134
 135     @property
 136     def sectors(self):
 137         return self.bytes // self._SECTOR_SIZE
 138
 139     def __str__(self):
 140         if self.bytes < self._KiB:
 141             return "{} B".format(self.B)
 142         elif self.bytes < self._MiB:
 143             return "{} KiB".format(self.KiB)
 144         elif self.bytes < self._GiB:
 145             return "{} MiB".format(self.MiB)
 146         elif self.bytes < self._TiB:
 147             return "{} GiB".format(self.GiB)
 148         else:
 149             return "{} TiB".format(self.TiB)
 150
 151
 152 def print_structure(struct, indent=0):
 153     print(struct)
 154     for field, field_type in struct._fields_:
 155         value = getattr(struct, field)
 156         if hasattr(value, "_fields_"):
 157             print("{}{: <20} :".format("   " * indent, field))
 158             print_structure(value, indent=indent + 1)
 159             continue
 160
 161         print("{}{: <20} : {}".format("   " * indent, field, value))
 162
 163
 164 def struct_to_dict(struct):
 165     d = {}
 166     for field, field_type in struct._fields_:
 167         value = getattr(struct, field)
 168         if hasattr(value, "_fields_"):
 169             d[field] = struct_to_dict(value)
 170             continue
 171         d[field] = value
 172
 173     return d