Skip to content

Base Kernel

penvm.kernels.base.client

Client-side base kernel support.

KernelClient

Bases: BaseObject

Base kernel client.

Provides name/oid and logger setup.

Source code in penvm/src/kernels/penvm/kernels/base/client.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
class KernelClient(BaseObject):
    """Base kernel client.

    Provides name/oid and logger setup.
    """

    name = "base"

    def __init__(
        self,
        session: "Session",
    ):
        """Initialize.

        Args:
            session: Owning session.
        """
        try:
            super().__init__(self.name, logger)
            tlogger = self.logger.enter()

            self.session = session
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def __repr__(self):
        return f"<KernelClient name={self.name}>"

__init__(session)

Initialize.

Parameters:

Name Type Description Default
session Session

Owning session.

 required
Source code in penvm/src/kernels/penvm/kernels/base/client.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
def __init__(
    self,
    session: "Session",
):
    """Initialize.

    Args:
        session: Owning session.
    """
    try:
        super().__init__(self.name, logger)
        tlogger = self.logger.enter()

        self.session = session
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

penvm.kernels.base.server

Server-side base Kernel, Op, and OpContext support.

Kernel

Bases: BaseObject

Base kernel.

Provides base functionality for working with the kernel.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
class Kernel(BaseObject):
    """Base kernel.

    Provides base functionality for working with the kernel.
    """

    name = "base"

    def __init__(self):
        """Initialize."""
        try:
            super().__init__(self.name, logger)
            tlogger = self.logger.enter()

            self.updated = False
            self.ops = {}
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def copy(self):
        """Deep copy of this kernel."""
        return copy.deepcopy(self)

    def get_bases(self):
        """Bases (MRO) for kernel."""
        return self.__class__.__mro__

    def list(self):
        """List operation names."""
        return list(self.ops.keys())

    def register(
        self,
        opname: str,
        op: "Op",
    ):
        """Register operation.

        Args:
            opname: Operation name.
            op: Op reference.
        """
        self.updated = True
        self.ops[opname] = op

    def register_code(
        self,
        opname: str,
        opclassname: str,
        code: str,
    ):
        """Register an op by code (in a string).

        Args:
            opname: Operation name.
            opclassname: Class name in code snippet.
            code: Code snippet.
        """
        try:
            spec = importlib.util.spec_from_loader("ext", loader=None)
            ext = importlib.util.module_from_spec(spec)
            exec(code, ext.__dict__)

            cls = getattr(ext, opclassname)
            if cls:
                op = cls()
                if hasattr(op, "run"):
                    self.ops[opname] = cls()
                    self.logger.debug(f"ops ({list(self.ops.keys())} {cls=}")
                    self.logger.debug("register code succeeded")
                else:
                    self.logger.debug("register class has no run method")
            else:
                self.logger.debug(f"class {opclassname} not found in code")
        except Exception as e:
            self.logger.warning(f"EXCEPTION register code failed {e}")

    def run(
        self,
        opname: str,
        ctxt: "OpContext",
        req: "Message",
    ):
        """Run the op for `opname`.

        As a *convenience*, a response return value is put into the
        `session.omq`.

        Args:
            opname: Operation to run, referenced by name.
            ctxt: Context to provide operation.
            req: Request message to provide operation.
        """
        t0 = time.time()
        resp = self.run_local(opname, ctxt, req)
        if isinstance(resp, Message):
            t1 = time.time()
            resp.payload["-oprun-elapsed"] = t1 - t0
            ctxt.session.omq.put(resp)
        else:
            if opname not in self.ops:
                self.logger.error(f"run opname={opname} not found")
            else:
                # TODO: why is this here? a non-responding op is ok!
                self.logger.error(f"run opname={opname} did not succeed")

    def run_local(
        self,
        opname: str,
        ctxt: "OpContext",
        req: "Message",
    ) -> "Message":
        """Run the op for `opname` and return result.

        Suitable for local use.

        Args:
            opname: See [penvm.kernels.base.server.Kernel.run][].
            ctxt: See [penvm.kernels.base.server.Kernel.run][].
            req: See [penvm.kernels.base.server.Kernel.run][].

        Returns:
            Response (if generated).
        """
        op = self.ops.get(opname)
        if op:
            return op.run(ctxt, req)

    def state(self) -> "State":
        """Get object state.

        Returns:
            `State` object.
        """
        try:
            opnames = sorted(list(self.ops.keys()))
            return State(
                "kernel",
                self.name,
                {
                    "bases": [getattr(cls, "name") for cls in self.get_bases() if cls != object],
                    "nops": len(opnames),
                    "ops": opnames,
                    "updated": self.updated,
                },
            )
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")

    def unregister(self, opname: str):
        """Unregister operation.

        Args:
            opname: Operation name.
        """
        try:
            del self.ops[opname]
        except Exception as e:
            pass

__init__()

Initialize.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
42
43
44
45
46
47
48
49
50
51
52
53
 
def __init__(self):
    """Initialize."""
    try:
        super().__init__(self.name, logger)
        tlogger = self.logger.enter()

        self.updated = False
        self.ops = {}
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

copy()

Deep copy of this kernel.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
55
56
57
 
def copy(self):
    """Deep copy of this kernel."""
    return copy.deepcopy(self)

get_bases()

Bases (MRO) for kernel.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
59
60
61
 
def get_bases(self):
    """Bases (MRO) for kernel."""
    return self.__class__.__mro__

list()

List operation names.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
63
64
65
 
def list(self):
    """List operation names."""
    return list(self.ops.keys())

register(opname, op)

Register operation.

Parameters:

Name Type Description Default
opname str

Operation name.

 required
op Op

Op reference.

 required
Source code in penvm/src/kernels/penvm/kernels/base/server.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
 
def register(
    self,
    opname: str,
    op: "Op",
):
    """Register operation.

    Args:
        opname: Operation name.
        op: Op reference.
    """
    self.updated = True
    self.ops[opname] = op

register_code(opname, opclassname, code)

Register an op by code (in a string).

Parameters:

Name Type Description Default
opname str

Operation name.

 required
opclassname str

Class name in code snippet.

 required
code str

Code snippet.

 required
Source code in penvm/src/kernels/penvm/kernels/base/server.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
 
def register_code(
    self,
    opname: str,
    opclassname: str,
    code: str,
):
    """Register an op by code (in a string).

    Args:
        opname: Operation name.
        opclassname: Class name in code snippet.
        code: Code snippet.
    """
    try:
        spec = importlib.util.spec_from_loader("ext", loader=None)
        ext = importlib.util.module_from_spec(spec)
        exec(code, ext.__dict__)

        cls = getattr(ext, opclassname)
        if cls:
            op = cls()
            if hasattr(op, "run"):
                self.ops[opname] = cls()
                self.logger.debug(f"ops ({list(self.ops.keys())} {cls=}")
                self.logger.debug("register code succeeded")
            else:
                self.logger.debug("register class has no run method")
        else:
            self.logger.debug(f"class {opclassname} not found in code")
    except Exception as e:
        self.logger.warning(f"EXCEPTION register code failed {e}")

run(opname, ctxt, req)

Run the op for opname.

As a convenience, a response return value is put into the session.omq.

Parameters:

Name Type Description Default
opname str

Operation to run, referenced by name.

 required
ctxt OpContext

Context to provide operation.

 required
req Message

Request message to provide operation.

 required
Source code in penvm/src/kernels/penvm/kernels/base/server.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
def run(
    self,
    opname: str,
    ctxt: "OpContext",
    req: "Message",
):
    """Run the op for `opname`.

    As a *convenience*, a response return value is put into the
    `session.omq`.

    Args:
        opname: Operation to run, referenced by name.
        ctxt: Context to provide operation.
        req: Request message to provide operation.
    """
    t0 = time.time()
    resp = self.run_local(opname, ctxt, req)
    if isinstance(resp, Message):
        t1 = time.time()
        resp.payload["-oprun-elapsed"] = t1 - t0
        ctxt.session.omq.put(resp)
    else:
        if opname not in self.ops:
            self.logger.error(f"run opname={opname} not found")
        else:
            # TODO: why is this here? a non-responding op is ok!
            self.logger.error(f"run opname={opname} did not succeed")

run_local(opname, ctxt, req)

Run the op for opname and return result.

Suitable for local use.

Parameters:

Name Type Description Default
opname str

See penvm.kernels.base.server.Kernel.run.

 required
ctxt OpContext

See penvm.kernels.base.server.Kernel.run.

 required
req Message

See penvm.kernels.base.server.Kernel.run.

 required

Returns:

Type Description
Message

Response (if generated).
Source code in penvm/src/kernels/penvm/kernels/base/server.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
def run_local(
    self,
    opname: str,
    ctxt: "OpContext",
    req: "Message",
) -> "Message":
    """Run the op for `opname` and return result.

    Suitable for local use.

    Args:
        opname: See [penvm.kernels.base.server.Kernel.run][].
        ctxt: See [penvm.kernels.base.server.Kernel.run][].
        req: See [penvm.kernels.base.server.Kernel.run][].

    Returns:
        Response (if generated).
    """
    op = self.ops.get(opname)
    if op:
        return op.run(ctxt, req)

state()

Get object state.

Returns:

Type Description
State

State object.
Source code in penvm/src/kernels/penvm/kernels/base/server.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
def state(self) -> "State":
    """Get object state.

    Returns:
        `State` object.
    """
    try:
        opnames = sorted(list(self.ops.keys()))
        return State(
            "kernel",
            self.name,
            {
                "bases": [getattr(cls, "name") for cls in self.get_bases() if cls != object],
                "nops": len(opnames),
                "ops": opnames,
                "updated": self.updated,
            },
        )
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")

unregister(opname)

Unregister operation.

Parameters:

Name Type Description Default
opname str

Operation name.

 required
Source code in penvm/src/kernels/penvm/kernels/base/server.py 
185
186
187
188
189
190
191
192
193
194
 
def unregister(self, opname: str):
    """Unregister operation.

    Args:
        opname: Operation name.
    """
    try:
        del self.ops[opname]
    except Exception as e:
        pass

Op

Bases: BaseObject

Encapsulates functionality for an operation.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
197
198
199
200
201
202
203
204
205
206
 
class Op(BaseObject):
    """Encapsulates functionality for an operation."""

    def __init__(self):
        """Initialize."""
        super().__init__(None, logger)

    def run(self, req: "Message"):
        """Run method."""
        pass

__init__()

Initialize.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
200
201
202
 
def __init__(self):
    """Initialize."""
    super().__init__(None, logger)

run(req)

Run method.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
204
205
206
 
def run(self, req: "Message"):
    """Run method."""
    pass

OpContext

Context to run operation.

Provides an easy way for an operation to access important and necessary objects and information.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
 
class OpContext:
    """Context to run operation.

    Provides an easy way for an operation to access important and
    necessary objects and information.
    """

    def __init__(self):
        """Initialize."""
        self.conn = None
        self.machine = None
        self.processor = None
        self.session = None

__init__()

Initialize.

Source code in penvm/src/kernels/penvm/kernels/base/server.py 
216
217
218
219
220
221
 
def __init__(self):
    """Initialize."""
    self.conn = None
    self.machine = None
    self.processor = None
    self.session = None
You are on penvm.dev