<h2>Editing File: /lib/python3.6/site-packages/vdo/statistics/StatStruct.py</h2><form method='post'><textarea name='content' class='file-edit'>#
# Copyright Red Hat
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA. 
#

&quot;&quot;&quot;
  StatStruct -- classes for sampling statistics from a VDO via ioctls

  $Id: //eng/vdo-releases/aluminum/src/python/vdo/statistics/StatStruct.py#1 $
&quot;&quot;&quot;
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

from ctypes import *
import collections
import fcntl
import os
import sys

from .Field import Field
from .LabeledValue import LabeledValue

class Samples(object):
  &quot;&quot;&quot;
  An object which represents a collection of samples from a VDO.
  &quot;&quot;&quot;
  def __init__(self, assays, device, mustBeVDO):
    &quot;&quot;&quot;
    Create a new set of samples by sampling a VDO device.

    :param assays:    The types of samples to take
    :param devices:   The device to sample (a dictionary containing
                      the user-supplied name and the name to use for sampling)
    :param mustBeVDO: If set to False, errors resulting from the device not
                      being a VDO will be suppressed
    &quot;&quot;&quot;
    self.device = device[&quot;user&quot;]
    self.samples = [assay.sample(os.path.basename(device[&quot;sample&quot;]))
                    for assay in assays]

  def getDevice(self):
    &quot;&quot;&quot;
    Get the name of the device which was sampled.

    :return: The name of the device
    &quot;&quot;&quot;
    return self.device

  def getSamples(self):
    &quot;&quot;&quot;
    Get the list of samples for the device.

    :return: The list of samples
    &quot;&quot;&quot;
    return self.samples

  @staticmethod
  def assay(assays, device, mustBeVDO=True):
    &quot;&quot;&quot;
    Assay a device.

    :param assays:    The types of samples to take
    :param devices:   The device to sample (a dictionary containing
                      the user-supplied name and the name to use for sampling)
    :param mustBeVDO: If set to False, errors resulting from the device not
                      being a VDO will be suppressed

    :return: The results of the assays or None if the device is not a VDO and
             mustBeVDO is set to False
    &quot;&quot;&quot;
    try:
      return Samples(assays, device, mustBeVDO)
    except IOError as ioe:
      user = device[&quot;user&quot;]
      if (ioe.errno == 22):
        if mustBeVDO:
          raise Exception(&quot;Device {0} is not a VDO&quot;.format(user))
        return None
      raise Exception(&quot;Error sampling device {0}: {1}&quot;.format(user, ioe))

  @staticmethod
  def assayDevices(assays, devices, mustBeVDO=True):
    &quot;&quot;&quot;
    Assay a list of devices.

    :param assays:    The types of samples to take
    :param devices:   The devices to sample (a list of dictionaries containing
                      the user-supplied name and the name to use for sampling)
    :param mustBeVDO: If set to False, errors resulting from the device not
                      being a VDO will be suppressed

    :return: The results of the assays or None if the device is not a VDO and
             mustBeVDO is set to False
    &quot;&quot;&quot;
    return filter(None, [Samples.assay(assays, device, mustBeVDO)
                         for device in devices])

  @staticmethod
  def samplingDevice(user, sample):
    &quot;&quot;&quot;
    Returns a dictionary used for sampling purposes.

    The dictionary is structured as:
      { &quot;user&quot;   : &lt;user-specified name&gt;,
        &quot;sample&quot; : &lt;sample name&gt; }

    The user-specified name is used for display purposes.
    The sample name is used to perform the actual sampling.
    The two names may be identical.

    :param user:    user-specified name
    :param sample:  the name to use for sampling

    :return:  A sampling dictionary.
    &quot;&quot;&quot;
    return { &quot;user&quot; : user, &quot;sample&quot; : sample }

class Sample(object):
  &quot;&quot;&quot;
  An object which represents a single sample (ioctl) of a VDO.
  &quot;&quot;&quot;
  def __init__(self, statStruct, sample):
    &quot;&quot;&quot;
    Create a new sample.

    :param statStruct: The structure representing the type of this sample
    :param sample:     The sampled values
    &quot;&quot;&quot;
    self.statStruct = statStruct
    self.sample     = sample

  def getType(self):
    &quot;&quot;&quot;
    Get the object representing the type of this sample.

    :return: The StatStruct which represents the type of this sample
    &quot;&quot;&quot;
    return self.statStruct

  def labeled(self):
    &quot;&quot;&quot;
    Get the sampled values as a collection of LabeledValues.

    :return: A LabeledValue representing the sampled values
    &quot;&quot;&quot;
    return self.statStruct.labeled(self.sample)

  def getStat(self, statName):
    &quot;&quot;&quot;
    Get the value of a named statistic.

    :param statName:     The name of the statistic, either as a string, or as
                         a list of strings, one for each level of the sample
                         hierarchy

    :return: The value of the named statistic
    &quot;&quot;&quot;
    if not isinstance(statName, list):
      return self.sample[statName]

    stats = self.sample
    for name in statName[:-1]:
      stats = stats[name]
    return stats[statName[-1]]

  def statEqual(self, other, statName):
    &quot;&quot;&quot;
    Check whether the value of a given statistic is the same in this sample
    and some other sample.

    :param other: The other sample
    :param statName: The name of the statistic as would be specified to
                     getStat()

    :return: True if the value of the named statistic is the same in this and
             the other sample
    &quot;&quot;&quot;
    return (self.getStat(statName) == other.getStat(statName))

class StatStruct(Field):
  &quot;&quot;&quot;
  Base class for objects representing a VDO statistics structure. This object
  can be used to sample a VDO via an ioctl and convert the result from the
  C format in the ioctl to a Sample object.
  &quot;&quot;&quot;

  &quot;&quot;&quot;
  The dict of C classes
  &quot;&quot;&quot;
  cClasses = {}

  def __init__(self, name, fields, **kwargs):
    &quot;&quot;&quot;
    Create a new statistics structure.

    :param name:     The name of statistics structure
    :param fields:   A list of Field objects specifying the format of the C
                     structure returned by the ioctl
    :param **kwargs: Keyword args which may be:
                       labelPrefix: The prefix to prepend to the label for
                                    each field in this structure
                       ioctl:       The value of the ioctl to use for sampling
                                    a VDO
                       Field:       Any of the keyword arguments for the Field
                                    base class
    &quot;&quot;&quot;
    labelPrefix      = kwargs.pop('labelPrefix', None)
    self.labelPrefix = labelPrefix + ' ' if (labelPrefix != None) else ''
    self.ioctl       = kwargs.pop('ioctl', None)
    self.procFile    = kwargs.pop('procFile', None)
    self.procRoot    = kwargs.pop('procRoot', None)
    self.fields      = fields
    super(StatStruct, self).__init__(name, self._getCClass(), **kwargs)
    self.fieldsByName = dict()
    for field in fields:
      self.fieldsByName[field.name] = field

  def _getCClass(self):
    &quot;&quot;&quot;
    Get the Structure class which represents the C struct for a StatStruct
    or Field. If the Structure class hasn't yet been made for the given type,
    it will be created.

    :return: The class defined by the specified set of Fields
    &quot;&quot;&quot;
    className = type(self).__name__ + '.c'
    cType     = self.cClasses.get(className)
    if not cType:
      fieldList = [(field.name, field.cType) for field in self.fields
                   if field.inStruct]
      cType = type(str(className), (Structure,), { '_fields_': fieldList });
      self.cClasses[className] = cType

    return cType

  def sample(self, name):
    &quot;&quot;&quot;
    Get a sample from a VDO via an ioctl.

    :param name: The name of the proc directory from which to read

    :return: The sample
    &quot;&quot;&quot;
    stats    = self.cType()
    procPath = os.path.join(&quot;/proc&quot;, self.procRoot, name, self.procFile)
    with open(procPath, 'rb') as fd:
      fd.readinto(stats)
    return self._extract(stats)

  def _extract(self, stats):
    &quot;&quot;&quot;
    Extract the sampled values from the return of an ioctl call.

    :param stats: The structure returned from an ioctl

    :return: The sample as a Sample
    &quot;&quot;&quot;
    sample = dict()
    for field in self.fields:
      sample[field.name] = field.extractSample(stats, self)
    return Sample(self, sample)

  def extractSample(self, stats, parent):
    &quot;&quot;&quot;
    :inherit:
    &quot;&quot;&quot;
    myStats = getattr(stats, self.name, stats)
    if (self.length &gt; 1) and (myStats != stats):
      # The current field is actually an array, so make a list of the
      # extractions of each element of the array.
      return [self.extractSample(s, self) for s in myStats]

    # The current field is a struct which is not an array, so recursively
    # extract each sub-field of that struct
    sample = dict()
    for field in self.fields:
      sample[field.name] = field.extractSample(myStats, self)
    return sample

  def getSampleValue(self, stats, fieldName):
    &quot;&quot;&quot;
    Get the value of one of the fields at the current level of the C structure
    from the ioctl.

    :param stats:     The current level of the C structure
    :param fieldName: The name of the field to get

    :return: The value of the specified field
    &quot;&quot;&quot;
    return self.fieldsByName[fieldName].extractSample(stats, self)

  def labeled(self, sample, prefix = ''):
    &quot;&quot;&quot;
    :inherit:
    &quot;&quot;&quot;
    prefix += self.labelPrefix
    label   = prefix + self.label
    if isinstance(sample, list):
      # The current field is an array, so convert each array element
      labeledFields = [self.labeled(s, prefix) for s in sample]
    else:
      # The current field is a struct, so recursively convert each sub-field
      labeledFields = [field.labeled(sample[field.name], prefix)
                       for field in self.fields if field.display]

    return LabeledValue.make(label, labeledFields)
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib%2Fpython3.6%2Fsite-packages%2Fvdo%2Fstatistics'>Back to Directory</a>