Skip to content

UltraLoader

UltraLoader

Bases: KoshLoader

Kosh Loader for ultra files

Source code in kosh/loaders/UltraLoader.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class UltraLoader(KoshLoader):
    """Kosh Loader for ultra files"""
    types = {"ultra": ["curves", "curves/pydv", "curves/pdv", "pydv", "pdv",
                       "dict", "numpy"]}

    def __init__(self, obj, **kargs):
        super(UltraLoader, self).__init__(obj, **kargs)
        self.curves = None

    def load_curves(self):
        # pydv import matplotlib.pyplot
        # on some systems with no X forwarding this causes
        # an uncatchable error.
        # Setting the matplotlib backend to a windowless
        # backend fixes this.
        if "DISPLAY" not in os.environ or os.environ["DISPLAY"] == "":
            import matplotlib
            matplotlib.use("agg", force=True)
        try:
            import pydvpy as pydvif
        except ImportError:
            import pydv
            sys.path.append(pydv.__path__[0])
            import pydv.pydvpy as pydvif
        self.curves = pydvif.read(self.uri)

    def load_from_ultra(self, variable):
        """Load variables from an ultra file
        :param variable: variables to load
        :type variable: list or str
        :return Dictionary containing 'x-axis' and 'y-axis' for each variable
        :rtype: dict
        """
        if self.curves is None:
            self.load_curves()
        if not isinstance(variable, (list, tuple)) and variable is not None:  # only one variable requested
            variable = [variable, ]

        pydv_format = self.format in ["curves", "curves/pydv", "curves/pdv", "pydv", "pdv"]

        if pydv_format or self.format == "numpy":
            variables = []
        else:
            variables = {}

        if variable is None:  # all curves
            if pydv_format:
                return self.curves
            elif self.format == "numpy":
                return numpy.array([[c.x, c.y] for c in self.curves])
            else:
                for c in self.curves:
                    name = c.name
                    variables[name] = {}
                    variables[name]['x-axis'] = c.x
                    variables[name]['y-axis'] = c.y
        else:
            for var in variable:
                for c in self.curves:
                    name = c.name
                    if name == var:
                        if pydv_format:
                            variables.append(c)
                        elif self.format == "numpy":
                            variables.append(numpy.array([c.x, c.y]))
                        else:
                            variables[name] = {}
                            variables[name]['x-axis'] = c.x
                            variables[name]['y-axis'] = c.y
                        break

        if self.format == "numpy":
            variables = numpy.array(variables)

        if len(variable) == 1:
            return variables[0]
        else:
            return variables

    def extract(self, *args, **kargs):
        """Extract a feature"""
        return self.load_from_ultra(self.feature)

    def open(self):
        """open/load matching ultra file

        :return: Dictionary containing 'x-axis' and 'y-axis' for each variable
        """
        return self.load_from_ultra(None)

    def list_features(self):
        """List features available in ultra file"""
        variables = []
        if self.curves is None:
            self.load_curves()
        for curve in self.curves:
            variables.append(curve.name)
        return variables

    def describe_feature(self, feature):
        """Describe a feature

        :param feature: feature to describe
        :type feature: str
        :return: dictionary with attributes describing the feature:
                 'name', 'size', 'first_time', 'last_time', 'min', 'max', 'type'
        :rtype: dict
        """
        info = {"name": feature}
        if self.curves is None:
            self.load_curves()
        for c in self.curves:
            if c.name == feature:
                info["size"] = len(c.x)
                info["first_time"] = c.x[0]
                info["last_time"] = c.x[-1]
                info["min"] = min(c.y)
                info["max"] = max(c.y)
                info["type"] = c.y.dtype
                break
        return info

describe_feature(feature)

Describe a feature

Parameters:

Name Type Description Default
feature str

feature to describe

 required

Returns:

Type Description
dict

dictionary with attributes describing the feature: 'name', 'size', 'first_time', 'last_time', 'min', 'max', 'type'
Source code in kosh/loaders/UltraLoader.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def describe_feature(self, feature):
    """Describe a feature

    :param feature: feature to describe
    :type feature: str
    :return: dictionary with attributes describing the feature:
             'name', 'size', 'first_time', 'last_time', 'min', 'max', 'type'
    :rtype: dict
    """
    info = {"name": feature}
    if self.curves is None:
        self.load_curves()
    for c in self.curves:
        if c.name == feature:
            info["size"] = len(c.x)
            info["first_time"] = c.x[0]
            info["last_time"] = c.x[-1]
            info["min"] = min(c.y)
            info["max"] = max(c.y)
            info["type"] = c.y.dtype
            break
    return info

extract(*args, **kargs)

Extract a feature

Source code in kosh/loaders/UltraLoader.py 
88
89
90
def extract(self, *args, **kargs):
    """Extract a feature"""
    return self.load_from_ultra(self.feature)

list_features()

List features available in ultra file

Source code in kosh/loaders/UltraLoader.py 
 99
100
101
102
103
104
105
106
def list_features(self):
    """List features available in ultra file"""
    variables = []
    if self.curves is None:
        self.load_curves()
    for curve in self.curves:
        variables.append(curve.name)
    return variables

load_from_ultra(variable)

Load variables from an ultra file

Parameters:

Name Type Description Default
variable list | str

variables to load

 required
Source code in kosh/loaders/UltraLoader.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def load_from_ultra(self, variable):
    """Load variables from an ultra file
    :param variable: variables to load
    :type variable: list or str
    :return Dictionary containing 'x-axis' and 'y-axis' for each variable
    :rtype: dict
    """
    if self.curves is None:
        self.load_curves()
    if not isinstance(variable, (list, tuple)) and variable is not None:  # only one variable requested
        variable = [variable, ]

    pydv_format = self.format in ["curves", "curves/pydv", "curves/pdv", "pydv", "pdv"]

    if pydv_format or self.format == "numpy":
        variables = []
    else:
        variables = {}

    if variable is None:  # all curves
        if pydv_format:
            return self.curves
        elif self.format == "numpy":
            return numpy.array([[c.x, c.y] for c in self.curves])
        else:
            for c in self.curves:
                name = c.name
                variables[name] = {}
                variables[name]['x-axis'] = c.x
                variables[name]['y-axis'] = c.y
    else:
        for var in variable:
            for c in self.curves:
                name = c.name
                if name == var:
                    if pydv_format:
                        variables.append(c)
                    elif self.format == "numpy":
                        variables.append(numpy.array([c.x, c.y]))
                    else:
                        variables[name] = {}
                        variables[name]['x-axis'] = c.x
                        variables[name]['y-axis'] = c.y
                    break

    if self.format == "numpy":
        variables = numpy.array(variables)

    if len(variable) == 1:
        return variables[0]
    else:
        return variables

open()

open/load matching ultra file

Returns:

Type Description

Dictionary containing 'x-axis' and 'y-axis' for each variable
Source code in kosh/loaders/UltraLoader.py 
92
93
94
95
96
97
def open(self):
    """open/load matching ultra file

    :return: Dictionary containing 'x-axis' and 'y-axis' for each variable
    """
    return self.load_from_ultra(None)