Skip to content

API Documentation

This API is still in development and may change without notice.

create_script()

Create a minimal script with available cues.

This function creates a minimal script with available cues. It includes an audio cue, a video cue and an action cue. The script is returned as a CuemsScript object.

Returns:

Name Type Description
CuemsScript

A minimal script with configured cues.
Source code in src/cuemsutils/create_script.py 
 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def create_script():
    """Create a minimal script with available cues.

    This function creates a minimal script with available cues.
    It includes an audio cue, a video cue and an action cue.
    The script is returned as a CuemsScript object.

    Returns:
        CuemsScript: A minimal script with configured cues.
    """
    target_uuid = '1f301cf8-dd03-4b40-ac17-ef0e5e7988be'
    act = ActionCue({'action_target': target_uuid, 'action_type': 'play', 'ui_properties' : {'warning' : 0}})
    ac = AudioCue({
        'master_vol': 66,
        'Media': Media({
            'file_name': 'file.ext',
            'id': '',
            'duration': '00:00:00.000',
            'regions': [
                Region({
                    'id': 0,
                    'loop': 1,
                    'in_time': None,
                    'out_time': None
                })
            ]
        }),
        'ui_properties' : {
            'warning': None
            }
    })
    vc = VideoCue({
        'Media': Media({
            'file_name': 'file_video.ext',
            'id': '',
            'duration': '00:00:00.000',
            'regions' : [
                Region({
                    'id': 0, 'loop': 1, 'in_time': None, 'out_time': None
                })
            ]
        }),
        'ui_properties' : {
            'warning': None
            }
    })
    dc = DmxCue({
        'fadein_time':0.0,
        'fadeout_time':0.0,
        'DmxScene': DmxScene({
            'id': 0,
            'DmxUniverse': DmxUniverse({
                'universe_num': 0,
                'dmx_channels': [
                    DmxChannel({
                        'channel': 0,
                        'value': 0
                    })
                ]

            })

        }),
        'time': 0,
        'ui_properties' : {
            'warning': None
            }
    })
    ac.outputs = [AudioCueOutput({
        "output_name": "0367f391-ebf4-48b2-9f26-000000000001_system:playback_1",
        "output_vol": 80,
        "channels": [
            {
                "channel": {
                    "channel_num": 0,
                    "channel_vol": 80
                }
            }
        ]
    })]

    vc.outputs = [
        VideoCueOutput({
            "output_name": "0367f391-ebf4-48b2-9f26-000000000001_0",
            "output_geometry": {
                "x_scale": 1,
                "y_scale": 1,
                "corners": {
                    "top_left": {
                        "x": 0,
                        "y": 0
                    },
                    "top_right": {
                        "x": 0,
                        "y": 0
                    },
                    "bottom_left": {
                        "x": 0,
                        "y": 0
                    },
                    "bottom_right": {
                        "x": 0,
                        "y": 0
                    }
                }
            }
        }),
        VideoCueOutput({
            "output_name": "0367f391-ebf4-48b2-9f26-000000000001_custom_0",
            "output_geometry": {
                "x_scale": 1,
                "y_scale": 1,
                "corners": {
                    "top_left": {"x": 0, "y": 0},
                    "top_right": {"x": 0, "y": 0},
                    "bottom_left": {"x": 0, "y": 0},
                    "bottom_right": {"x": 0, "y": 0}
                }
            },
            "canvas_region": {
                "x": 0.1,
                "y": 0.1,
                "width": 0.5,
                "height": 0.5
            }
        })
    ]

    dc.outputs = [DmxCueOutput({
        "output_name": "0367f391-ebf4-48b2-9f26-000000000001"
    })]


    fc = FadeCue({
        'action_target': target_uuid,
        'curve_type': FadeCurveType.linear,
        'duration': '00:00:02.000',
        'target_value': 0,
        'ui_properties': {'warning': None},
    })

    custom_cue_list = CueList({'contents': [ac]})
    custom_cue_list.append(vc)
    custom_cue_list.append(dc)
    custom_cue_list.append(act)
    custom_cue_list.append(fc)

    script = CuemsScript({'CueList': custom_cue_list})
    script.name = "Test Script"
    script.description = "This is a test script"

    # set dates and ids so it can be validated
    script.created = now
    script.modified = now
    script['id'] = new_uuid()
    script['CueList']['id'] = new_uuid()
    script.cuelist['contents'][0]['id'] = new_uuid()
    script.cuelist['contents'][1]['id'] = new_uuid()
    script.cuelist['contents'][2]['id'] = new_uuid()
    script.cuelist['contents'][3]['id'] = new_uuid()
    script.cuelist['contents'][4]['id'] = new_uuid()
    script['ui_properties'] = {
        'warning': 0,
    }
    Logger.debug(f'Created test script: {script.cuelist}')

    validate_template(script)

    # remove dates and ids so we send it empty
    script.created = None
    script.modified = None
    script.id = None
    script.cuelist.id = None
    script.cuelist.contents[0]['id'] = None
    script.cuelist.contents[1]['id'] = None
    script.cuelist.contents[2]['id'] = None
    script.cuelist.contents[3]['id'] = None
    script.cuelist.contents[4]['id'] = None

    return script

Set of helper functions for the cuemsutils package.

CuemsDict

Bases: dict

Custom dictionary class to handle cuemsutils specific items.

Source code in src/cuemsutils/helpers.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class CuemsDict(dict):
    """Custom dictionary class to handle cuemsutils specific items."""

    def build(self, parent: Element):
        build_xml_dict(self, parent)

    def setter(self, settings: dict):
        """Set the object properties from a dictionary.

        Args:
            settings (dict): Dictionary containing property values to set.

        Raises:
            AttributeError: If settings is not a dictionary.
        """
        if not isinstance(settings, dict):
            raise AttributeError(f"Invalid type {type(settings)}. Expected dict.")
        for k, v in settings.items():
            try:
                x = getattr(self, f"set_{k}")
                x(v)
            except AttributeError:
                pass

setter(settings)

Set the object properties from a dictionary.

Parameters:

Name Type Description Default
settings dict

Dictionary containing property values to set.

 required

Raises:

Type Description
AttributeError

If settings is not a dictionary.
Source code in src/cuemsutils/helpers.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def setter(self, settings: dict):
    """Set the object properties from a dictionary.

    Args:
        settings (dict): Dictionary containing property values to set.

    Raises:
        AttributeError: If settings is not a dictionary.
    """
    if not isinstance(settings, dict):
        raise AttributeError(f"Invalid type {type(settings)}. Expected dict.")
    for k, v in settings.items():
        try:
            x = getattr(self, f"set_{k}")
            x(v)
        except AttributeError:
            pass

build_xml_dict(x, parent)

Build an xml element from a dictionary

Source code in src/cuemsutils/helpers.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def build_xml_dict(x, parent: Element) -> None:
    """Build an xml element from a dictionary"""
    if not isinstance(x, dict):
        raise AttributeError(f"Invalid type {type(x)}. Expected dict.")
    if not isinstance(parent, Element):
        raise AttributeError(f"Invalid type {type(parent)}. Expected ElementTree.")
    for k, v in x.items():
        if isinstance(v, list):
            for item in v:
                if hasattr(item, 'build'):
                    item.build(parent)
                else:
                    SubElement(parent, k).text = str(item)
        elif hasattr(v, 'build'):
            s = SubElement(parent, k)
            v.build(s)
        else:
            SubElement(parent, k).text = str(v)

check_path(x, dir_only=False)

Check if a path is valid. Raise an error if not.

Source code in src/cuemsutils/helpers.py 
68
69
70
71
72
73
74
75
76
77
78
def check_path(x: str, dir_only: bool = False) -> bool:
    """Check if a path is valid. Raise an error if not."""
    x = path.realpath(x)
    if dir_only:
        dir_ok = _check_dir(path.dirname(x))
        return dir_ok
    if not path.exists(x):
        raise FileNotFoundError(f"Path {x} does not exist")
    if not _readable(x) or not _writable(x):
        raise PermissionError(f"Path {x} is not readable or writable")
    return True

ensure_items(x, requiered)

Ensure that all the items are present in a dictionary

Parameters:

Name Type Description Default
x dict

The dictionary to check

 required
requiered dict

The items (key-value pairs) to check for

 required
Source code in src/cuemsutils/helpers.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def ensure_items(x: dict, requiered: dict) -> dict:
    """Ensure that all the items are present in a dictionary

    Args:
        x (dict): The dictionary to check
        requiered (dict): The items (key-value pairs) to check for

    """
    for k,v in requiered.items():
        if k not in x.keys():
            if v == None:
                x[k] = None
            elif callable(v):
                x[k] = v()
            else:
                x[k] = v

    ## Order the dictionary
    x = {k: x[k] for k in sorted(x.keys())}

    return x

extract_items(x, keys)

Extract list of keys and values from a dictionary

Parameters:

Name Type Description Default
x items

The dictionary items to extract from

 required
keys list

The keys to extract

 required
Source code in src/cuemsutils/helpers.py 
102
103
104
105
106
107
108
109
110
def extract_items(x, keys: list[str] | KeysView[str]) -> ItemsView[str, Any]:
    """Extract list of keys and values from a dictionary

    Args:
        x (items): The dictionary items to extract from
        keys (list): The keys to extract
    """
    d = dict(x)
    return {k: d[k] for k in keys}.items()

mkdir_recursive(folder)

Creates a directory recursively.

Parameters:

Name Type Description Default
folder str

The folder to be created.

 required
Source code in src/cuemsutils/helpers.py 
132
133
134
135
136
137
138
139
140
141
142
143
def mkdir_recursive(folder: str) -> None:
    """
    Creates a directory recursively.

    Args:
        folder (str): The folder to be created.
    """
    if path.exists(folder):
        return
    if not path.exists(path.dirname(folder)):
        mkdir_recursive(path.dirname(folder))
    mkdir(folder)

new_datetime()

Generate a new datetime string.

Source code in src/cuemsutils/helpers.py 
145
146
147
def new_datetime():
    """Generate a new datetime string."""
    return datetime.now().strftime(DATETIME_FORMAT)

new_uuid()

Generate a new Uuid class instance.

Source code in src/cuemsutils/helpers.py 
149
150
151
def new_uuid():
    """Generate a new Uuid class instance."""
    return Uuid()

strtobool(val)

Convert a string value representation of truth to true (1) or false (0).

True values are y, yes, t, true, on and 1. False values are n, no, f, false, off and 0. Raises ValueError if val is anything else.

Source code in src/cuemsutils/helpers.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
def strtobool(val: str) -> bool:
    """Convert a string value representation of truth to true (1) or false (0).

        True values are y, yes, t, true, on and 1.
        False values are n, no, f, false, off and 0.
        Raises ValueError if val is anything else.
    """ 
    if val.lower() in ['y', 'yes', 't', 'true', 'on', '1']:
        return True
    elif val.lower() in ['n', 'no', 'f', 'false', 'off', '0']:
        return False
    else:
        raise ValueError(f'Invalid truth value {val}')

unique_values_to_list(x)

Convert a dictionary to a sorted list of its unique values.

Parameters:

Name Type Description Default
x dict

The dictionary to convert

 required
Source code in src/cuemsutils/helpers.py 
167
168
169
170
171
172
173
def unique_values_to_list(x: dict) -> list:
    """Convert a dictionary to a sorted list of its unique values.

    Args:
        x (dict): The dictionary to convert
    """
    return sorted(list(set(x.values())))

CuemsLoggerAdapter

Bases: LoggerAdapter

Custom LoggerAdapter that properly merges extra dictionaries.

Source code in src/cuemsutils/log.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class CuemsLoggerAdapter(LoggerAdapter):
    """Custom LoggerAdapter that properly merges extra dictionaries."""

    def process(self, msg, kwargs):
        """
        Process the logging call to merge extra dictionaries.
        Ensures that both adapter-level and call-level extra dicts are merged.
        """
        # Start with a copy of the adapter's extra dict (with default caller='')
        extra = {'caller': ''}
        extra.update(self.extra)

        # Merge in any extra dict from the logging call
        if 'extra' in kwargs:
            extra.update(kwargs['extra'])

        kwargs['extra'] = extra
        return msg, kwargs

process(msg, kwargs)

Process the logging call to merge extra dictionaries. Ensures that both adapter-level and call-level extra dicts are merged.

Source code in src/cuemsutils/log.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def process(self, msg, kwargs):
    """
    Process the logging call to merge extra dictionaries.
    Ensures that both adapter-level and call-level extra dicts are merged.
    """
    # Start with a copy of the adapter's extra dict (with default caller='')
    extra = {'caller': ''}
    extra.update(self.extra)

    # Merge in any extra dict from the logging call
    if 'extra' in kwargs:
        extra.update(kwargs['extra'])

    kwargs['extra'] = extra
    return msg, kwargs

Logger

A class for logging messages with different log levels.

This class provides static methods for logging messages with different log levels. It dynamically detects the calling module to use the appropriate logger.

Source code in src/cuemsutils/log.py 
 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
130
131
132
133
class Logger:
    """
    A class for logging messages with different log levels.

    This class provides static methods for logging messages with different log levels.
    It dynamically detects the calling module to use the appropriate logger.
    """

    @staticmethod
    def _get_caller_module():
        """
        Get the module name of the caller by inspecting the call stack.
        """
        frame = inspect.currentframe()
        try:
            # Go up the stack: _get_caller_module -> log/debug/info/etc -> actual caller
            caller_frame = frame.f_back.f_back.f_back
            module_name = caller_frame.f_globals.get('__name__', __name__)
            return module_name
        finally:
            del frame

    @staticmethod
    def log(level, message, **kwargs):
        module_name = Logger._get_caller_module()
        logger = main_logger(module_name=module_name)
        logger.log(level, message, stacklevel = 4, **kwargs)

    @staticmethod
    def debug(message, **kwargs):
        Logger.log(DEBUG, message, **kwargs)

    @staticmethod
    def info(message, **kwargs):
        Logger.log(INFO, message, **kwargs)

    @staticmethod
    def error(message, **kwargs):
        Logger.log(ERROR, message, **kwargs)

    @staticmethod
    def exception(message, **kwargs):
        Logger.log(ERROR, message, **kwargs)

    @staticmethod
    def warning(message, **kwargs):
        Logger.log(WARNING, message, **kwargs)

    @staticmethod
    def critical(message, **kwargs):
        Logger.log(CRITICAL, message, **kwargs)

log_level_to_obj(log_level)

Convert a log level string to a logging level object.

Source code in src/cuemsutils/log.py 
13
14
15
16
17
18
19
20
21
22
23
def log_level_to_obj(log_level):
    """
    Convert a log level string to a logging level object.
    """
    return {
        'DEBUG': DEBUG,
        'INFO': INFO,
        'WARNING': WARNING,
        'ERROR': ERROR,
        'CRITICAL': CRITICAL
    }[log_level]

logged(func)

A decorator function to log information about function calls and their results.

Source code in src/cuemsutils/log.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def logged(func):
    """
    A decorator function to log information about function calls and their results.
    """
    # Get logger for the function's module
    func_logger = main_logger(module_name=func.__module__)

    @wraps(func)
    def wrapper(*args, **kwargs):
        """
        The wrapper function that logs function calls and their results.
        """
        # Only set caller field (the decorated function name)
        # funcName is automatically set by logging to the actual calling function (wrapper)
        d = {"caller": func.__name__}
        func_logger.debug(f"Call recieved", extra = d)
        func_logger.debug(f"Using args: {args} and kwargs: {kwargs}", extra = d)
        try:
            result = func(*args, **kwargs)
            func_logger.debug(f"Finished with result: {result}", extra = d)
        except Warning as w:
            func_logger.warning(f"Warning occurred: {w}", extra = d)
            return result
        except Exception as e:
            func_logger.error(f"Error occurred: {e}", extra = d)
            raise

        else:
            return result

    return wrapper

main_logger(module_name=None, with_syslog=True, with_stdout=True)

Create a root logger with a custom formatter.

Parameters:

Name Type Description Default
module_name

Name of the module to create logger for. Defaults to name if None.

 None
with_syslog

Whether to add syslog handler.

 True
with_stdout

Whether to add stdout handler.

 True
Source code in src/cuemsutils/log.py 
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
def main_logger(module_name = None, with_syslog = True, with_stdout = True):
    """
    Create a root logger with a custom formatter.

    Args:
        module_name: Name of the module to create logger for. Defaults to __name__ if None.
        with_syslog: Whether to add syslog handler.
        with_stdout: Whether to add stdout handler.
    """
    if module_name is None:
        module_name = __name__

    # Return cached logger if it exists
    if module_name in _logger_cache:
        return _logger_cache[module_name]

    logger = getLogger(module_name)
    try:
        log_level = log_level_to_obj(environ['CUEMS_LOG_LEVEL'].upper())
    except KeyError:
        log_level = DEBUG
    logger.setLevel(log_level)

    if with_stdout:
        sh = StreamHandler(sys.stdout)
        sh.setFormatter(cuemsFormatter)
        logger.addHandler(sh)

    if with_syslog:
        syslog_handler = SysLogHandler(
            address = '/dev/log', facility = 'local0'
        )
        syslog_handler.setFormatter(cuemsFormatter)
        logger.addHandler(syslog_handler)

    logger_adapter = CuemsLoggerAdapter(logger, {})
    _logger_cache[module_name] = logger_adapter
    return logger_adapter

Timeoutloop

universal for time-out loop

Source code in src/cuemsutils/timeoutloop.py 
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
class Timeoutloop():
    """ universal for time-out loop """
    def __init__(self, timeout, interval=None):
        self.timeout = timeout
        self.delay = interval
    def __iter__(self):
        self.start_time = time()
        return self
    def __next__(self):
        if self.delay is not None:
            sleep(self.delay)
        now = time()
        time_passed = now - self.start_time
        if time_passed > self.timeout: 
            raise TimeoutError(f"Timeout after {self.timeout} seconds")
        else:
            return time_passed