diamondback.commons package
Submodules
diamondback.commons.Log module
- Description
A log instance formats and writes log entries with a specified level and stream using the loguru package. Log entries contain an ISO-8601 datetime and level. Dynamic stream redirection and level specification are supported.
Log uses lazy initialization to coexist with loguru, and removes or creates loguru handlers only on explicit stream assignment or write. In lazy initialization an existing default loguru handler, with an identity equal to 0, and a stream assignment of sys.stdout is removed, and a new loguru handler with a stream assignment of sys.stdout and a level of ‘Info’ is created.
In stream assignments subsequent to initialization, only loguru handlers previously created by Log will be removed, as the Log design pattern does not define multicast. The ability to create and modify externally defined loguru handlers, multicast, and utilize any native loguru functionality is supported.
Levels defined by loguru are supported, including custom definitions, which may have an associated numerical value greater than or equal to zero. Levels may be dynamically modified without creating, deleting, or modifying a loguru handler. Levels are case insensitive on assignment, though loguru uses upper case.
Singleton.
Thread safe.
Example
from diamondback import Log import io import numpy import sys try : # Set Log level to 'Info', the default level. Log.level( 'Info' ) Log.write( 'Info', 'Test Log write.' ) # Set Log stream to sys.stdout. Log.stream( sys.stdout ) Log.write( 'Info', f'Valid = {True}' ) # Set Log stream to a memory stream. stream = io.StringIO( ) Log.stream( stream ) x = numpy.random.rand( 2, 2 ) Log.write( 'Info', f'X = {x}' ) # Read and reset memory stream. value = stream.getvalue( ) _, _ = stream.truncate( 0 ), stream.seek( 0 ) # Set Log stream to a file. with open( 'log-2112.txt', 'w' ) as fout : Log.stream( fout ) x = numpy.random.rand( 2, 2 ) Log.write( 'Warning', f'X = {x}' ) except Exception as ex : Log.write( 'Error', ex )
- License
BSD-3C. © 2018 - 2024 Larry Turner, Schneider Electric Industries SAS. All rights reserved.
- Author
Larry Turner, Schneider Electric, AI Hub, 2018-03-22.
- class diamondback.commons.Log.Log[source]
Bases:
object
Log.
- LEVEL = ('Critical', 'Error', 'Warning', 'Success', 'Info', 'Debug', 'Trace')
- classmethod level(level: str) None [source]
Level.
- Arguments :
level : str - in LEVEL.
- classmethod stream(stream: Any) None [source]
Stream.
- Arguments :
stream : Any, hasattr( ‘write’ ) - in ( sys.stderr, sys.stdout, open( < path >, ‘w’ or ‘a’ ) ).
- classmethod write(level: str, entry: str | Exception) None [source]
Formats and writes log entries using the loguru package with a specified level and stream. Log entries contain an ISO-8601 datetime and level.
- Arguments :
level : str - in LEVEL. entry : Union[ str, Exception ].
diamondback.commons.RestClient module
- Description
REST client instances define a client for simple REST service requests using the requests package. An API and an elective dictionary of parameter strings are encoded to build a URL, elective binary or JSON data are defined in the body of a request, and a requests response containing JSON, text, or binary data is returned.
Proxy, timeout, and URL definition are supported.
Live makes a head request to a URL and detects a live service.
Example
from diamondback import RestClient import numpy class TestClient( RestClient ) : def __init__( self ) -> None : super( ).__init__( ) self.proxy = dict( http = '', https = '' ) def add( self, json : dict[ str, numpy.ndarray ] ) -> numpy.ndarray : return self.request( 'get', 'test/add', json = json ).json( ) client = TestClient( ) client.url = 'http://127.0.0.1:8080' client.timeout = ( 10.0, 60.0 ) # connect, read value = client.add( dict( x = numpy.random.rand( 3 ), y = numpy.random.rand( 3 ) ) )
- License
BSD-3C. © 2018 - 2024 Larry Turner, Schneider Electric Industries SAS. All rights reserved.
- Author
Larry Turner, Schneider Electric, AI Hub, 2020-10-22.
- class diamondback.commons.RestClient.RestClient[source]
Bases:
object
REST client.
Initialize.
- METHOD = ('Delete', 'Get', 'Head', 'Options', 'Patch', 'Post', 'Put')
- property live
- property proxy
- request(method: str, api: str, auth: Any = None, header: dict[str, str] | None = None, item: dict[str, str] | None = None, data: Any = None, json: Any = None) Response [source]
Request client for simple REST service requests. An API and an elective dictionary of parameter strings are encoded to build a URL, elective binary or JSON data are defined in the body of a request, and a requests response containing JSON, text, or binary data is returned.
- Arguments :
method : str - in ( ‘delete’, ‘get’, ‘head’, ‘options’, ‘patch’, ‘post’, ‘put’ ). api : str - relative to the URL. auth : Any. header : dict[ str, str ]. item : dict[ str, str ]. data : Any. json : Any.
- Returns :
value : requests.Response.
- property timeout
- property url
diamondback.commons.Serial module
- Description
A serial instance encodes and decodes an instance or collection in BSON or JSON, and generates SHA3-256 codes, using the jsonpickle package.
BSON, Base-85 encoded gzip JSON, embeds a datetime context, and code will not be consistent or useful for validation.
An instance may be an object or a collection, referenced by abstract or concrete types, and the instance will be correctly encoded and decoded, without custom encoding definitions. BSON binary format is selected by electing to compress. Encoding may be specified if an alternative to UTF-8 is required.
Comments may be filtered from JSON by electing to clean. Python style docstring and line comments are supported, though line comments must be terminated by a new line.
Singleton.
Thread safe.
Example
from diamondback import Serial import numpy import pandas # Encode and decode a dictionary in JSON. n = numpy.random.randint( 1, 10 ) x = dict( a = numpy.random.rand( n ), b = list( numpy.random.rand( n ) ) ) y = Serial.encode( x, indent = True ) z = Serial.decode( y ) # Encode and decode a dictionary in BSON. y = Serial.encode( x, compress = True ) z = Serial.decode( y, compress = True ) # Encode and decode a pandas DataFrame in BSON. model = pandas.DataFrame( dict( fruit = [ 'orange', 'apple', 'kiwi' ], value = [ 1.25, 1.5, 0.30 ] ) ) y = Serial.encode( model ) # Generate an SHA3-256 code. code = Serial.code( y ) z = Serial.decode( y ) # Decode in JSON. z = Serial.decode( '{ "a" : 1.0, "b" : 2.0, "c" : 3.14159 }' ) z = Serial.decode( '[ 1.0, 2.0, 3.0 ]' )
- License
BSD-3C. © 2018 - 2024 Larry Turner, Schneider Electric Industries SAS. All rights reserved.
- Author
Larry Turner, Schneider Electric, AI Hub, 2018-07-13.
- class diamondback.commons.Serial.Serial[source]
Bases:
object
Serial.
- static code(state: str, encoding: str = 'utf_8') str [source]
Code generation. SHA3-256 hash.
- Arguments :
state : str. encoding : str.
- Returns :
code : str.
- static decode(state: str, compress: bool = False, encoding: str = 'utf_8', clean: bool = False) Any [source]
Decodes an instance or collection from a BSON or JSON state. Encoding may be specified if an alternative to UTF-8 is required. Python style docstring and line comments may be cleaned, though line comments must be terminated by a new line.
- Arguments :
state : str. compress : bool. encoding : str. clean : bool - clean comments.
- Returns :
instance : Any.
- static encode(instance: Any, compress: bool = False, encoding: str = 'utf_8', indent: bool = False) str [source]
Encodes BSON or JSON. Encoding may be specified if an alternative to UTF-8 is required.
- Arguments :
instance : Any. compress : bool. encoding : str. indent : bool.
- Returns :
state : str.
Module contents
- Description
Initialize.
- License
BSD-3C. © 2018 - 2024 Larry Turner, Schneider Electric Industries SAS. All rights reserved.
- Author
Larry Turner, Schneider Electric, AI Hub, 2018-03-22.
- class diamondback.commons.Log[source]
Bases:
object
Log.
- LEVEL = ('Critical', 'Error', 'Warning', 'Success', 'Info', 'Debug', 'Trace')
- classmethod level(level: str) None [source]
Level.
- Arguments :
level : str - in LEVEL.
- classmethod stream(stream: Any) None [source]
Stream.
- Arguments :
stream : Any, hasattr( ‘write’ ) - in ( sys.stderr, sys.stdout, open( < path >, ‘w’ or ‘a’ ) ).
- classmethod write(level: str, entry: str | Exception) None [source]
Formats and writes log entries using the loguru package with a specified level and stream. Log entries contain an ISO-8601 datetime and level.
- Arguments :
level : str - in LEVEL. entry : Union[ str, Exception ].
- class diamondback.commons.RestClient[source]
Bases:
object
REST client.
Initialize.
- METHOD = ('Delete', 'Get', 'Head', 'Options', 'Patch', 'Post', 'Put')
- property live
- property proxy
- request(method: str, api: str, auth: Any = None, header: dict[str, str] | None = None, item: dict[str, str] | None = None, data: Any = None, json: Any = None) Response [source]
Request client for simple REST service requests. An API and an elective dictionary of parameter strings are encoded to build a URL, elective binary or JSON data are defined in the body of a request, and a requests response containing JSON, text, or binary data is returned.
- Arguments :
method : str - in ( ‘delete’, ‘get’, ‘head’, ‘options’, ‘patch’, ‘post’, ‘put’ ). api : str - relative to the URL. auth : Any. header : dict[ str, str ]. item : dict[ str, str ]. data : Any. json : Any.
- Returns :
value : requests.Response.
- property timeout
- property url
- class diamondback.commons.Serial[source]
Bases:
object
Serial.
- static code(state: str, encoding: str = 'utf_8') str [source]
Code generation. SHA3-256 hash.
- Arguments :
state : str. encoding : str.
- Returns :
code : str.
- static decode(state: str, compress: bool = False, encoding: str = 'utf_8', clean: bool = False) Any [source]
Decodes an instance or collection from a BSON or JSON state. Encoding may be specified if an alternative to UTF-8 is required. Python style docstring and line comments may be cleaned, though line comments must be terminated by a new line.
- Arguments :
state : str. compress : bool. encoding : str. clean : bool - clean comments.
- Returns :
instance : Any.
- static encode(instance: Any, compress: bool = False, encoding: str = 'utf_8', indent: bool = False) str [source]
Encodes BSON or JSON. Encoding may be specified if an alternative to UTF-8 is required.
- Arguments :
instance : Any. compress : bool. encoding : str. indent : bool.
- Returns :
state : str.