cengal.data_manipulation.conversion.reinterpret_cast_management.manager.versions.v_0.manager
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18 19__all__ = ['AutoDerivedClass', 'BaseAutoDerivedObjWrapper', 'ClassWrappingFactory', 'ClassWrappingFactoryWithObjDataSizeLog'] 20 21 22""" 23Module Docstring 24Docstrings: http://www.python.org/dev/peps/pep-0257/ 25""" 26 27 28__author__ = "ButenkoMS <gtalk@butenkoms.space>" 29__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 30__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 31__license__ = "Apache License, Version 2.0" 32__version__ = "4.4.1" 33__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 34__email__ = "gtalk@butenkoms.space" 35# __status__ = "Prototype" 36__status__ = "Development" 37# __status__ = "Production" 38 39 40from sys import getsizeof 41from cengal.data_generation.id_generator import IDGenerator 42from cengal.data_manipulation.conversion.reinterpret_cast import TemporaryReinterpretCast, reinterpret_cast 43from cengal.entities.copyable import copy__impl, deepcopy__impl 44from typing import Dict, Type, Callable, Any, Tuple, FrozenSet, Set, Optional 45from copy import Error as CopyError, copy, deepcopy 46from socket import SocketIO 47 48 49class AutoDerivedClass: 50 instance_id = IDGenerator() 51 52 def __init__(self): 53 self._instance_id: int = self.instance_id() 54 self.class_name = type(self).__name__ 55 self.derived: Dict[Type, Type] = dict() 56 57 @property 58 def methods(self) -> Dict[str, Callable]: 59 raise NotImplementedError 60 61 def base_classes(self, base_type) -> Tuple[Type]: 62 return (base_type,) 63 64 def __call__(self, base_type: Type): 65 base_type_name = base_type.__name__ 66 try: 67 return self.derived[base_type_name] 68 except KeyError: 69 result: Type = type(f'{self.class_name}__{self._instance_id}__from__{base_type_name}', self.base_classes(base_type), self.methods) 70 self.derived[base_type_name] = result 71 return result 72 73 74class BaseAutoDerivedObjWrapper: 75 instance_id = IDGenerator() 76 77 def __init__(self): 78 self._instance_id: int = self.instance_id() 79 self.class_name = type(self).__name__ 80 self.derived: Dict[Type, Type] = dict() 81 self.t = self.type 82 83 def wrapping_required(self, obj: Any, base_type: Type, fields: Tuple[str], planned_type_name: str) -> bool: 84 return True 85 86 def methods(self, obj: Any, base_type: Type, fields: Tuple[str]) -> Dict[str, Callable]: 87 raise NotImplementedError 88 89 def base_classes(self, obj: Any, base_type: Type, fields: Tuple[str]) -> Tuple[Type]: 90 return (base_type,) 91 92 def gen_fields_tuple(self, obj: Any): 93 try: 94 return obj.__slots__ 95 except AttributeError: 96 pass 97 98 return tuple(obj.__dict__.keys()) 99 100 def type(self, obj: Any): 101 base_type = type(obj) 102 base_type_name = base_type.__name__ 103 try: 104 return self.derived[base_type_name] 105 except KeyError: 106 fields: Tuple = self.gen_fields_tuple(obj) 107 planned_type_name: str = f'{self.class_name}__{self._instance_id}__from__{base_type_name}' 108 if self.wrapping_required(obj, base_type, fields, planned_type_name): 109 result: Type = type(planned_type_name, self.base_classes(obj, base_type, fields), self.methods(obj, base_type, fields)) 110 self.derived[base_type_name] = result 111 return result 112 else: 113 return base_type 114 115 def __call__(self, obj: Any): 116 obj.__class__ = self.type(obj) 117 return obj 118 119 def temporary(self, obj: Any): 120 return TemporaryReinterpretCast(self.type(obj), obj) 121 122 def persistent(self, obj: Any): 123 return reinterpret_cast(self.type(obj), obj) 124 125 126# class ClassWrapper: 127# def __init__(self, auto_derived_types: AutoDerivedTyped) -> None: 128# self._class_wrapper__data_size: int = 0 129# self._class_wrapper__auto_derived_types: AutoDerivedTyped = auto_derived_types 130 131# def __call__(self, *args: Any, **kwds: Any) -> Any: 132# pass 133 134 135class ClassWrappingFactory: 136 instance_id = IDGenerator() 137 138 def __init__(self): 139 self._instance_id: int = self.instance_id() 140 self.class_name = type(self).__name__ 141 self.derived: Dict[Type, Type] = dict() 142 self._attributes_and_methods: Dict[str, Callable] = { 143 '_class_wrapper__field_names': None, 144 '_class_wrapper__factory': self, 145 } 146 147 @property 148 def attributes_and_methods(self) -> Dict[str, Callable]: 149 return self._attributes_and_methods.copy() 150 151 def edit_derived_type(self, obj: Any, attributes_and_methods, field_names: Tuple[str]) -> Dict[str, Any]: 152 """Might be overloaded in a derived class if needed 153 154 Args: 155 obj (Any): _description_ 156 attributes_and_methods (_type_): _description_ 157 field_names (Tuple[str]): _description_ 158 data_size (int): _description_ 159 160 Returns: 161 Dict[str, Any]: _description_ 162 """ 163 return attributes_and_methods 164 165 def gen_kwargs(self, obj: Any, attributes_and_methods, field_names: Tuple[str]) -> Optional[Dict[str, Any]]: 166 """Might be overloaded in a derived class if needed 167 168 Args: 169 obj (Any): _description_ 170 attributes_and_methods (_type_): _description_ 171 field_names (Tuple[str]): _description_ 172 data_size (int): _description_ 173 174 Returns: 175 Optional[Dict[str, Any]]: _description_ 176 """ 177 return None 178 179 def __call__(self, obj: Any): 180 base_type = type(obj) 181 base_type_name = base_type.__name__ 182 try: 183 return self.derived[base_type_name] 184 except KeyError: 185 field_names: Tuple = self.gen_fields_tuple(obj) 186 187 attributes_and_methods = self.attributes_and_methods 188 attributes_and_methods['_class_wrapper__field_names'] = field_names 189 attributes_and_methods = self.edit_derived_type(obj, attributes_and_methods, field_names) 190 kwargs = self.gen_kwargs(obj, attributes_and_methods, field_names) 191 192 if kwargs is None: 193 derived: Type = type(f'{self.class_name}__{self._instance_id}__from__{base_type_name}', (base_type,), attributes_and_methods) 194 else: 195 derived: Type = type(f'{self.class_name}__{self._instance_id}__from__{base_type_name}', (base_type,), attributes_and_methods, **kwargs) 196 197 self.derived[base_type_name] = derived 198 return derived 199 200 def gen_fields_tuple(self, obj: Any): 201 try: 202 return obj.__slots__ 203 except AttributeError: 204 pass 205 206 try: 207 return tuple(obj.__dict__.keys()) 208 except AttributeError: 209 pass 210 211 return tuple() 212 213 214class ClassWrappingFactoryWithObjDataSizeLog(ClassWrappingFactory): 215 def __init__(self): 216 super().__init__() 217 self.attributes_and_methods['_class_wrapper__obj_data_size_log'] = 0 218 219 def edit_derived_type(self, obj: Any, attributes_and_methods, field_names: Tuple[str], data_size: int) -> Dict[str, Any]: 220 data_size = 0 221 for field_name in field_names: 222 data_size += getsizeof(getattr(obj, field_name)) 223 224 attributes_and_methods['_class_wrapper__obj_data_size_log'] = data_size 225 return attributes_and_methods
50class AutoDerivedClass: 51 instance_id = IDGenerator() 52 53 def __init__(self): 54 self._instance_id: int = self.instance_id() 55 self.class_name = type(self).__name__ 56 self.derived: Dict[Type, Type] = dict() 57 58 @property 59 def methods(self) -> Dict[str, Callable]: 60 raise NotImplementedError 61 62 def base_classes(self, base_type) -> Tuple[Type]: 63 return (base_type,) 64 65 def __call__(self, base_type: Type): 66 base_type_name = base_type.__name__ 67 try: 68 return self.derived[base_type_name] 69 except KeyError: 70 result: Type = type(f'{self.class_name}__{self._instance_id}__from__{base_type_name}', self.base_classes(base_type), self.methods) 71 self.derived[base_type_name] = result 72 return result
75class BaseAutoDerivedObjWrapper: 76 instance_id = IDGenerator() 77 78 def __init__(self): 79 self._instance_id: int = self.instance_id() 80 self.class_name = type(self).__name__ 81 self.derived: Dict[Type, Type] = dict() 82 self.t = self.type 83 84 def wrapping_required(self, obj: Any, base_type: Type, fields: Tuple[str], planned_type_name: str) -> bool: 85 return True 86 87 def methods(self, obj: Any, base_type: Type, fields: Tuple[str]) -> Dict[str, Callable]: 88 raise NotImplementedError 89 90 def base_classes(self, obj: Any, base_type: Type, fields: Tuple[str]) -> Tuple[Type]: 91 return (base_type,) 92 93 def gen_fields_tuple(self, obj: Any): 94 try: 95 return obj.__slots__ 96 except AttributeError: 97 pass 98 99 return tuple(obj.__dict__.keys()) 100 101 def type(self, obj: Any): 102 base_type = type(obj) 103 base_type_name = base_type.__name__ 104 try: 105 return self.derived[base_type_name] 106 except KeyError: 107 fields: Tuple = self.gen_fields_tuple(obj) 108 planned_type_name: str = f'{self.class_name}__{self._instance_id}__from__{base_type_name}' 109 if self.wrapping_required(obj, base_type, fields, planned_type_name): 110 result: Type = type(planned_type_name, self.base_classes(obj, base_type, fields), self.methods(obj, base_type, fields)) 111 self.derived[base_type_name] = result 112 return result 113 else: 114 return base_type 115 116 def __call__(self, obj: Any): 117 obj.__class__ = self.type(obj) 118 return obj 119 120 def temporary(self, obj: Any): 121 return TemporaryReinterpretCast(self.type(obj), obj) 122 123 def persistent(self, obj: Any): 124 return reinterpret_cast(self.type(obj), obj)
101 def type(self, obj: Any): 102 base_type = type(obj) 103 base_type_name = base_type.__name__ 104 try: 105 return self.derived[base_type_name] 106 except KeyError: 107 fields: Tuple = self.gen_fields_tuple(obj) 108 planned_type_name: str = f'{self.class_name}__{self._instance_id}__from__{base_type_name}' 109 if self.wrapping_required(obj, base_type, fields, planned_type_name): 110 result: Type = type(planned_type_name, self.base_classes(obj, base_type, fields), self.methods(obj, base_type, fields)) 111 self.derived[base_type_name] = result 112 return result 113 else: 114 return base_type
136class ClassWrappingFactory: 137 instance_id = IDGenerator() 138 139 def __init__(self): 140 self._instance_id: int = self.instance_id() 141 self.class_name = type(self).__name__ 142 self.derived: Dict[Type, Type] = dict() 143 self._attributes_and_methods: Dict[str, Callable] = { 144 '_class_wrapper__field_names': None, 145 '_class_wrapper__factory': self, 146 } 147 148 @property 149 def attributes_and_methods(self) -> Dict[str, Callable]: 150 return self._attributes_and_methods.copy() 151 152 def edit_derived_type(self, obj: Any, attributes_and_methods, field_names: Tuple[str]) -> Dict[str, Any]: 153 """Might be overloaded in a derived class if needed 154 155 Args: 156 obj (Any): _description_ 157 attributes_and_methods (_type_): _description_ 158 field_names (Tuple[str]): _description_ 159 data_size (int): _description_ 160 161 Returns: 162 Dict[str, Any]: _description_ 163 """ 164 return attributes_and_methods 165 166 def gen_kwargs(self, obj: Any, attributes_and_methods, field_names: Tuple[str]) -> Optional[Dict[str, Any]]: 167 """Might be overloaded in a derived class if needed 168 169 Args: 170 obj (Any): _description_ 171 attributes_and_methods (_type_): _description_ 172 field_names (Tuple[str]): _description_ 173 data_size (int): _description_ 174 175 Returns: 176 Optional[Dict[str, Any]]: _description_ 177 """ 178 return None 179 180 def __call__(self, obj: Any): 181 base_type = type(obj) 182 base_type_name = base_type.__name__ 183 try: 184 return self.derived[base_type_name] 185 except KeyError: 186 field_names: Tuple = self.gen_fields_tuple(obj) 187 188 attributes_and_methods = self.attributes_and_methods 189 attributes_and_methods['_class_wrapper__field_names'] = field_names 190 attributes_and_methods = self.edit_derived_type(obj, attributes_and_methods, field_names) 191 kwargs = self.gen_kwargs(obj, attributes_and_methods, field_names) 192 193 if kwargs is None: 194 derived: Type = type(f'{self.class_name}__{self._instance_id}__from__{base_type_name}', (base_type,), attributes_and_methods) 195 else: 196 derived: Type = type(f'{self.class_name}__{self._instance_id}__from__{base_type_name}', (base_type,), attributes_and_methods, **kwargs) 197 198 self.derived[base_type_name] = derived 199 return derived 200 201 def gen_fields_tuple(self, obj: Any): 202 try: 203 return obj.__slots__ 204 except AttributeError: 205 pass 206 207 try: 208 return tuple(obj.__dict__.keys()) 209 except AttributeError: 210 pass 211 212 return tuple()
152 def edit_derived_type(self, obj: Any, attributes_and_methods, field_names: Tuple[str]) -> Dict[str, Any]: 153 """Might be overloaded in a derived class if needed 154 155 Args: 156 obj (Any): _description_ 157 attributes_and_methods (_type_): _description_ 158 field_names (Tuple[str]): _description_ 159 data_size (int): _description_ 160 161 Returns: 162 Dict[str, Any]: _description_ 163 """ 164 return attributes_and_methods
Might be overloaded in a derived class if needed
Args: obj (Any): _description_ attributes_and_methods (_type_): _description_ field_names (Tuple[str]): _description_ data_size (int): _description_
Returns: Dict[str, Any]: _description_
166 def gen_kwargs(self, obj: Any, attributes_and_methods, field_names: Tuple[str]) -> Optional[Dict[str, Any]]: 167 """Might be overloaded in a derived class if needed 168 169 Args: 170 obj (Any): _description_ 171 attributes_and_methods (_type_): _description_ 172 field_names (Tuple[str]): _description_ 173 data_size (int): _description_ 174 175 Returns: 176 Optional[Dict[str, Any]]: _description_ 177 """ 178 return None
Might be overloaded in a derived class if needed
Args: obj (Any): _description_ attributes_and_methods (_type_): _description_ field_names (Tuple[str]): _description_ data_size (int): _description_
Returns: Optional[Dict[str, Any]]: _description_
215class ClassWrappingFactoryWithObjDataSizeLog(ClassWrappingFactory): 216 def __init__(self): 217 super().__init__() 218 self.attributes_and_methods['_class_wrapper__obj_data_size_log'] = 0 219 220 def edit_derived_type(self, obj: Any, attributes_and_methods, field_names: Tuple[str], data_size: int) -> Dict[str, Any]: 221 data_size = 0 222 for field_name in field_names: 223 data_size += getsizeof(getattr(obj, field_name)) 224 225 attributes_and_methods['_class_wrapper__obj_data_size_log'] = data_size 226 return attributes_and_methods
220 def edit_derived_type(self, obj: Any, attributes_and_methods, field_names: Tuple[str], data_size: int) -> Dict[str, Any]: 221 data_size = 0 222 for field_name in field_names: 223 data_size += getsizeof(getattr(obj, field_name)) 224 225 attributes_and_methods['_class_wrapper__obj_data_size_log'] = data_size 226 return attributes_and_methods
Might be overloaded in a derived class if needed
Args: obj (Any): _description_ attributes_and_methods (_type_): _description_ field_names (Tuple[str]): _description_ data_size (int): _description_
Returns: Dict[str, Any]: _description_