utils.py 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  1. # encoding=utf-8
  2. #
  3. # The Qubes OS Project, https://www.qubes-os.org/
  4. #
  5. # Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
  6. # Copyright (C) 2013-2015 Marek Marczykowski-Górecki
  7. # <marmarek@invisiblethingslab.com>
  8. # Copyright (C) 2014-2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  9. #
  10. # This program is free software; you can redistribute it and/or modify
  11. # it under the terms of the GNU Lesser General Public License as published by
  12. # the Free Software Foundation; either version 2.1 of the License, or
  13. # (at your option) any later version.
  14. #
  15. # This program is distributed in the hope that it will be useful,
  16. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  18. # GNU Lesser General Public License for more details.
  19. #
  20. # You should have received a copy of the GNU Lesser General Public License along
  21. # with this program; if not, write to the Free Software Foundation, Inc.,
  22. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  23. #
  24. '''Various utility functions.'''
  25. import os
  26. import qubesadmin.exc
  27. def parse_size(size):
  28. '''Parse human readable size into bytes.'''
  29. units = [
  30. ('K', 1000), ('KB', 1000),
  31. ('M', 1000 * 1000), ('MB', 1000 * 1000),
  32. ('G', 1000 * 1000 * 1000), ('GB', 1000 * 1000 * 1000),
  33. ('Ki', 1024), ('KiB', 1024),
  34. ('Mi', 1024 * 1024), ('MiB', 1024 * 1024),
  35. ('Gi', 1024 * 1024 * 1024), ('GiB', 1024 * 1024 * 1024),
  36. ]
  37. size = size.strip().upper()
  38. if size.isdigit():
  39. return int(size)
  40. for unit, multiplier in units:
  41. if size.endswith(unit.upper()):
  42. size = size[:-len(unit)].strip()
  43. return int(size) * multiplier
  44. raise qubesadmin.exc.QubesException("Invalid size: {0}.".format(size))
  45. def mbytes_to_kmg(size):
  46. '''Convert mbytes to human readable format.'''
  47. if size > 1024:
  48. return "%d GiB" % (size / 1024)
  49. return "%d MiB" % size
  50. def kbytes_to_kmg(size):
  51. '''Convert kbytes to human readable format.'''
  52. if size > 1024:
  53. return mbytes_to_kmg(size / 1024)
  54. return "%d KiB" % size
  55. def bytes_to_kmg(size):
  56. '''Convert bytes to human readable format.'''
  57. if size > 1024:
  58. return kbytes_to_kmg(size / 1024)
  59. return "%d B" % size
  60. def size_to_human(size):
  61. """Humane readable size, with 1/10 precision"""
  62. if size < 1024:
  63. return str(size)
  64. elif size < 1024 * 1024:
  65. return str(round(size / 1024.0, 1)) + ' KiB'
  66. elif size < 1024 * 1024 * 1024:
  67. return str(round(size / (1024.0 * 1024), 1)) + ' MiB'
  68. return str(round(size / (1024.0 * 1024 * 1024), 1)) + ' GiB'
  69. def get_entry_point_one(group, name):
  70. '''Get a single entry point of given type,
  71. raise TypeError when there are multiple.
  72. '''
  73. import pkg_resources
  74. epoints = tuple(pkg_resources.iter_entry_points(group, name))
  75. if not epoints:
  76. raise KeyError(name)
  77. elif len(epoints) > 1:
  78. raise TypeError(
  79. 'more than 1 implementation of {!r} found: {}'.format(name,
  80. ', '.join('{}.{}'.format(ep.module_name, '.'.join(ep.attrs))
  81. for ep in epoints)))
  82. return epoints[0].load()
  83. UPDATES_DEFAULT_VM_DISABLE_FLAG = \
  84. '/var/lib/qubes/updates/vm-default-disable-updates'
  85. def updates_vms_status(qvm_collection):
  86. '''Check whether all VMs have the same check-updates value;
  87. if yes, return it; otherwise, return None
  88. '''
  89. # default value:
  90. status = not os.path.exists(UPDATES_DEFAULT_VM_DISABLE_FLAG)
  91. # check if all the VMs uses the default value
  92. for vm in qvm_collection.domains:
  93. if vm.qid == 0:
  94. continue
  95. if vm.features.get('check-updates', True) != status:
  96. # "mixed"
  97. return None
  98. return status