cengal.parallel_execution.coroutines.coro_standard_services.remote_nodes.versions.v_0.class_info

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

  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"""
 20Module Docstring
 21Docstrings: http://www.python.org/dev/peps/pep-0257/
 22"""
 23
 24
 25__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 28__license__ = "Apache License, Version 2.0"
 29__version__ = "4.4.1"
 30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 31__email__ = "gtalk@butenkoms.space"
 32# __status__ = "Prototype"
 33__status__ = "Development"
 34# __status__ = "Production"
 35
 36
 37from enum import Enum
 38from cengal.parallel_execution.coroutines.coro_scheduler import *
 39from cengal.parallel_execution.coroutines.coro_tools.await_coro import *
 40from cengal.parallel_execution.coroutines.coro_standard_services.asyncio_loop import *
 41from cengal.parallel_execution.coroutines.coro_standard_services.loop_yield import CoroPriority
 42from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import *
 43from cengal.parallel_execution.coroutines.coro_standard_services.timer_func_runner import *
 44from cengal.file_system.file_manager import path_relative_to_current_dir
 45from cengal.time_management.cpu_clock_cycles import perf_counter
 46from cengal.data_manipulation.serialization import *
 47from typing import Hashable, Tuple, List, Any, Dict, Callable, Type
 48from cengal.introspection.inspect import get_exception, entity_owning_module_importable_str, entity_owning_module_info_and_owning_path, entity_properties
 49from cengal.io.core.memory_management import IOCoreMemoryManagement
 50from cengal.parallel_execution.asyncio.efficient_streams import StreamManagerIOCoreMemoryManagement, TcpStreamManager, UdpStreamManager, StreamManagerAbstract
 51from cengal.code_flow_control.smart_values import ValueExistence
 52from cengal.io.named_connections.named_connections_manager import NamedConnectionsManager
 53from cengal.code_flow_control.args_manager import number_of_provided_args
 54from cengal.data_manipulation.serialization import Serializer, Serializers, best_serializer_for_standard_data
 55from cengal.code_flow_control.args_manager import find_arg_position_and_value, UnknownArgumentError
 56from cengal.data_generation.id_generator import IDGenerator, GeneratorType
 57from cengal.system import PLATFORM_NAME, PYTHON_VERSION
 58from importlib import import_module
 59import sys
 60import os
 61import asyncio
 62
 63from .exceptions import *
 64from .commands import *
 65
 66
 67class LocalClassInfo:
 68    def __init__(self, local_id: Hashable, class_type: Type) -> None:
 69        self.class_type: Type = class_type
 70        self._class_name: str = self.class_type.__name__
 71        self._module_importable_str: str = entity_owning_module_importable_str(self.class_type)
 72        self._local_id: Hashable = local_id
 73    
 74    def __call__(self) -> Type:
 75        return {
 76            CommandDataFieldsDeclareServiceClass.local_id.value: self._local_id,
 77            CommandDataFieldsDeclareServiceClass.class_name.value: self._class_name,
 78            CommandDataFieldsDeclareServiceClass.module_importable_str.value: self._module_importable_str,
 79        }
 80    
 81    @property
 82    def class_name(self):
 83        return self._class_name
 84    
 85    @property
 86    def module_importable_str(self):
 87        return self._module_importable_str
 88    
 89    @property
 90    def local_id(self):
 91        return self._local_id
 92
 93
 94class RemoteClassInfo:
 95    def __init__(self, local_id: Hashable, class_name: str, module_importable_str: str) -> None:
 96        self.local_id: Hashable = local_id
 97        self.class_name: str = class_name
 98        self.module_importable_str: str = module_importable_str
 99        self.module = import_module(self.module_importable_str)
100        self._class_type: Type = getattr(self.module, self.class_name)
101    
102    @classmethod
103    def from_data(cls, data: Dict[Hashable, Any]) -> 'RemoteClassInfo':
104        local_id: Hashable = data[CommandDataFieldsDeclareServiceClass.local_id.value]
105        class_name: str = data[CommandDataFieldsDeclareServiceClass.class_name.value]
106        module_importable_str: str = data[CommandDataFieldsDeclareServiceClass.module_importable_str.value]
107        return cls(local_id, class_name, module_importable_str)
108    
109    def __call__(self) -> Type:
110        return self._class_type
111    
112    @property
113    def class_type(self):
114        return self()

class LocalClassInfo: View Source

68class LocalClassInfo:
69    def __init__(self, local_id: Hashable, class_type: Type) -> None:
70        self.class_type: Type = class_type
71        self._class_name: str = self.class_type.__name__
72        self._module_importable_str: str = entity_owning_module_importable_str(self.class_type)
73        self._local_id: Hashable = local_id
74    
75    def __call__(self) -> Type:
76        return {
77            CommandDataFieldsDeclareServiceClass.local_id.value: self._local_id,
78            CommandDataFieldsDeclareServiceClass.class_name.value: self._class_name,
79            CommandDataFieldsDeclareServiceClass.module_importable_str.value: self._module_importable_str,
80        }
81    
82    @property
83    def class_name(self):
84        return self._class_name
85    
86    @property
87    def module_importable_str(self):
88        return self._module_importable_str
89    
90    @property
91    def local_id(self):
92        return self._local_id

LocalClassInfo(local_id: typing.Hashable, class_type: typing.Type) View Source

69    def __init__(self, local_id: Hashable, class_type: Type) -> None:
70        self.class_type: Type = class_type
71        self._class_name: str = self.class_type.__name__
72        self._module_importable_str: str = entity_owning_module_importable_str(self.class_type)
73        self._local_id: Hashable = local_id

class_type: Type

class_name View Source

82    @property
83    def class_name(self):
84        return self._class_name

module_importable_str View Source

86    @property
87    def module_importable_str(self):
88        return self._module_importable_str

local_id View Source

90    @property
91    def local_id(self):
92        return self._local_id

class RemoteClassInfo: View Source

 95class RemoteClassInfo:
 96    def __init__(self, local_id: Hashable, class_name: str, module_importable_str: str) -> None:
 97        self.local_id: Hashable = local_id
 98        self.class_name: str = class_name
 99        self.module_importable_str: str = module_importable_str
100        self.module = import_module(self.module_importable_str)
101        self._class_type: Type = getattr(self.module, self.class_name)
102    
103    @classmethod
104    def from_data(cls, data: Dict[Hashable, Any]) -> 'RemoteClassInfo':
105        local_id: Hashable = data[CommandDataFieldsDeclareServiceClass.local_id.value]
106        class_name: str = data[CommandDataFieldsDeclareServiceClass.class_name.value]
107        module_importable_str: str = data[CommandDataFieldsDeclareServiceClass.module_importable_str.value]
108        return cls(local_id, class_name, module_importable_str)
109    
110    def __call__(self) -> Type:
111        return self._class_type
112    
113    @property
114    def class_type(self):
115        return self()

RemoteClassInfo( local_id: typing.Hashable, class_name: str, module_importable_str: str) View Source

 96    def __init__(self, local_id: Hashable, class_name: str, module_importable_str: str) -> None:
 97        self.local_id: Hashable = local_id
 98        self.class_name: str = class_name
 99        self.module_importable_str: str = module_importable_str
100        self.module = import_module(self.module_importable_str)
101        self._class_type: Type = getattr(self.module, self.class_name)

local_id: Hashable

class_name: str

module_importable_str: str

module

@classmethod

def from_data( cls, data: typing.Dict[typing.Hashable, typing.Any]) -> RemoteClassInfo: View Source

103    @classmethod
104    def from_data(cls, data: Dict[Hashable, Any]) -> 'RemoteClassInfo':
105        local_id: Hashable = data[CommandDataFieldsDeclareServiceClass.local_id.value]
106        class_name: str = data[CommandDataFieldsDeclareServiceClass.class_name.value]
107        module_importable_str: str = data[CommandDataFieldsDeclareServiceClass.module_importable_str.value]
108        return cls(local_id, class_name, module_importable_str)

class_type View Source

113    @property
114    def class_type(self):
115        return self()