diamondback.commons package

Submodules

diamondback.commons.log module

Description

Log formats and writes log entries with a specified level and stream using loguru. 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.")

    # Standard output.

    Log.stream(sys.stdout)
    Log.write("Info", f"Valid = {True}")

    # 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)

    # File.

    with open("log.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 - 2026 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: ClassVar[tuple[str, ...]] = ('Critical', 'Error', 'Warning', 'Success', 'Info', 'Debug', 'Trace')
classmethod level(level: str) → None[source]

Level.

Parameters:

level (str - in LEVEL)

classmethod stream(stream: Any) → None[source]

Stream.

Parameters:

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 loguru with a specified level and stream. Log entries contain an ISO 8601 datetime and level.

Parameters:

  • level (str - in LEVEL)

  • entry (str | Exception)

diamondback.commons.rest_client module

Description

REST client defines a client for simple REST service requests using 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.

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 - 2026 Larry Turner, Schneider Electric Industries SAS. All rights reserved.

Author

Larry Turner, Schneider Electric, AI Hub, 2020-10-22.

class diamondback.commons.rest_client.RestClient[source]

Bases: object

REST client.

Initialize.

METHOD: ClassVar[tuple[str, ...]] = ('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.

Parameters:

  • method (str - in ("delete", "get", "head", "options", "patch", "post", "put"))

  • api (str - relative to the URL)

  • auth (Any)

  • header (dict[str, str] | None)

  • item (dict[str, str] | None)

  • data (Any)

  • json (Any)

Returns:

value

Return type:

requests.Response

property timeout
property url

diamondback.commons.serial module

Description

Serial encodes and decodes an instance to a Base-85 encoded serialized string with elective gzip compression, and generates SHA3-256 hash codes.

Decode decodes a Base-85 encoded serialized string to an instance. Elective compression is applied with gzip and automatically detected.

Encode encodes an instance to a Base-85 encoded serialized string. Elective compression is applied with gzip.

Code generates an SHA3-256 hash code of an encoded serialized string, with applications in security and version control. Encode with compression embeds a datetime context, and code will not be consistent or deterministic.

An instance may be an object or a collection, referenced by abstract or concrete types, and the instance will be completely encoded and decoded, without custom encoding definitions.

Singleton.

Thread safe.

Example

from diamondback import Serial
import pandas

# Define a model instance of any supported type.

x = pandas.DataFrame(dict(fruit = ["orange", "apple", "kiwi"], value = [1.25, 1.5, 0.30]))

# Encode and decode a model instance with compression.

y = Serial.encode(x)
z = Serial.decode(y)

# Encode and decode a model instance without compression.

y = Serial.encode(x, compress=False)
z = Serial.decode(y)

# Generate an SHA3-256 code.

code = Serial.code(y)
License

BSD-3C. © 2018 - 2026 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(x: str, encoding: str = 'utf-8') → str[source]

Code generates an SHA3-256 hash code.

Parameters:

  • x (str)

  • encoding (str)

Returns:

y

Return type:

str

static decode(x: str, encoding: str = 'utf-8') → Any[source]

Decodes a Base-85 encoded serialized string to an instance. Elective compression is applied with gzip and automatically detected.

Parameters:

  • x (str)

  • encoding (str)

Returns:

y

Return type:

Any

static encode(x: Any, compress: bool = True, encoding: str = 'utf-8') → str[source]

Encodes an instance to a Base-85 encoded serialized string. Elective compression is applied with gzip.

Parameters:

  • x (Any)

  • compress (bool)

  • encoding (str)

Returns:

y

Return type:

str

Module contents

Description

Initialize.

License

BSD-3C. © 2018 - 2026 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: ClassVar[tuple[str, ...]] = ('Critical', 'Error', 'Warning', 'Success', 'Info', 'Debug', 'Trace')
classmethod level(level: str) → None[source]

Level.

Parameters:

level (str - in LEVEL)

classmethod stream(stream: Any) → None[source]

Stream.

Parameters:

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 loguru with a specified level and stream. Log entries contain an ISO 8601 datetime and level.

Parameters:

  • level (str - in LEVEL)

  • entry (str | Exception)

class diamondback.commons.RestClient[source]

Bases: object

REST client.

Initialize.

METHOD: ClassVar[tuple[str, ...]] = ('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.

Parameters:

  • method (str - in ("delete", "get", "head", "options", "patch", "post", "put"))

  • api (str - relative to the URL)

  • auth (Any)

  • header (dict[str, str] | None)

  • item (dict[str, str] | None)

  • data (Any)

  • json (Any)

Returns:

value

Return type:

requests.Response

property timeout
property url
class diamondback.commons.Serial[source]

Bases: object

Serial.

static code(x: str, encoding: str = 'utf-8') → str[source]

Code generates an SHA3-256 hash code.

Parameters:

  • x (str)

  • encoding (str)

Returns:

y

Return type:

str

static decode(x: str, encoding: str = 'utf-8') → Any[source]

Decodes a Base-85 encoded serialized string to an instance. Elective compression is applied with gzip and automatically detected.

Parameters:

  • x (str)

  • encoding (str)

Returns:

y

Return type:

Any

static encode(x: Any, compress: bool = True, encoding: str = 'utf-8') → str[source]

Encodes an instance to a Base-85 encoded serialized string. Elective compression is applied with gzip.

Parameters:

  • x (Any)

  • compress (bool)

  • encoding (str)

Returns:

y

Return type:

str