File: convenience.py

package info (click to toggle)
python-sigima 1.0.3-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 24,956 kB
  • sloc: python: 33,326; makefile: 3
file content (157 lines) | stat: -rw-r--r-- 4,421 bytes parent folder | download
1
2
3
4
5
6
7
8
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
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
# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.

"""Convenience I/O functions

This module provides convenient wrapper functions for input/output operations with
signals and images. These functions offer a simplified interface to the underlying
I/O system, making common tasks easier to perform.
"""

from __future__ import annotations

import os.path as osp
from typing import Generator, Sequence

import guidata.dataset as gds

from sigima.config import _
from sigima.io.common.basename import format_basenames
from sigima.io.image.base import ImageIORegistry
from sigima.io.signal.base import SignalIORegistry
from sigima.objects import ImageObj, SignalObj, TypeObj


class SaveToDirectoryParam(gds.DataSet):
    """Save to directory parameters."""

    def build_filenames(self, objs: list[TypeObj]) -> list[str]:
        """Build filenames according to current parameters."""
        filenames = format_basenames(objs, self.basename + self.extension)
        used: set[str] = set()  # Ensure all filenames are unique.
        for i, filename in enumerate(filenames):
            root, ext = osp.splitext(filename)
            filepath = osp.join(self.directory, filename)
            k = 1
            while (filename in used) or (not self.overwrite and osp.exists(filepath)):
                filename = f"{root}_{k}{ext}"
                filepath = osp.join(self.directory, filename)
                k += 1
            used.add(filename)
            filenames[i] = filename
        return filenames

    def generate_filepath_obj_pairs(
        self, objs: list[TypeObj]
    ) -> Generator[tuple[str, TypeObj], None, None]:
        """Iterate over (filepath, object) pairs to be saved."""
        for filename, obj in zip(self.build_filenames(objs), objs):
            yield osp.join(self.directory, filename), obj

    directory = gds.DirectoryItem(_("Directory"))
    basename = gds.StringItem(
        _("Basename pattern"),
        default="{title}",
        help=_("""Pattern accepts a Python format string.

Standard Python formatting fields may be used, including:
{title}, {index}, {count}, {xlabel}, {xunit}, {ylabel}, {yunit},
{metadata}, {metadata[key]}"""),
    )
    extension = gds.StringItem(
        _("Extension"),
        help=_("File extension with leading dot (e.g. .txt or .csv)"),
        regexp=r"^\.\w+$",
    )
    overwrite = gds.BoolItem(
        _("Overwrite"), default=False, help=_("Overwrite existing files")
    )


def read_signals(filename: str) -> Sequence[SignalObj]:
    """Read a list of signals from a file.

    Args:
        filename: File name.

    Returns:
        List of signals.
    """
    return SignalIORegistry.read(filename)


def read_signal(filename: str) -> SignalObj:
    """Read a signal from a file.

    Args:
        filename: File name.

    Returns:
        Signal.
    """
    return read_signals(filename)[0]


def write_signal(filename: str, signal: SignalObj) -> None:
    """Write a signal to a file.

    Args:
        filename: File name.
        signal: Signal.
    """
    SignalIORegistry.write(filename, signal)


def write_signals(p: SaveToDirectoryParam, signals: list[SignalObj]) -> None:
    """Write a list of signals to a file.

    Args:
        p: Save to directory parameters.
        signals: List of signals.
    """
    for filepath, signal in p.generate_filepath_obj_pairs(signals):
        SignalIORegistry.write(filepath, signal)


def read_images(filename: str) -> Sequence[ImageObj]:
    """Read a list of images from a file.

    Args:
        filename: File name.

    Returns:
        List of images.
    """
    return ImageIORegistry.read(filename)


def read_image(filename: str) -> ImageObj:
    """Read an image from a file.

    Args:
        filename: File name.

    Returns:
        Image.
    """
    return read_images(filename)[0]


def write_image(filename: str, image: ImageObj) -> None:
    """Write an image to a file.

    Args:
        filename: File name.
        image: Image.
    """
    ImageIORegistry.write(filename, image)


def write_images(p: SaveToDirectoryParam, images: list[ImageObj]) -> None:
    """Write a list of images to files in a directory.

    Args:
        p: Save to directory parameters.
        images: List of images.
    """
    for filepath, image in p.generate_filepath_obj_pairs(images):
        ImageIORegistry.write(filepath, image)