civet.bases

Base classes for file types.

View Source

 1"""
 2Base classes for file types.
 3"""
 4import abc
 5from os import PathLike
 6from typing import Sequence, TypeVar, Generic, Callable
 7from dataclasses import dataclass
 8from civet.abstract_data import AbstractDataCommand
 9from civet.memoization import Session
10from civet.shells import Shell, subprocess_run
11
12
13@dataclass(frozen=True)
14class DataSource(AbstractDataCommand, abc.ABC):
15    """
16    A `DataSource` provides the `DataSource.save` method to `civet.abstract_data.AbstractDataCommand`,
17    which is an alias for typical usage of `civet.memoization.Session`.
18    """
19    def save(self, output: str | PathLike,
20             require_output: bool = True,
21             shell: Shell = subprocess_run) -> None:
22        r"""
23        Save the result of this command to the given output path.
24
25        The behavior of `save` can be customized by specifying `shell` as different
26        wrappers around `subprocess.run`. For example, to save the logs of subprocesses:
27
28        ```python
29        from pathlib import Path
30        import subprocess as sp
31
32        log_path = Path('output.log')
33        with log_path.open('wb') as log_file:
34            def saves_log_shell(cmd):
35                log_file.write(b'Running: ')
36                log_file.write(str(cmd).encode('utf-8'))
37                log_file.write(b'\n')
38                log_file.flush()
39                sp.run(cmd, stdout=log_file, stderr=sp.STDOUT, check=True)
40            GenericSurface('input.obj').slide_left().save('lefter.obj', shell=saves_log_shell)
41        ```
42        """
43        with Session(require_output, shell) as s:
44            s.save(self, output)
45
46
47_D = TypeVar('_D', bound='DataFile')
48
49
50@dataclass(frozen=True)
51class DataFile(DataSource, Generic[_D], abc.ABC):
52    """
53    A `DataFile` represents a file type. It can wrap input files of that file type,
54    or commands which produce results of that file type.
55    """
56
57    input: str | PathLike | AbstractDataCommand
58
59    def command(self, output: str | PathLike) -> Sequence[str | PathLike | AbstractDataCommand]:
60        return 'cp', self.input, output
61
62    def create_command(self, command: Callable[[str | PathLike], Sequence[str | PathLike | AbstractDataCommand]]
63                       ) -> _D:
64        """
65        Chain a command which produces output of the same type as this `DataFile`.
66        """
67        class Intermediate(self.__class__):
68            def command(self, output: str | PathLike) -> Sequence[str | PathLike | AbstractDataCommand]:
69                return command(output)
70        return Intermediate(self)

@dataclass(frozen=True)

class DataSource(civet.abstract_data.AbstractDataCommand, abc.ABC): View Source

14@dataclass(frozen=True)
15class DataSource(AbstractDataCommand, abc.ABC):
16    """
17    A `DataSource` provides the `DataSource.save` method to `civet.abstract_data.AbstractDataCommand`,
18    which is an alias for typical usage of `civet.memoization.Session`.
19    """
20    def save(self, output: str | PathLike,
21             require_output: bool = True,
22             shell: Shell = subprocess_run) -> None:
23        r"""
24        Save the result of this command to the given output path.
25
26        The behavior of `save` can be customized by specifying `shell` as different
27        wrappers around `subprocess.run`. For example, to save the logs of subprocesses:
28
29        ```python
30        from pathlib import Path
31        import subprocess as sp
32
33        log_path = Path('output.log')
34        with log_path.open('wb') as log_file:
35            def saves_log_shell(cmd):
36                log_file.write(b'Running: ')
37                log_file.write(str(cmd).encode('utf-8'))
38                log_file.write(b'\n')
39                log_file.flush()
40                sp.run(cmd, stdout=log_file, stderr=sp.STDOUT, check=True)
41            GenericSurface('input.obj').slide_left().save('lefter.obj', shell=saves_log_shell)
42        ```
43        """
44        with Session(require_output, shell) as s:
45            s.save(self, output)

A DataSource provides the DataSource.save method to civet.abstract_data.AbstractDataCommand, which is an alias for typical usage of civet.memoization.Session.

def save( self, output: str | os.PathLike, require_output: bool = True, shell: Callable[[Sequence[Union[str, os.PathLike]]], NoneType] = <function subprocess_run>) -> None: View Source

20    def save(self, output: str | PathLike,
21             require_output: bool = True,
22             shell: Shell = subprocess_run) -> None:
23        r"""
24        Save the result of this command to the given output path.
25
26        The behavior of `save` can be customized by specifying `shell` as different
27        wrappers around `subprocess.run`. For example, to save the logs of subprocesses:
28
29        ```python
30        from pathlib import Path
31        import subprocess as sp
32
33        log_path = Path('output.log')
34        with log_path.open('wb') as log_file:
35            def saves_log_shell(cmd):
36                log_file.write(b'Running: ')
37                log_file.write(str(cmd).encode('utf-8'))
38                log_file.write(b'\n')
39                log_file.flush()
40                sp.run(cmd, stdout=log_file, stderr=sp.STDOUT, check=True)
41            GenericSurface('input.obj').slide_left().save('lefter.obj', shell=saves_log_shell)
42        ```
43        """
44        with Session(require_output, shell) as s:
45            s.save(self, output)

Save the result of this command to the given output path.

The behavior of save can be customized by specifying shell as different wrappers around subprocess.run. For example, to save the logs of subprocesses:

from pathlib import Path
import subprocess as sp

log_path = Path('output.log')
with log_path.open('wb') as log_file:
    def saves_log_shell(cmd):
        log_file.write(b'Running: ')
        log_file.write(str(cmd).encode('utf-8'))
        log_file.write(b'\n')
        log_file.flush()
        sp.run(cmd, stdout=log_file, stderr=sp.STDOUT, check=True)
    GenericSurface('input.obj').slide_left().save('lefter.obj', shell=saves_log_shell)

Inherited Members

civet.abstract_data.AbstractDataCommand: command; preferred_suffix

@dataclass(frozen=True)

class DataFile(DataSource, typing.Generic[~_D], abc.ABC): View Source

51@dataclass(frozen=True)
52class DataFile(DataSource, Generic[_D], abc.ABC):
53    """
54    A `DataFile` represents a file type. It can wrap input files of that file type,
55    or commands which produce results of that file type.
56    """
57
58    input: str | PathLike | AbstractDataCommand
59
60    def command(self, output: str | PathLike) -> Sequence[str | PathLike | AbstractDataCommand]:
61        return 'cp', self.input, output
62
63    def create_command(self, command: Callable[[str | PathLike], Sequence[str | PathLike | AbstractDataCommand]]
64                       ) -> _D:
65        """
66        Chain a command which produces output of the same type as this `DataFile`.
67        """
68        class Intermediate(self.__class__):
69            def command(self, output: str | PathLike) -> Sequence[str | PathLike | AbstractDataCommand]:
70                return command(output)
71        return Intermediate(self)

A DataFile represents a file type. It can wrap input files of that file type, or commands which produce results of that file type.

DataFile(input: str | os.PathLike | civet.abstract_data.AbstractDataCommand)

def command( self, output: str | os.PathLike) -> Sequence[str | os.PathLike | civet.abstract_data.AbstractDataCommand]: View Source

60    def command(self, output: str | PathLike) -> Sequence[str | PathLike | AbstractDataCommand]:
61        return 'cp', self.input, output

def create_command( self, command: Callable[[str | os.PathLike], Sequence[str | os.PathLike | civet.abstract_data.AbstractDataCommand]]) -> ~_D: View Source

63    def create_command(self, command: Callable[[str | PathLike], Sequence[str | PathLike | AbstractDataCommand]]
64                       ) -> _D:
65        """
66        Chain a command which produces output of the same type as this `DataFile`.
67        """
68        class Intermediate(self.__class__):
69            def command(self, output: str | PathLike) -> Sequence[str | PathLike | AbstractDataCommand]:
70                return command(output)
71        return Intermediate(self)

Chain a command which produces output of the same type as this DataFile.

Inherited Members

DataSource: save
civet.abstract_data.AbstractDataCommand: preferred_suffix