k1lib.callbacks module

callbacks module

Bare example of how this module works:

import k1lib

class CbA(k1lib.Callback):
    def __init__(self):
        super().__init__()
        self.initialState = 3
    def startBatch(self):
        print("startBatch - CbA")
    def startPass(self):
        print("startPass - CbA")

class CbB(k1lib.Callback):
    def startBatch(self):
        print("startBatch - CbB")
    def endLoss(self):
        print("endLoss - CbB")

# initialization
cbs = k1lib.Callbacks()
cbs.add(CbA()).add(CbB())
model = lambda xb: xb + 3
lossF = lambda y, yb: y - yb

# training loop
cbs("startBatch"); xb = 6; yb = 2
cbs("startPass"); y = model(xb); cbs("endPass")
cbs("startLoss"); loss = lossF(y, yb); cbs("endLoss")
cbs("endBatch")

print(cbs.CbA) # can reference the Callback object directly

So, point is, you can define lots of Callback classes that defines a number of checkpoint functions, like startBatch. Then, you can create a Callbacks object that includes Callback objects. When you do cbs("checkpoint"), this will execute cb.checkpoint() of all the Callback objects.

Pretty much everything here is built upon this. The core training loop has nothing to do with ML stuff. In fact, it’s just a bunch of cbs("...") statements. Everything meaningful about the training loop comes from different Callback classes. Advantage of this is that you can tack on wildly different functions, have them play nicely with each other, and remove entire complex functionalities by commenting out a single line.

k1lib.callbacks.callbacks.Cbs

Object of all available callbacks. Automatically exposed, so you can do this:

from k1lib.imports import *
cbs = k1lib.Callbacks().add(Cbs.Beep())
cbs("endRun") # plays a beep sound

class k1lib.callbacks.callbacks.Callback[source]

Bases: object

Represents a callback. Define specific functions inside to intercept certain parts of the training loop. Can access k1lib.Learner like this:

self.l.xb = self.l.xb[None]

This takes x batch of learner, unsqueeze it at the 0 position, then sets the x batch again.

Normally, you will define a subclass of this and define specific intercept functions, but if you want to create a throwaway callback, then do this:

Callback().withCheckpoint("startRun", lambda: print("start running"))

You can use Cbs (automatically exposed) for a list of default Callback classes, for any particular needs.

order

You can also use .order to set the order of execution of the callback. The higher, the later it gets executed. Value suggestions:

7: pre-default runs, like LossLandscape
10: default runs, like DontTrainValid
13: custom mods, like ModifyBatch
15: pre-recording mod
17: recording mods, like Profiler.memory
20: default recordings, like Loss
23: post-default recordings, like ParamFinder
25: guards, like TimeLimit, CancelOnExplosion

Just leave as default (10) if you don’t know what values to choose.

dependsOn

If you’re going to extend this class, you can also specify dependencies like this:

class CbC(k1lib.Callback):
    def __init__(self):
        super().__init__()
        self.dependsOn = {"Loss", "Accuracy"}

This is so that if somewhere, Loss callback class is temporarily suspended, then CbC will be suspended also, therefore avoiding errors.

Suspension

If your Callback is mainly dormant, then you can do something like this:

class CbD(k1lib.Callback):
    def __init__(self):
        super().__init__()
        self.suspended = True
    def startBatch(self):
        # these types of methods will only execute
        # if ``self.suspended = False``
        pass
    def analyze(self):
        self.suspended = False
        # do something that sometimes call ``startBatch``
        self.suspended = True

cbs = k1lib.Callbacks().add(CbD())
# dormant phase:
cbs("startBatch") # does not execute CbD.startBatch()
# active phase
cbs.CbB.analyze() # does execute CbD.startBatch()

So yeah, you can easily make every checkpoint active/dormant by changing a single variable, how convenient. See over Callbacks.suspend() for more.

suspend()[source]: Checkpoint, called when the Callback is temporarily suspended. Overridable

restore()[source]: Checkpoint, called when the Callback is back from suspension. Overridable

withCheckpoint(checkpoint: str, f: Callable[[Callback], None])[source]

Quickly set a checkpoint, for simple, inline-able functions

Parameters

checkpoint – checkpoints like “startRun”
f – function that takes in the Callback itself

attached()[source]: Called when this is added to a Callback. Overrides this to do custom stuff when this happens.

detach()[source]: Detaches from the parent Callbacks

class k1lib.callbacks.callbacks.Timings[source]

Bases: object

List of checkpoint timings. Not intended to be instantiated by the end user. Used within Callbacks, accessible via Callbacks.timings to record time taken to execute a single checkpoint. This is useful for profiling stuff.

property state

property checkpoints: List[str]: List of all checkpoints encountered

plot()[source]: Plot all checkpoints’ execution times

clear()[source]: Clears all timing data

class k1lib.callbacks.callbacks.Callbacks[source]

Bases: object

property timings: Timings: Returns Timings object

property l: Learner: k1lib.Learner object. Will be set automatically when you set k1lib.Learner.cbs to this Callbacks

property cbs: List[Callback]: List of Callback

add(cb: Callback, name: Optional[str] = None)[source]

Adds a callback to the collection. Example:

cbs = k1lib.Callbacks()
cbs.add(k1lib.Callback().withCheckpoint("startBatch", lambda self: print("here")))

If you just want to insert a simple callback with a single checkpoint, then you can do something like:

cbs.add(["startBatch", lambda _: print("here")])

__contains__(e: str) → bool[source]: Whether a specific Callback name is in this Callback.

remove(*names: List[str])[source]: Removes a callback from the collection.

removePrefix(prefix: str)[source]: Removes any callback with the specified prefix

__call__(*checkpoints: List[str]) → bool[source]: Calls a number of checkpoints one after another. Returns True if any of the checkpoints return anything at all

__getitem__(idx: Union[int, str]) → Callback[source]

Get specific cbs.

Parameters: idx – if str, then get the Callback with this specific name, if int, then get the Callback in that index.

__iter__() → Iterator[Callback][source]: Iterates through all Callback.

__len__()[source]: How many Callback are there in total?

withBasics()[source]: Adds a bunch of very basic Callbacks that’s needed for everything. Also includes Callbacks that are not necessary, but don’t slow things down

withQOL()[source]: Adds quality of life Callbacks.

withAdvanced()[source]: Adds advanced Callbacks that do fancy stuff, but may slow things down if not configured specifically.

checkpointGraph(highlightCb: Optional[Union[str, Callback]] = None)

Graphs what checkpoints follows what checkpoints. Has to run at least once first. Requires graphviz package though. Example:

cbs = Callbacks()
cbs("a", "b", "c", "d", "b")
cbs.checkpointGraph() # returns graph object. Will display image if using notebooks

Parameters: highlightCb – if available, will highlight the checkpoints the callback uses. Can be name/class-name/class/self of callback.

context(*initCbs: List[Callback]) → ContextManager

Add context.

Works like this:

cbs = k1lib.Callbacks().add(CbA())
# CbA is available
with cbs.context(CbE(), CbF()):
    cbs.add(CbB())
    # CbA, CbB, CbE and CbF available
    cbs.add(CbC())
    # all 5 are available
# only CbA is available

For maximum shortness, you can do this:

with k1lib.Callbacks().context(CbA()) as cbs:
    # Cba is available

suspend(*cbNames: List[str]) → ContextManager

Creates suspension context for specified Callbacks. Matches callbacks with their name. Works like this:

cbs = k1lib.Callbacks().add(CbA()).add(CbB()).add(CbC())
with cbs.suspend("CbA", "CbC"):
    pass # inside here, only CbB will be active, and its checkpoints executed
# CbA, CbB and CbC are all active

confusionMatrix module

class k1lib.callbacks.confusionMatrix.ConfusionMatrix(categories: ~typing.Optional[~typing.List[str]] = None, condF: ~typing.Callable[[~k1lib.callbacks.confusionMatrix.ConfusionMatrix], bool] = <function ConfusionMatrix.<lambda>>)[source]

k1lib.callbacks module

callbacks module

confusionMatrix module

core module

hookModule module

hookParam module

limits module

landscape module

loss_accuracy module

paramFinder module

profiler module

progress module

recorder module

shorts module