Cheatsheet¶
One page of the syntax you'll reach for daily. For explanations, follow the section links.
Entry — guide¶
from srforge.data import Entry
entry = Entry(name="scene_01", image=t, target=u) # create (kwargs only!)
entry["image"] / entry.image # read (both styles)
entry["edges"] = e / entry.edges = e # write
"mask" in entry; del entry["mask"]; entry.keys() # check / delete / list
entry = entry.to("cuda") # move ALL tensors (returns new)
entry = entry.numpy() # tensors → numpy (returns new)
batch = Entry.collate([e1, e2, e3]) # batch (adds dim 0)
batch.is_batched; batch.batch_size; len(batch) # True / 3 / 3
sample = batch[0] # unbatch one (removes dim)
sub = batch[0:2] # sub-batch (keeps dim)
samples = batch.unbatch() # list of unbatched
Single samples use natural shapes — [C, H, W], bare strings, flat lists. Collation adds the batch dim (rules per type).
IO binding — guide¶
component.set_io({
"inputs": {param_name: entry_field, ...}, # LHS = your code, RHS = your data
"outputs": "field" | ["f1", "f2"], # where results go
})
| Component | Inputs from | Outputs | Notes |
|---|---|---|---|
| Model | forward() params |
required | can't overwrite existing fields |
| DataTransform | transform() params |
optional (in-place default) | annotation controls dict recursion |
| EntryTransform | constructor _key params |
— | no io: — keys go in params: |
| Loss | calculate_score() params |
— (returns MetricScores) | identity default if set_io skipped |
Rule: if you map it, the field must exist (defaults apply only to unmapped params).
YAML config — guide¶
component:
_target: srforge.models.basic.Bicubic # class (import path or registry name)
params: # constructor kwargs
scale: 3
io: # field binding (Model/DataTransform/Loss)
inputs: {image: lr}
outputs: sr
optimizer:
params:
params: ${ref:model}.trainable_params() # ${ref:...} = the INSTANTIATED object
lr: ${training.lr} # ${...} = raw config value
CLI overrides (Hydra): python train.py training.batch_size=32 'system.device=[0,1]'
Flow DSL (SequentialModel) — guide¶
x -> module -> y # single in/out
(x, g) -> module -> (out, conf) # multi, positional by signature
(image=x, guide=g) -> module -> out # named parameter mapping
-> entry_transform -> # EntryTransform: opaque step (keys in constructor)
Same instance on several lines = shared weights, different fields.
Losses — guide¶
loss:
_target: srforge.loss.LossCombiner
params:
losses:
- _target: srforge.loss.metrics.L1
params: {weight: 1.0}
io: {inputs: {x: sr, y: hr, y_mask: mask}} # y_mask optional
- _target: srforge.loss.metrics.PSNR
params: {weight: 0.0} # weight 0 = track only
io: {inputs: {x: sr, y: hr}}
scores = loss(entry) # MetricScores
scores.total_weighted().mean().backward() # the backprop chain
scores.as_raw_flat_dict() # {name: Tensor[B]} for logging
Custom loss = override pointwise(x, y) → per-element map; framework masks + reduces (two levels). Wrappers: UncertaintyLoss + Regularizer, CorrectedLoss.
Datasets — guide¶
class MyDataset(Dataset):
def __init__(self, root, **kwargs):
super().__init__(**kwargs) # REQUIRED: forwards name/transforms/cache_dir/recache
...
def __len__(self): ...
def __getitem__(self, idx) -> Entry:
return Entry(name=..., image=...) # natural shapes, always set name
ds.take(100); ds.shuffle(seed=42); ds.filter(pred) # → SubsetDataset
ds_a + ds_b # → ConcatDataset (needs names)
ds.cache("/tmp/cache") # disk-cache transformed entries
PatchedDataset(ds, field_sizes={"image": 32}) # split into patches
Built-ins: decision table.
Hooks — guide¶
training_runner:
params:
hooks:
- {_target: srforge.training.hooks.ProgressBar, params: {name: "T"}}
- {_target: srforge.training.hooks.GradientClip, params: {max_norm: 1.0}}
validation_runner:
params:
hooks:
- {_target: srforge.training.hooks.ProgressBar, params: {name: "V"}}
- _target: srforge.training.hooks.BatchImageLogger
params: {batch_id: 0, img_key: sr, tracker: ${ref:tracker}}
trainer:
params:
hooks:
- {_target: srforge.training.hooks.LossLogger, params: {tracker: ${ref:tracker}}}
- {_target: srforge.training.hooks.PyTorchModelSaver, params: {tracker: ${ref:tracker}}}
Attachment is the scope — runner hooks fire during that runner's epochs, trainer hooks at trainer-level points. Custom hook:
class MyHook(Hook):
@hooks_into("on_epoch_end") # HookPoint name on runner/trainer
def handler(self, ctx): ... # ctx = mutable Context
(Legacy observers: + EventBus: deprecated, removal in v0.16.0 — porting guide.)
Script skeleton — guide¶
resolve = init(cfg) # pre-flight audit + resolver
model = resolve(cfg.model)
loss = resolve(cfg.loss)
trainer = resolve(cfg.trainer) # hooks attach here, from YAML
tracker = resolve(cfg.tracker)
ckpt = resume_from_checkpoint(model, trainer.training_runner, scheduler, tracker=tracker)
trainer.restore(ckpt) # None-safe
trainer.train(cfg.training.epochs, train_loader, val_loader)
tracker.finish(0)
Common errors — debugging FAQ¶
| Error | Meaning | Fix |
|---|---|---|
KeyError: field 'x' not found |
IO binding maps to a missing field | check binding vs entry.keys() |
KeyError: attempted to overwrite existing entry fields |
model output name collides | change outputs: |
TypeError ... @audit_subclasses (class def) |
Loss/Dataset subclass drops framework kwargs | add **kwargs + super().__init__(**kwargs) |
TypeError: unexpected keyword argument 'cache_dir' |
subclass doesn't accept framework kwargs | same fix as above |
Parameter 'x' ... must have an annotated type |
Loss calculate_score param lacks annotation |
annotate every param |
DeprecationWarning: ... Observer |
legacy observers in use | port to Hooks |