brane.core package
Submodules
brane.core.base module
- class brane.core.base.BaseSubclassRegister
Bases:
object- valid: bool = True
- priority: int = -1
- classmethod get_registered_subclasses() dict[str, type]
- class brane.core.base.MetaFalse(classname: str, bases: tuple[type], class_info: dict)
Bases:
type
- class brane.core.base.Context
Bases:
dictThe context/state infomation in the IO flows.
- object
- Type
Optional[Any]
- objects
- Type
Union[None, Any, list[Any], dict[str, Any]]
- path
- Type
Optional[Union[str, pathlib.Path]]
- paths
- Type
Union[None, list[Union[str, pathlib.Path]], dict[str, Union[str, pathlib.Path]]]
- protocol
- Type
Optional[str]
- file
- Type
Optional[_io._IOBase]
- files
- Type
Union[None, list[_io._IOBase], dict[str, _io._IOBase]]
- args
- Type
tuple
- kwargs
- Type
dict[str, Any]
- Module
- Type
Optional[brane.typing.ModuleClassType]
- Format
- Type
Optional[brane.typing.FormatClassType]
- object: Optional[Any] = None
- objects: Union[None, Any, list[Any], dict[str, Any]] = None
- path: Optional[Union[str, Path]] = None
- paths: Union[None, list[Union[str, pathlib.Path]], dict[str, Union[str, pathlib.Path]]] = None
- protocol: Optional[str] = None
- file: Optional[_IOBase] = None
- files: Union[None, list[_io._IOBase], dict[str, _io._IOBase]] = None
- args: tuple = ()
- kwargs: dict[str, Any] = {}
- Module: Optional[ModuleClassType] = None
- Format: Optional[FormatClassType] = None
- ext: Optional[str]
brane.core.event module
- brane.core.event.convert_obj_into_list(obj: Union[list[T], tuple[T], set[T], T]) list[T]
- class brane.core.event.MarkerRule(marker: 'str', action: "Literal['allowed', 'denied']", priority: 'int')
Bases:
object- marker: str
- action: Literal['allowed', 'denied']
- priority: int
- class brane.core.event.Event(event_name: str = '', hook_funcs: Optional[Union[HookClassType, list[brane.typing.HookClassType], tuple[brane.typing.HookClassType], set[brane.typing.HookClassType]]] = None)
Bases:
EventClassType- get_hook_names(exclude_none: bool = True) list[Optional[str]]
- search_hook_name_with_index(target_name: str, exact: bool = False) tuple[int, str]
- add_hooks(hook_funcs: Union[HookClassType, list[brane.typing.HookClassType], tuple[brane.typing.HookClassType], set[brane.typing.HookClassType]], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after')
- remove_hooks(hook_names: Union[str, Container[str]], strict: bool = False)
- clear_hooks()
- add_marker_rule(name: Union[str, int], marker: str, action: Literal['allowed', 'denied'], priority: Optional[int] = None)
- remove_marker_rule(name: Union[str, int]) MarkerRule
- clean_marker_rules()
- check_marker(marker: Optional[Union[str, set[str]]]) bool
- fire(info: ContextInterface, verbose: bool = False) ContextInterface
brane.core.factory module
- class brane.core.factory.ModuleConfigType
Bases:
dict[str,dict[str,Any]]
- class brane.core.factory.FormatConfigType
Bases:
dict[str,dict[str,Any]]
- class brane.core.factory.ObjectConfigType
Bases:
dict[str,dict[str,Any]]
- class brane.core.factory.HookConfigType
Bases:
dict[str,Union[str,dict]]
- brane.core.factory.load_config(config_path: Union[str, Path], strict: bool = False) dict[str, Union[str, dict]]
- brane.core.factory.load_multiple_config(config_path_list: list[Union[str, pathlib.Path]], strict: bool = False) list[dict[str, Union[str, dict]]]
- class brane.core.factory.BraneClassGenerator
Bases:
object- className2Module: dict[str, brane.typing.ModuleClassType] = {'CV2': <class 'brane.core.module.CV2Module'>, 'Csv': <class 'brane.core.module.CsvModule'>, 'Json': <class 'brane.core.module.JsonModule'>, 'Numpy': <class 'brane.core.module.NumpyModule'>, 'Numpz1': <class 'brane.core.module.Numpz1Module'>, 'Numpz2': <class 'brane.core.module.Numpz2Module'>, 'OpenPyxl': <class 'brane.core.module.OpenPyxlModule'>, 'Pandas': <class 'brane.core.module.PandasModule'>, 'Pickle': <class 'brane.core.module.PickleModule'>, 'Pillow': <class 'brane.core.module.PillowModule'>, 'PyYAML': <class 'brane.core.module.PyYAMLModule'>, 'TextIO': <class 'brane.core.module.TextIOModule'>, 'Toml': <class 'brane.core.module.TomlModule'>, 'Wave': <class 'brane.core.module.WaveModule'>}
- className2Format: dict[str, brane.typing.FormatClassType] = {'CSV': <class 'brane.core.format.CSVFormat'>, 'CSV_builtin': <class 'brane.core.format.CSV_builtinFormat'>, 'JPEG': <class 'brane.core.format.JPEGFormat'>, 'JSON': <class 'brane.core.format.JSONFormat'>, 'NPY': <class 'brane.core.format.NPYFormat'>, 'NPZ': <class 'brane.core.format.NPZFormat'>, 'PKL': <class 'brane.core.format.PKLFormat'>, 'PNG': <class 'brane.core.format.PNGFormat'>, 'TOML': <class 'brane.core.format.TOMLFormat'>, 'TSV': <class 'brane.core.format.TSVFormat'>, 'TXT': <class 'brane.core.format.TXTFormat'>, 'WAV': <class 'brane.core.format.WAVFormat'>, 'XLSX': <class 'brane.core.format.XLSXFormat'>, 'YAML': <class 'brane.core.format.YAMLFormat'>}
- className2Object: dict[str, brane.typing.ObjectClassType] = {'Csv_CSV': <class 'brane.core.object.Csv_CSV_Object'>, 'Json_JSON': <class 'brane.core.object.Json_JSON_Object'>, 'Numpy_NdArray': <class 'brane.core.object.Numpy_NdArray_Object'>, 'PIL_Image': <class 'brane.core.object.PIL_Image_Object'>, 'PIL_JPEG': <class 'brane.core.object.PIL_JPEG_Object'>, 'PIL_PNG': <class 'brane.core.object.PIL_PNG_Object'>, 'Pandas_DataFrame': <class 'brane.core.object.Pandas_DataFrame_Object'>, 'Pickle_PKL': <class 'brane.core.object.Pickle_PKL_Object'>, 'Pyyaml_Dict': <class 'brane.core.object.Pyyaml_Dict_Object'>, 'TextIO_TXT': <class 'brane.core.object.TextIO_TXT_Object'>, 'Toml_Dict': <class 'brane.core.object.Toml_Dict_Object'>}
- static generate_classes_from_configs(config_list: list[dict[str, Union[str, dict]]], suffix: str, cls: T, apply_attributes: Optional[Callable[[str, ClassAttributeType], ClassAttributeType]] = None) dict[str, T]
- classmethod load_brane_modules(config_path_list: list[Union[str, pathlib.Path]]) dict[str, brane.typing.ModuleClassType]
- classmethod load_brane_formats(config_path_list: list[Union[str, pathlib.Path]]) dict[str, brane.typing.FormatClassType]
- classmethod load_brane_objects(config_path_list: list[Union[str, pathlib.Path]]) dict[str, brane.typing.ObjectClassType]
- classmethod setup(module_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, format_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, object_config_paths: Optional[list[Union[str, pathlib.Path]]] = None)
- classmethod activate(module_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, format_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, object_config_paths: Optional[list[Union[str, pathlib.Path]]] = None)
- class brane.core.factory.BraneHooksGenerator
Bases:
object- event2hooks: dict[str, list] = {'pre_read': [<function check_path_existence>], 'pre_write': [<function create_parent_directory>]}
- classmethod load_hooks_from_config(config_path: Union[str, Path]) dict[str, list[Union[Callable, brane.typing.HookClassType]]]
- classmethod load_event2hooks(config_paths: Optional[list[Union[str, pathlib.Path]]] = None) dict[str, list[Union[Callable, brane.typing.HookClassType]]]
- classmethod setup(config_paths: Optional[list[Union[str, pathlib.Path]]] = None, use_builtin_config: bool = True)
Overwrite the base hook set based on the config.
- classmethod activate(config_paths: Optional[list[Union[str, pathlib.Path]]])
brane.core.format module
- brane.core.format.normalize_extension_default(ext: str) str
- class brane.core.format.MetaFormat(classname: str, bases: tuple[type], class_info: dict)
Bases:
type- property registered_formats: dict[str, brane.typing.FormatClassType]
- class brane.core.format.Format
Bases:
FormatClassType,BaseSubclassRegister- priority: int = 50
- module: Optional[ModuleClassType] = None
- default_extension: Optional[str] = None
- variation: list[str] = [None]
- classmethod check_extension(ext: str) bool
- name = None
- class brane.core.format.FormatConfig
Bases:
object- name: Optional[str] = None
- priority: int = 50
- module_name: Optional[str] = None
- default_extension: Optional[str] = None
- variation: list[str] = []
- class brane.core.format.FormatTemplate
Bases:
Format,FormatConfig- name: Optional[str] = None
- variation: list[str] = [None]
- class brane.core.format.MetaNoneFormat(classname: str, bases: tuple[type], class_info: dict)
Bases:
MetaFormat,MetaFalse
- class brane.core.format.NoneFormat
Bases:
FormatClassType- valid = False
- name = None
- variation = [None]
- module: Optional[ModuleClassType]
- check_extension: Callable[[str], bool]
brane.core.hook module
- brane.core.hook.generate_hash_from_objects(*immutable_objects) str
- class brane.core.hook.Hook(name: Optional[str] = None, marker: Optional[Union[str, set[str]]] = None)
Bases:
HookClassType- hook_name: Optional[str] = None
- marker: Optional[Union[str, set[str]]] = None
- condition(info: ContextInterface) bool
- property active: bool
- activate()
- deactivate()
- class brane.core.hook.FunctionHook(func: ~typing.Callable[[~brane.typing.ContextInterface], ~brane.typing.ContextInterface], condition_func: ~typing.Callable[[~brane.typing.ContextInterface], bool] = <function FunctionHook.<lambda>>, name: ~typing.Optional[str] = None, marker: ~typing.Optional[~typing.Union[str, set[str]]] = None, **kwargs_exp)
Bases:
Hook- Parameters
func – hook function. It recevices context including path/paths, object/objects and return
condition_func – evaluate whether this hook is called or not based on the context.
name –
marker –
aruguments (# experimental keyword) –
skip_when_error –
object_type (type) –
container_type (None|'list'|'tuple'|'dict') –
is_multiple_objects (bool) –
- condition(info: ContextInterface) bool
brane.core.mapper module
- class brane.core.mapper.ExtensionMapper
Bases:
object- classmethod get_format_class_from_extension(ext: str) FormatClassType
- classmethod get_module_class_from_extension(ext: str) ModuleClassType
- class brane.core.mapper.ObjectFormat2Module
Bases:
object- get_module_from_object(fmt=None) ModuleClassType
brane.core.module module
- class brane.core.module.MetaModule(classname: str, bases: tuple[type], class_info: dict)
Bases:
type- property registered_modules: dict[str, brane.typing.ModuleClassType]
- class brane.core.module.Module
Bases:
ModuleClassType,BaseSubclassRegister- priority: int = 50
- name: Optional[str] = None
- loaded: bool = False
- classmethod load_modules()
lazy loading of modules and setting their attributes or methods to this class.
- classmethod reload_modules()
reload modules or re-setting attributes or methods.
- classmethod read(path: ~typing.Optional[~typing.Union[str, ~pathlib.Path]] = None, file: ~typing.Optional[~_io._IOBase] = None, io: ~typing.Optional[~typing.Union[~brane.typing._generate_file_related_types.<locals>.IO, ~fsspec.asyn.AsyncFileSystem]] = None, *args, **kwargs) Any
read from path or file stream.
- Parameters
path – path object
file – file stream object
- Returns
loaded objects
- Return type
Any
- classmethod write(obj: ~typing.Any, path: ~typing.Optional[~typing.Union[str, ~pathlib.Path]] = None, file: ~typing.Optional[~_io._IOBase] = None, io: ~typing.Optional[~typing.Union[~brane.typing._generate_file_related_types.<locals>.IO, ~fsspec.asyn.AsyncFileSystem]] = None, *args, **kwargs) Any
write objects to the file specified by path object or file object.
- Parameters
obj – object to save
path – path object
file – file stream object
- class brane.core.module.ModuleConfig
Bases:
object- name: Optional[str] = None
- priority: int = 50
- file_open_for_read: bool = False
- open_mode_for_read: dict[str, Any] = {'mode': 'r'}
- module_read_method_name: Optional[str] = None
- file_read_method_name: Optional[str] = None
- base_args_read: tuple = ()
- base_kwargs_read: dict = {}
- transform_name: Optional[str] = None
- transform_info: Optional[tuple[str, ...]] = None
- object_transform_method_name: Optional[str] = None
- file_open_for_write: bool = False
- open_mode_for_write: dict[str, Any] = {'mode': 'w'}
- module_write_method_name: Optional[str] = None
- writer_method_name: Optional[str] = None
- object_unpacking_type: Optional[Literal['sequence', 'mapping']] = None
- file_arg_first: bool = True
- file_keyword_at_write: Optional[str] = None
- obj_keyword_at_write: Optional[str] = None
- object_write_method_name: Optional[str] = None
- file_write_method_name: Optional[str] = None
- base_args_write: tuple = ()
- base_kwargs_write: dict = {}
- class brane.core.module.ModuleTemplate
Bases:
Module,ModuleConfigNeeded to be inheritated. The child class correspond to module.
- Parameters
name –
loaded –
reload_modules –
read –
write –
module_name –
module –
module_read_method_name –
module_read_method –
file_read_method_name –
file_open_for_read –
open_mode_for_read –
transform –
transform_name –
transform_info –
object_transform_method_name –
module_write_method_name –
module_write_method –
writer_method_name –
object_write_method_name –
file_write_method_name –
file_open_for_write –
open_mode_for_write –
file_arg_first –
object_unpacking_type –
file_keyword_at_write –
obj_keyword_at_write –
base_args_read –
base_kwargs_read –
base_args_write –
base_kwargs_write –
Note
Argument version v0.0
- module: Any = None
- module_name: Optional[str] = None
- module_read_method: Optional[Callable] = None
- transform: Optional[Callable] = None
- module_write_method: Optional[Callable] = None
- generate_params_write = None
- classmethod load_modules()
lazy loading of modules and setting their attributes or methods to this class.
- classmethod read(path: ~typing.Optional[~typing.Union[str, ~pathlib.Path]] = None, file: ~typing.Optional[~_io._IOBase] = None, io: ~typing.Optional[~typing.Union[~brane.typing._generate_file_related_types.<locals>.IO, ~fsspec.asyn.AsyncFileSystem]] = None, *args, **kwargs) Any
read from path or file stream.
- Parameters
path – path object
file – file stream object
- Returns
loaded objects
- Return type
Any
- classmethod write(obj: ~typing.Any, path: ~typing.Optional[~typing.Union[str, ~pathlib.Path]] = None, file: ~typing.Optional[~_io._IOBase] = None, io: ~typing.Optional[~typing.Union[~brane.typing._generate_file_related_types.<locals>.IO, ~fsspec.asyn.AsyncFileSystem]] = None, *args, **kwargs) Any
write objects to the file specified by path object or file object.
- Parameters
obj – object to save
path – path object
file – file stream object
- name: Optional[str] = None
- class brane.core.module.MetaNoneModule(classname: str, bases: tuple[type], class_info: dict)
Bases:
MetaModule,MetaFalse
brane.core.object module
- class brane.core.object.MetaObject(classname: str, bases, class_info: dict[str, Any])
Bases:
type- property object: Any
- property registered_objects: dict[str, brane.typing.ObjectClassType]
- class brane.core.object.Object
Bases:
ObjectClassType,BaseSubclassRegister- name: Optional[str] = None
- priority: int = 50
- loaded: bool = False
- format: Optional[FormatClassType] = None
- module: Optional[ModuleClassType] = None
- object_type = None
- object_type_info: Optional[tuple[str, ...]] = None
- type_evaluation = None
- format_checker: Optional[Callable[[Any], FormatClassType]] = None
- module_checker: Optional[Callable[[Any, FormatClassType], ModuleClassType]] = None
- classmethod load_objects()
brane.core.utils module
- brane.core.utils.integrate_args(base_args: tuple, new_args: tuple) tuple
- brane.core.utils.integrate_kwargs(base_kwargs: dict, new_kwargs: dict) dict
- brane.core.utils.sort_mapper(mapper: dict[str, T], key: str, ascending: bool = False) dict[str, T]
- Parameters
mapper –
key –
ascending –
- brane.core.utils.get_extension_from_filname_default(path: Union[str, Path]) str
brane.core.xio module
- class brane.core.xio.HookManager
Bases:
object- pre_read = 1. 3cd5ff3f11893ad8: <function check_path_existence>
- post_read
- pre_write = 1. a08a3abc8f88e692: <function create_parent_directory>
- post_write
- pre_readall
- post_readall
- pre_writeall
- post_writeall
- classmethod get_events() dict[str, brane.core.event.Event]
Get events managing hooks.
- classmethod show_events()
Show all registered hooks at each event.
- classmethod connect_hook_and_event(event: EventClassType, hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
- classmethod register_pre_read_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs before loading.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_post_read_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs after loading.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_pre_write_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs before saving.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_post_write_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs after saving.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_pre_readall_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs before loading all.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_post_readall_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs after loading all.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_pre_writeall_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs before saving all.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- classmethod register_post_writeall_hook(hook: Union[Hook, Callable], ref_index: Optional[int] = None, ref_name: Optional[str] = None, loc: Literal['before', 'after'] = 'after', **hook_kwargs)
Add new hook which runs after saving all.
- Parameters
hook – The hook to register. It should be a function or an instance of the Hook Class.
- static remove_hooks_for_event(event: EventClassType, hook_names: Container[str], *args, **kwargs)
- classmethod remove_hooks(hook_names: Union[str, Container[str]], *args, **kwargs)
- static clear_hooks(event: EventClassType)
- classmethod clear_pre_read_hook()
Remove all the registered hooks which run before loading.
- classmethod clear_post_read_hook()
Remove all the registered hooks which run after loading.
- classmethod clear_pre_write_hook()
Remove all the registered hooks which run before saving.
- classmethod clear_post_write_hook()
Remove all the registered hooks which run after saving.
- classmethod clear_pre_readall_hook()
Remove all the registered hooks which run before loading all.
- classmethod clear_post_readall_hook()
Remove all the registered hooks which run after loading all.
- classmethod clear_pre_writeall_hook()
Remove all the registered hooks which run before saving all.
- classmethod clear_post_writeall_hook()
Remove all the registered hooks which run after saving all.
- classmethod clear_all_hooks()
Remove all the registered hooks at any event.
- classmethod setup_hooks(hook_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, use_builtin_config: bool = True)
- class brane.core.xio.ExtendedIO
Bases:
HookManager- get_extension_from_filename() str
- logger = <brane.core.xio.IOLogger object>
- kept_storage_options = {}
- classmethod set_storage_option(protocol: str, storage_options: dict)
- classmethod read(path: Optional[Union[str, Path]] = None, file: Optional[_IOBase] = None, ext: str = '', module_name: str = '', read_args: Optional[tuple] = None, read_kwargs: Optional[dict] = None, storage_options: dict = {}, *args, **kwargs) Any
- Parameters
path – File path to read. Currently, only local file system path is available.
file – File object such as byte stream to read.
ext – The extension name. Used only when the extension is not given in the path.
module_name – The module name. Specified only in the case the module is fixed.
- Returns
loaded object.
Note
The priority order is module_name > ext > path
- classmethod read_all_as_list(multiple_paths: Union[str, list[Union[str, pathlib.Path]]], read_args: Optional[tuple] = None, read_kwargs: Optional[dict] = None, *args, **kwargs) Union[list[Any], Any]
- Parameters
multiple_paths – Several file paths to read once. The glob format is also allowed. Currently, only local file system paths are available.
read_args –
read_kwargs –
- Returns
loaded objects as list. The ordering is same as given paths. If the specified path is in the glob format, paths will be sorted.
- classmethod read_all_as_dict(multiple_paths: dict[str, Union[str, pathlib.Path]], read_args: Optional[tuple] = None, read_kwargs: Optional[dict] = None, *args, **kwargs) Union[dict[str, Any], Any]
- classmethod write(obj: ~typing.Any, path: ~typing.Union[str, ~pathlib.Path] = <class 'NoneType'>, file: ~typing.Optional[~_io._IOBase] = None, ext: str = '', module_name: str = '', write_args: ~typing.Optional[tuple] = None, write_kwargs: ~typing.Optional[dict] = None, storage_options: dict = {}, *args, **kwargs) Any
- Parameters
obj – Any object to save.
path – File path to write. Currently, only local file system path is available.
file – File object such as byte stream to write.
ext – The extension name. Used only when the extension is not given in the path.
module_name – The module name. Specified only in the case the module is fixed.
- classmethod write_all_from_list(obj_list: list[Any], output_dir: Optional[Union[str, Path]] = None, path_ruler: Optional[Callable[[int], str]] = None, write_args: Optional[tuple] = None, write_kwargs: Optional[dict] = None, *args, **kwargs)
- classmethod write_all_from_dict(obj_dict: dict[str, Any], output_dir: Optional[Union[str, Path]] = None, path_ruler: Optional[Callable[[str], str]] = None, write_args: Optional[tuple] = None, write_kwargs: Optional[dict] = None, *args, **kwargs)
- classmethod reload(module_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, format_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, object_config_paths: Optional[list[Union[str, pathlib.Path]]] = None, hook_config_paths: Optional[list[Union[str, pathlib.Path]]] = None)
- classmethod all_modules()
- classmethod all_formats()
- classmethod all_objects()