artificial_artwork.tf_session_runner

Complexity

Total Complexity

8

Size/Duplication

Total Lines	42
Duplicated Lines	0 %

Importance

Changes

0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	TensorflowSessionRunner.__init__()	0	3	1
A	TensorflowSessionRunner.with_default_graph_reset()	0	6	1
A	TensorflowSessionRunnerSubject.request()	0	2	1
A	TensorflowSessionRunnerSubject.__init__()	0	2	1
A	TensorflowSessionRunner.run()	0	2	1
A	TensorflowSessionRunner.request()	0	7	2
A	TensorflowSessionRunner.session()	0	3	1

from typing import List
import tensorflow as tf

from .utils.proxy import RealSubject, Proxy


class TensorflowSessionRunnerSubject(RealSubject):
    def __init__(self, interactive_session) -> None:
        self.interactive_session = interactive_session

    def request(self, *args, **kwargs):
        return self.interactive_session.run(*args, **kwargs)


class TensorflowSessionRunner(Proxy):
    def __init__(self, real_subject) -> None:
        super().__init__(real_subject)
        self.args_history: List[str] = []

    def request(self, *args, **kwargs):
        self.args_history.append(f"ARGS: [{', '.join((str(_) for _ in args))}], KWARGS: [{', '.join((f'{k}={v}' for k, v in kwargs.items()))}]")
        try:
        # We know that Proxy executes request by executing the request method on the subject
            return super().request(*args, **kwargs)
        except Exception as e:
            raise e

    @property
    def session(self):
        return self._real_subject.interactive_session

    def run(self, *args, **kwargs):
        return self.request(*args, **kwargs)

    @classmethod
    def with_default_graph_reset(cls):
        tf.compat.v1.reset_default_graph()
        tf.compat.v1.disable_eager_execution()
        return TensorflowSessionRunner(TensorflowSessionRunnerSubject(
            tf.compat.v1.InteractiveSession()))