Skip to content

input

input ¤

TorchBinomialLayer ¤

Bases: TorchExpFamilyLayer

The Binomial distribution layer.

This is fully factorized down to univariate Binomial distributions.

Source code in cirkit/backend/torch/layers/input.py
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
class TorchBinomialLayer(TorchExpFamilyLayer):
    """The Binomial distribution layer.

    This is fully factorized down to univariate Binomial distributions.
    """

    # DISABLE: It's designed to have these arguments.
    # pylint: disable-next=too-many-arguments
    def __init__(
        self,
        scope_idx: Tensor,
        num_output_units: int,
        *,
        num_channels: int = 1,
        total_count: int = 1,
        probs: TorchParameter | None = None,
        logits: TorchParameter | None = None,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initialize a Binomial layer.

        Args:
            scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
                $D$ is the number of variables on which the input layers in each fold are defined on.
                Alternatively, a tensor of shape $(D,)$ can be specified, which will be interpreted
                as a tensor of shape $(1, D)$, i.e., with $F = 1$.
            num_output_units: The number of output units.
            num_channels: The number of channels.
            total_count: The number of trials.
            probs: The probabilities parameter of shape $(F, K, C)$, where $K$ is the number of
                output units, and $C$ is the number of channels.
            logits: The logits parameter of shape $(F, K, C)$, where $K$ is the number of
                output units, and $C$ is the number of channels.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the scope contains more than one variable.
            ValueError: If the total count is not positive.
            ValueError: If both the probs and logits parameters are provided, or none of them.
            ValueError: If the parameter's shape is incorrect.
        """
        num_variables = scope_idx.shape[-1]
        if num_variables != 1:
            raise ValueError("The Binomial layer encodes a univariate distribution")
        if total_count < 0:
            raise ValueError("The number of trials must be non-negative")
        super().__init__(
            scope_idx,
            num_output_units,
            num_channels=num_channels,
            semiring=semiring,
        )
        self.total_count = total_count
        if not ((logits is None) ^ (probs is None)):
            raise ValueError("Exactly one between 'logits' and 'probs' must be specified")
        if logits is None:
            assert probs is not None
            if not self._valid_parameter_shape(probs):
                raise ValueError(
                    f"Expected number of folds {self.num_folds} "
                    f"and shape {self._probs_logits_shape} for 'probs', found"
                    f"{probs.num_folds} and {probs.shape}, respectively"
                )
        else:
            if not self._valid_parameter_shape(logits):
                raise ValueError(
                    f"Expected number of folds {self.num_folds} "
                    f"and shape {self._probs_logits_shape} for 'logits', found"
                    f"{logits.num_folds} and {logits.shape}, respectively"
                )
        self.probs = probs
        self.logits = logits

    def _valid_parameter_shape(self, p: TorchParameter) -> bool:
        if p.num_folds != self.num_folds:
            return False
        return p.shape == self._probs_logits_shape

    @property
    def _probs_logits_shape(self) -> tuple[int, ...]:
        return self.num_output_units, self.num_channels

    @property
    def config(self) -> Mapping[str, Any]:
        return {
            "num_output_units": self.num_output_units,
            "num_channels": self.num_channels,
            "total_count": self.total_count,
        }

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        if self.logits is None:
            return {"probs": self.probs}
        return {"logits": self.logits}

    def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
        if x.is_floating_point():
            x = x.long()  # The input to Binomial should be discrete
        x = x.permute(0, 2, 3, 1)  # (F, C, B, 1) -> (F, B, 1, C)
        if self.logits is not None:
            logits = self.logits().unsqueeze(dim=1)  # (F, 1, K, C)
            dist = distributions.Binomial(self.total_count, logits=logits)
        else:
            probs = self.probs().unsqueeze(dim=1)  # (F, 1, K, C)
            dist = distributions.Binomial(self.total_count, probs=probs)
        x = dist.log_prob(x)  # (F, B, K, C)
        return torch.sum(x, dim=3)

    def log_partition_function(self) -> Tensor:
        device = self.logits.device if self.logits is not None else self.probs.device
        return torch.zeros(size=(self.num_folds, 1, self.num_output_units), device=device)

    def sample(self, num_samples: int = 1) -> Tensor:
        logits = torch.log(self.probs()) if self.logits is None else self.logits()
        dist = distributions.Binomial(self.total_count, logits=logits)
        samples = dist.sample((num_samples,))  # (num_samples, F, K, C)
        samples = samples.permute(1, 3, 2, 0)  # (F, C, K, num_samples)
        return samples

_probs_logits_shape property ¤

config property ¤

logits = logits instance-attribute ¤

params property ¤

probs = probs instance-attribute ¤

total_count = total_count instance-attribute ¤

__init__(scope_idx, num_output_units, *, num_channels=1, total_count=1, probs=None, logits=None, semiring=None) ¤

Initialize a Binomial layer.

Parameters:

Name Type Description Default
scope_idx Tensor

A tensor of shape \((F, D)\), where \(F\) is the number of folds, and \(D\) is the number of variables on which the input layers in each fold are defined on. Alternatively, a tensor of shape \((D,)\) can be specified, which will be interpreted as a tensor of shape \((1, D)\), i.e., with \(F = 1\).

required
num_output_units int

The number of output units.

required
num_channels int

The number of channels.

1
total_count int

The number of trials.

1
probs TorchParameter | None

The probabilities parameter of shape \((F, K, C)\), where \(K\) is the number of output units, and \(C\) is the number of channels.

None
logits TorchParameter | None

The logits parameter of shape \((F, K, C)\), where \(K\) is the number of output units, and \(C\) is the number of channels.

None
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the scope contains more than one variable.

ValueError

If the total count is not positive.

ValueError

If both the probs and logits parameters are provided, or none of them.

ValueError

If the parameter's shape is incorrect.

Source code in cirkit/backend/torch/layers/input.py
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def __init__(
    self,
    scope_idx: Tensor,
    num_output_units: int,
    *,
    num_channels: int = 1,
    total_count: int = 1,
    probs: TorchParameter | None = None,
    logits: TorchParameter | None = None,
    semiring: Semiring | None = None,
) -> None:
    r"""Initialize a Binomial layer.

    Args:
        scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
            $D$ is the number of variables on which the input layers in each fold are defined on.
            Alternatively, a tensor of shape $(D,)$ can be specified, which will be interpreted
            as a tensor of shape $(1, D)$, i.e., with $F = 1$.
        num_output_units: The number of output units.
        num_channels: The number of channels.
        total_count: The number of trials.
        probs: The probabilities parameter of shape $(F, K, C)$, where $K$ is the number of
            output units, and $C$ is the number of channels.
        logits: The logits parameter of shape $(F, K, C)$, where $K$ is the number of
            output units, and $C$ is the number of channels.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the scope contains more than one variable.
        ValueError: If the total count is not positive.
        ValueError: If both the probs and logits parameters are provided, or none of them.
        ValueError: If the parameter's shape is incorrect.
    """
    num_variables = scope_idx.shape[-1]
    if num_variables != 1:
        raise ValueError("The Binomial layer encodes a univariate distribution")
    if total_count < 0:
        raise ValueError("The number of trials must be non-negative")
    super().__init__(
        scope_idx,
        num_output_units,
        num_channels=num_channels,
        semiring=semiring,
    )
    self.total_count = total_count
    if not ((logits is None) ^ (probs is None)):
        raise ValueError("Exactly one between 'logits' and 'probs' must be specified")
    if logits is None:
        assert probs is not None
        if not self._valid_parameter_shape(probs):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._probs_logits_shape} for 'probs', found"
                f"{probs.num_folds} and {probs.shape}, respectively"
            )
    else:
        if not self._valid_parameter_shape(logits):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._probs_logits_shape} for 'logits', found"
                f"{logits.num_folds} and {logits.shape}, respectively"
            )
    self.probs = probs
    self.logits = logits

_valid_parameter_shape(p) ¤

Source code in cirkit/backend/torch/layers/input.py
535
536
537
538
def _valid_parameter_shape(self, p: TorchParameter) -> bool:
    if p.num_folds != self.num_folds:
        return False
    return p.shape == self._probs_logits_shape

log_partition_function() ¤

Source code in cirkit/backend/torch/layers/input.py
571
572
573
def log_partition_function(self) -> Tensor:
    device = self.logits.device if self.logits is not None else self.probs.device
    return torch.zeros(size=(self.num_folds, 1, self.num_output_units), device=device)

log_unnormalized_likelihood(x) ¤

Source code in cirkit/backend/torch/layers/input.py
558
559
560
561
562
563
564
565
566
567
568
569
def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
    if x.is_floating_point():
        x = x.long()  # The input to Binomial should be discrete
    x = x.permute(0, 2, 3, 1)  # (F, C, B, 1) -> (F, B, 1, C)
    if self.logits is not None:
        logits = self.logits().unsqueeze(dim=1)  # (F, 1, K, C)
        dist = distributions.Binomial(self.total_count, logits=logits)
    else:
        probs = self.probs().unsqueeze(dim=1)  # (F, 1, K, C)
        dist = distributions.Binomial(self.total_count, probs=probs)
    x = dist.log_prob(x)  # (F, B, K, C)
    return torch.sum(x, dim=3)

sample(num_samples=1) ¤

Source code in cirkit/backend/torch/layers/input.py
575
576
577
578
579
580
def sample(self, num_samples: int = 1) -> Tensor:
    logits = torch.log(self.probs()) if self.logits is None else self.logits()
    dist = distributions.Binomial(self.total_count, logits=logits)
    samples = dist.sample((num_samples,))  # (num_samples, F, K, C)
    samples = samples.permute(1, 3, 2, 0)  # (F, C, K, num_samples)
    return samples

TorchCategoricalLayer ¤

Bases: TorchExpFamilyLayer

The Categorical distribution layer, parameterized by either probabilities or logits.

Source code in cirkit/backend/torch/layers/input.py
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
class TorchCategoricalLayer(TorchExpFamilyLayer):
    """The Categorical distribution layer, parameterized by either probabilities or logits."""

    # pylint: disable-next=too-many-arguments
    def __init__(
        self,
        scope_idx: Tensor,
        num_output_units: int,
        num_channels: int = 1,
        *,
        num_categories: int = 2,
        probs: TorchParameter | None = None,
        logits: TorchParameter | None = None,
        semiring: Semiring | None = None,
    ) -> None:
        """Initialize a Categorical layer.

        Args:
            scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
                $D$ is the number of variables on which the input layers in each fold are defined on.
                Alternatively, a tensor of shape $(D,)$ can be specified, which will be interpreted
                as a tensor of shape $(1, D)$, i.e., with $F = 1$.
            num_output_units: The number of output units.
            num_channels: The number of channels.
            num_categories: The number of categories for Categorical distribution.
            probs: The probabilities parameter of shape $(F, K, C, V)$, where $K$ is the number of
                output units, $C$ is the number of channels, and $V$ is the number of categories.
            logits: The logits parameter of shape $(F, K, C, V)$, where $K$ is the number of
                output units, $C$ is the number of channels, and $V$ is the number of categories.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the scope contains more than one variable.
            ValueError: If the number of categories is negative.
            ValueError: If both the probs and logits parameters are provided, or none of them.
            ValueError: If the parameter's shape is incorrect.
        """
        num_variables = scope_idx.shape[-1]
        if num_variables != 1:
            raise ValueError("The Gaussian layer encodes a univariate distribution")
        if num_categories <= 0:
            raise ValueError(
                "The number of categories for Categorical distribution must be positive"
            )
        super().__init__(
            scope_idx,
            num_output_units,
            num_channels=num_channels,
            semiring=semiring,
        )
        self.num_categories = num_categories
        if not ((logits is None) ^ (probs is None)):
            raise ValueError("Exactly one between 'logits' and 'probs' must be specified")
        if logits is None:
            assert probs is not None
            if not self._valid_parameter_shape(probs):
                raise ValueError(
                    f"Expected number of folds {self.num_folds} "
                    f"and shape {self._probs_logits_shape} for 'probs', found"
                    f"{probs.num_folds} and {probs.shape}, respectively"
                )
        elif not self._valid_parameter_shape(logits):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._probs_logits_shape} for 'logits', found"
                f"{logits.num_folds} and {logits.shape}, respectively"
            )
        self.probs = probs
        self.logits = logits

    def _valid_parameter_shape(self, p: TorchParameter) -> bool:
        if p.num_folds != self.num_folds:
            return False
        return p.shape == self._probs_logits_shape

    @property
    def _probs_logits_shape(self) -> tuple[int, ...]:
        return self.num_output_units, self.num_channels, self.num_categories

    @property
    def config(self) -> Mapping[str, Any]:
        return {
            "num_output_units": self.num_output_units,
            "num_channels": self.num_channels,
            "num_categories": self.num_categories,
        }

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        if self.logits is None:
            return {"probs": self.probs}
        return {"logits": self.logits}

    def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
        if x.is_floating_point():
            x = x.long()  # The input to Categorical should be discrete
        # x: (F, C, B, 1) -> (F, C, B)
        x = x.squeeze(dim=3)
        # logits: (F, K, C, N)
        logits = torch.log(self.probs()) if self.logits is None else self.logits()
        if self.num_channels == 1:
            idx_fold = torch.arange(self.num_folds)
            x = logits[:, :, 0][idx_fold[:, None], :, x[:, 0]]
        else:
            idx_fold = torch.arange(self.num_folds)[:, None, None]
            idx_channel = torch.arange(self.num_channels)[None, :, None]
            x = torch.sum(logits[idx_fold, :, idx_channel, x], dim=1)
        return x

    def log_partition_function(self) -> Tensor:
        if self.logits is None:
            return torch.zeros(
                size=(self.num_folds, 1, self.num_output_units), device=self.probs.device
            )
        logits = self.logits()
        return torch.sum(torch.logsumexp(logits, dim=3), dim=2).unsqueeze(dim=1)

    def sample(self, num_samples: int = 1) -> Tensor:
        logits = torch.log(self.probs()) if self.logits is None else self.logits()
        dist = distributions.Categorical(logits=logits)
        samples = dist.sample((num_samples,))  # (N, F, K, C)
        samples = samples.permute(1, 3, 2, 0)  # (F, C, K, N)
        return samples

_probs_logits_shape property ¤

config property ¤

logits = logits instance-attribute ¤

num_categories = num_categories instance-attribute ¤

params property ¤

probs = probs instance-attribute ¤

__init__(scope_idx, num_output_units, num_channels=1, *, num_categories=2, probs=None, logits=None, semiring=None) ¤

Initialize a Categorical layer.

Parameters:

Name Type Description Default
scope_idx Tensor

A tensor of shape \((F, D)\), where \(F\) is the number of folds, and \(D\) is the number of variables on which the input layers in each fold are defined on. Alternatively, a tensor of shape \((D,)\) can be specified, which will be interpreted as a tensor of shape \((1, D)\), i.e., with \(F = 1\).

required
num_output_units int

The number of output units.

required
num_channels int

The number of channels.

1
num_categories int

The number of categories for Categorical distribution.

2
probs TorchParameter | None

The probabilities parameter of shape \((F, K, C, V)\), where \(K\) is the number of output units, \(C\) is the number of channels, and \(V\) is the number of categories.

None
logits TorchParameter | None

The logits parameter of shape \((F, K, C, V)\), where \(K\) is the number of output units, \(C\) is the number of channels, and \(V\) is the number of categories.

None
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the scope contains more than one variable.

ValueError

If the number of categories is negative.

ValueError

If both the probs and logits parameters are provided, or none of them.

ValueError

If the parameter's shape is incorrect.

Source code in cirkit/backend/torch/layers/input.py
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
def __init__(
    self,
    scope_idx: Tensor,
    num_output_units: int,
    num_channels: int = 1,
    *,
    num_categories: int = 2,
    probs: TorchParameter | None = None,
    logits: TorchParameter | None = None,
    semiring: Semiring | None = None,
) -> None:
    """Initialize a Categorical layer.

    Args:
        scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
            $D$ is the number of variables on which the input layers in each fold are defined on.
            Alternatively, a tensor of shape $(D,)$ can be specified, which will be interpreted
            as a tensor of shape $(1, D)$, i.e., with $F = 1$.
        num_output_units: The number of output units.
        num_channels: The number of channels.
        num_categories: The number of categories for Categorical distribution.
        probs: The probabilities parameter of shape $(F, K, C, V)$, where $K$ is the number of
            output units, $C$ is the number of channels, and $V$ is the number of categories.
        logits: The logits parameter of shape $(F, K, C, V)$, where $K$ is the number of
            output units, $C$ is the number of channels, and $V$ is the number of categories.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the scope contains more than one variable.
        ValueError: If the number of categories is negative.
        ValueError: If both the probs and logits parameters are provided, or none of them.
        ValueError: If the parameter's shape is incorrect.
    """
    num_variables = scope_idx.shape[-1]
    if num_variables != 1:
        raise ValueError("The Gaussian layer encodes a univariate distribution")
    if num_categories <= 0:
        raise ValueError(
            "The number of categories for Categorical distribution must be positive"
        )
    super().__init__(
        scope_idx,
        num_output_units,
        num_channels=num_channels,
        semiring=semiring,
    )
    self.num_categories = num_categories
    if not ((logits is None) ^ (probs is None)):
        raise ValueError("Exactly one between 'logits' and 'probs' must be specified")
    if logits is None:
        assert probs is not None
        if not self._valid_parameter_shape(probs):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._probs_logits_shape} for 'probs', found"
                f"{probs.num_folds} and {probs.shape}, respectively"
            )
    elif not self._valid_parameter_shape(logits):
        raise ValueError(
            f"Expected number of folds {self.num_folds} "
            f"and shape {self._probs_logits_shape} for 'logits', found"
            f"{logits.num_folds} and {logits.shape}, respectively"
        )
    self.probs = probs
    self.logits = logits

_valid_parameter_shape(p) ¤

Source code in cirkit/backend/torch/layers/input.py
406
407
408
409
def _valid_parameter_shape(self, p: TorchParameter) -> bool:
    if p.num_folds != self.num_folds:
        return False
    return p.shape == self._probs_logits_shape

log_partition_function() ¤

Source code in cirkit/backend/torch/layers/input.py
445
446
447
448
449
450
451
def log_partition_function(self) -> Tensor:
    if self.logits is None:
        return torch.zeros(
            size=(self.num_folds, 1, self.num_output_units), device=self.probs.device
        )
    logits = self.logits()
    return torch.sum(torch.logsumexp(logits, dim=3), dim=2).unsqueeze(dim=1)

log_unnormalized_likelihood(x) ¤

Source code in cirkit/backend/torch/layers/input.py
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
    if x.is_floating_point():
        x = x.long()  # The input to Categorical should be discrete
    # x: (F, C, B, 1) -> (F, C, B)
    x = x.squeeze(dim=3)
    # logits: (F, K, C, N)
    logits = torch.log(self.probs()) if self.logits is None else self.logits()
    if self.num_channels == 1:
        idx_fold = torch.arange(self.num_folds)
        x = logits[:, :, 0][idx_fold[:, None], :, x[:, 0]]
    else:
        idx_fold = torch.arange(self.num_folds)[:, None, None]
        idx_channel = torch.arange(self.num_channels)[None, :, None]
        x = torch.sum(logits[idx_fold, :, idx_channel, x], dim=1)
    return x

sample(num_samples=1) ¤

Source code in cirkit/backend/torch/layers/input.py
453
454
455
456
457
458
def sample(self, num_samples: int = 1) -> Tensor:
    logits = torch.log(self.probs()) if self.logits is None else self.logits()
    dist = distributions.Categorical(logits=logits)
    samples = dist.sample((num_samples,))  # (N, F, K, C)
    samples = samples.permute(1, 3, 2, 0)  # (F, C, K, N)
    return samples

TorchConstantLayer ¤

Bases: TorchInputLayer, ABC

An input layer encoding a constant vector or, equivalently, a vector of functions defined over empty variable scopes.

Source code in cirkit/backend/torch/layers/input.py
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
195
196
197
198
199
class TorchConstantLayer(TorchInputLayer, ABC):
    """An input layer encoding a constant vector or, equivalently, a vector of functions
    defined over empty variable scopes.
    """

    def __init__(
        self,
        num_output_units: int,
        num_folds: int,
        *,
        semiring: Semiring | None = None,
    ) -> None:
        """Initializes a constant layer.

        Args:
            num_output_units: The number of output units.
            num_folds: The number of folds.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].
        """
        super().__init__(
            torch.empty(size=(num_folds, 0), dtype=torch.int64), num_output_units, semiring=semiring
        )

    def __call__(self, batch_size: int) -> Tensor:
        # IGNORE: Idiom for nn.Module.__call__.
        return super().__call__(batch_size)  # type: ignore[no-any-return,misc]

    @abstractmethod
    def forward(self, batch_size: int) -> Tensor:
        r"""Invoke the forward function.

        Args:
            batch_size: The batch size $B$ of the output tensor.

        Returns:
            Tensor: The tensor output of this layer, having shape $(F, B, K)$, where $K$
                is the number of output units, and $B$ is the batch size given as input.
        """

__call__(batch_size) ¤

Source code in cirkit/backend/torch/layers/input.py
185
186
187
def __call__(self, batch_size: int) -> Tensor:
    # IGNORE: Idiom for nn.Module.__call__.
    return super().__call__(batch_size)  # type: ignore[no-any-return,misc]

__init__(num_output_units, num_folds, *, semiring=None) ¤

Initializes a constant layer.

Parameters:

Name Type Description Default
num_output_units int

The number of output units.

required
num_folds int

The number of folds.

required
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None
Source code in cirkit/backend/torch/layers/input.py
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def __init__(
    self,
    num_output_units: int,
    num_folds: int,
    *,
    semiring: Semiring | None = None,
) -> None:
    """Initializes a constant layer.

    Args:
        num_output_units: The number of output units.
        num_folds: The number of folds.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].
    """
    super().__init__(
        torch.empty(size=(num_folds, 0), dtype=torch.int64), num_output_units, semiring=semiring
    )

forward(batch_size) abstractmethod ¤

Invoke the forward function.

Parameters:

Name Type Description Default
batch_size int

The batch size \(B\) of the output tensor.

required

Returns:

Name Type Description
Tensor Tensor

The tensor output of this layer, having shape \((F, B, K)\), where \(K\) is the number of output units, and \(B\) is the batch size given as input.

Source code in cirkit/backend/torch/layers/input.py
189
190
191
192
193
194
195
196
197
198
199
@abstractmethod
def forward(self, batch_size: int) -> Tensor:
    r"""Invoke the forward function.

    Args:
        batch_size: The batch size $B$ of the output tensor.

    Returns:
        Tensor: The tensor output of this layer, having shape $(F, B, K)$, where $K$
            is the number of output units, and $B$ is the batch size given as input.
    """

TorchConstantValueLayer ¤

Bases: TorchConstantLayer

An input layer having empty scope and computing a constant value.

Source code in cirkit/backend/torch/layers/input.py
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
class TorchConstantValueLayer(TorchConstantLayer):
    """An input layer having empty scope and computing a constant value."""

    def __init__(
        self,
        num_output_units: int,
        *,
        log_space: bool = False,
        value: TorchParameter,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initialize a constant value input layer.

        Args:
            num_output_units: The number of output units.
            log_space: Whether the given value is in the log-space, i.e., this constant
                layer should encode functions $\exp(x)$ rather than just x.
            value: The tensor value encoded by the layer, given by a parameter of shape $(F, K)$,
                where $F$ is the number of folds and $K$ is the numer of output units.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the number of folds of the shape of the given value is incorrect.
        """
        super().__init__(
            num_output_units,
            value.num_folds,
            semiring=semiring,
        )
        if value.num_folds != self.num_folds:
            raise ValueError(
                f"The value must have number of folds {self.num_folds}, "
                f"but found {value.num_folds}"
            )
        if value.shape != (num_output_units,):
            raise ValueError(
                f"The shape of the value must be ({num_output_units},), " f"but found {value.shape}"
            )
        self.value = value
        self.log_space = log_space
        self._source_semiring = LSESumSemiring if log_space else SumProductSemiring

    @property
    def config(self) -> Mapping[str, Any]:
        return {"num_output_units": self.num_output_units, "log_space": self.log_space}

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        return {"value": self.value}

    def forward(self, batch_size: int) -> Tensor:
        value = self.value()  # (F, Ko)
        # value: (F, B, Ko)
        value = value.unsqueeze(dim=1).expand(value.shape[0], batch_size, value.shape[1])
        return self.semiring.map_from(value, self._source_semiring)

_source_semiring = LSESumSemiring if log_space else SumProductSemiring instance-attribute ¤

config property ¤

log_space = log_space instance-attribute ¤

params property ¤

value = value instance-attribute ¤

__init__(num_output_units, *, log_space=False, value, semiring=None) ¤

Initialize a constant value input layer.

Parameters:

Name Type Description Default
num_output_units int

The number of output units.

required
log_space bool

Whether the given value is in the log-space, i.e., this constant layer should encode functions \(\exp(x)\) rather than just x.

False
value TorchParameter

The tensor value encoded by the layer, given by a parameter of shape \((F, K)\), where \(F\) is the number of folds and \(K\) is the numer of output units.

required
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the number of folds of the shape of the given value is incorrect.

Source code in cirkit/backend/torch/layers/input.py
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
def __init__(
    self,
    num_output_units: int,
    *,
    log_space: bool = False,
    value: TorchParameter,
    semiring: Semiring | None = None,
) -> None:
    r"""Initialize a constant value input layer.

    Args:
        num_output_units: The number of output units.
        log_space: Whether the given value is in the log-space, i.e., this constant
            layer should encode functions $\exp(x)$ rather than just x.
        value: The tensor value encoded by the layer, given by a parameter of shape $(F, K)$,
            where $F$ is the number of folds and $K$ is the numer of output units.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the number of folds of the shape of the given value is incorrect.
    """
    super().__init__(
        num_output_units,
        value.num_folds,
        semiring=semiring,
    )
    if value.num_folds != self.num_folds:
        raise ValueError(
            f"The value must have number of folds {self.num_folds}, "
            f"but found {value.num_folds}"
        )
    if value.shape != (num_output_units,):
        raise ValueError(
            f"The shape of the value must be ({num_output_units},), " f"but found {value.shape}"
        )
    self.value = value
    self.log_space = log_space
    self._source_semiring = LSESumSemiring if log_space else SumProductSemiring

forward(batch_size) ¤

Source code in cirkit/backend/torch/layers/input.py
760
761
762
763
764
def forward(self, batch_size: int) -> Tensor:
    value = self.value()  # (F, Ko)
    # value: (F, B, Ko)
    value = value.unsqueeze(dim=1).expand(value.shape[0], batch_size, value.shape[1])
    return self.semiring.map_from(value, self._source_semiring)

TorchEmbeddingLayer ¤

Bases: TorchInputFunctionLayer

The embedding input layer, where each input function maps a discrete variable having finite support \(\{0,\ldots,V-1\}\) to the corresponding entry of a \(V\)-th dimensional vector.

Source code in cirkit/backend/torch/layers/input.py
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
class TorchEmbeddingLayer(TorchInputFunctionLayer):
    r"""The embedding input layer, where each input function maps a discrete variable having
    finite support $\{0,\ldots,V-1\}$ to the corresponding entry of a $V$-th dimensional vector.
    """

    def __init__(
        self,
        scope_idx: Tensor,
        num_output_units: int,
        num_channels: int = 1,
        *,
        num_states: int = 2,
        weight: TorchParameter,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initialize an embedding input layer.

        Args:
            scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
                $D$ is the number of variables on which the input layers in each fold are defined
                on. Alternatively, a tensor of shape $(D,)$ can be specified, which will be
                interpreted as a tensor of shape $(1, D)$, i.e., with $F = 1$.
            num_output_units: The number of output units.
            num_channels: The number of channels.
            num_states: The number of states $V$ each variable can assume.
            weight: The weight parameter of shape $(F, K, C, N)$, where $K$ is the number of output
                units, $C$ is the number of channels, and $V$ is the number of states.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the scope contains more than one variable.
            ValueError: If the number of states $V$ is less than 2.
            ValueError: If the parameter's shape is incorrect.
        """
        num_variables = scope_idx.shape[-1]
        if num_variables != 1:
            raise ValueError("The Embedding layer encodes univariate functions")
        if num_states <= 1:
            raise ValueError("The number of states must be at least 2")
        super().__init__(
            scope_idx,
            num_output_units,
            num_channels=num_channels,
            semiring=semiring,
        )
        self.num_states = num_states
        if not self._valid_weight_shape(weight):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._weight_shape} for 'weight', found"
                f"{weight.num_folds} and {weight.shape}, respectively"
            )
        self.weight = weight

    def _valid_weight_shape(self, p: TorchParameter) -> bool:
        if p.num_folds != self.num_folds:
            return False
        return p.shape == self._weight_shape

    @property
    def _weight_shape(self) -> tuple[int, ...]:
        return self.num_output_units, self.num_channels, self.num_states

    @property
    def config(self) -> Mapping[str, Any]:
        return {
            "num_output_units": self.num_output_units,
            "num_channels": self.num_channels,
            "num_states": self.num_states,
        }

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        return {"weight": self.weight}

    def forward(self, x: Tensor) -> Tensor:
        if x.is_floating_point():
            x = x.long()  # The input to Embedding should be discrete
        x = x.squeeze(dim=3)  # (F, C, B)
        weight = self.weight()
        if self.num_channels == 1:
            idx_fold = torch.arange(self.num_folds)
            x = weight[:, :, 0][idx_fold[:, None], :, x[:, 0]]
            x = self.semiring.map_from(x, SumProductSemiring)
        else:
            idx_fold = torch.arange(self.num_folds)[:, None, None]
            idx_channel = torch.arange(self.num_channels)[None, :, None]
            x = weight[idx_fold, :, idx_channel, x]
            x = self.semiring.map_from(x, SumProductSemiring)
            x = self.semiring.prod(x, dim=1)
        return x  # (F, B, K)

_weight_shape property ¤

config property ¤

num_states = num_states instance-attribute ¤

params property ¤

weight = weight instance-attribute ¤

__init__(scope_idx, num_output_units, num_channels=1, *, num_states=2, weight, semiring=None) ¤

Initialize an embedding input layer.

Parameters:

Name Type Description Default
scope_idx Tensor

A tensor of shape \((F, D)\), where \(F\) is the number of folds, and \(D\) is the number of variables on which the input layers in each fold are defined on. Alternatively, a tensor of shape \((D,)\) can be specified, which will be interpreted as a tensor of shape \((1, D)\), i.e., with \(F = 1\).

required
num_output_units int

The number of output units.

required
num_channels int

The number of channels.

1
num_states int

The number of states \(V\) each variable can assume.

2
weight TorchParameter

The weight parameter of shape \((F, K, C, N)\), where \(K\) is the number of output units, \(C\) is the number of channels, and \(V\) is the number of states.

required
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the scope contains more than one variable.

ValueError

If the number of states \(V\) is less than 2.

ValueError

If the parameter's shape is incorrect.

Source code in cirkit/backend/torch/layers/input.py
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def __init__(
    self,
    scope_idx: Tensor,
    num_output_units: int,
    num_channels: int = 1,
    *,
    num_states: int = 2,
    weight: TorchParameter,
    semiring: Semiring | None = None,
) -> None:
    r"""Initialize an embedding input layer.

    Args:
        scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
            $D$ is the number of variables on which the input layers in each fold are defined
            on. Alternatively, a tensor of shape $(D,)$ can be specified, which will be
            interpreted as a tensor of shape $(1, D)$, i.e., with $F = 1$.
        num_output_units: The number of output units.
        num_channels: The number of channels.
        num_states: The number of states $V$ each variable can assume.
        weight: The weight parameter of shape $(F, K, C, N)$, where $K$ is the number of output
            units, $C$ is the number of channels, and $V$ is the number of states.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the scope contains more than one variable.
        ValueError: If the number of states $V$ is less than 2.
        ValueError: If the parameter's shape is incorrect.
    """
    num_variables = scope_idx.shape[-1]
    if num_variables != 1:
        raise ValueError("The Embedding layer encodes univariate functions")
    if num_states <= 1:
        raise ValueError("The number of states must be at least 2")
    super().__init__(
        scope_idx,
        num_output_units,
        num_channels=num_channels,
        semiring=semiring,
    )
    self.num_states = num_states
    if not self._valid_weight_shape(weight):
        raise ValueError(
            f"Expected number of folds {self.num_folds} "
            f"and shape {self._weight_shape} for 'weight', found"
            f"{weight.num_folds} and {weight.shape}, respectively"
        )
    self.weight = weight

_valid_weight_shape(p) ¤

Source code in cirkit/backend/torch/layers/input.py
257
258
259
260
def _valid_weight_shape(self, p: TorchParameter) -> bool:
    if p.num_folds != self.num_folds:
        return False
    return p.shape == self._weight_shape

forward(x) ¤

Source code in cirkit/backend/torch/layers/input.py
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def forward(self, x: Tensor) -> Tensor:
    if x.is_floating_point():
        x = x.long()  # The input to Embedding should be discrete
    x = x.squeeze(dim=3)  # (F, C, B)
    weight = self.weight()
    if self.num_channels == 1:
        idx_fold = torch.arange(self.num_folds)
        x = weight[:, :, 0][idx_fold[:, None], :, x[:, 0]]
        x = self.semiring.map_from(x, SumProductSemiring)
    else:
        idx_fold = torch.arange(self.num_folds)[:, None, None]
        idx_channel = torch.arange(self.num_channels)[None, :, None]
        x = weight[idx_fold, :, idx_channel, x]
        x = self.semiring.map_from(x, SumProductSemiring)
        x = self.semiring.prod(x, dim=1)
    return x  # (F, B, K)

TorchEvidenceLayer ¤

Bases: TorchConstantLayer

The input layer computing the output of another input layer on a given observation.

Source code in cirkit/backend/torch/layers/input.py
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
class TorchEvidenceLayer(TorchConstantLayer):
    """The input layer computing the output of another input layer on a given observation."""

    def __init__(
        self,
        layer: TorchInputLayer,
        *,
        observation: TorchParameter,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initializes an evidence layer.

        Args:
            layer: The input layer on which compute the evidence of.
            observation: The observation, i.e., the input to pass to the given input layer.
                It must be a parameter of shape $(F, C, D)$, where $F$ is the number of folds
                of the given layer, $D$ is the number variables the given layer is defined on,
                and $C$ is the number channels per variable.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the number of folds or the shape of the given observation is incorrect,
                with respect to the expected input shape of the given input layer.
        """
        if observation.num_folds != layer.num_folds:
            raise ValueError(
                f"The number of folds in the observation and in the layer should be the same, "
                f"but found {observation.num_folds} and {layer.num_folds} respectively"
            )
        if len(observation.shape) != 2:
            raise ValueError(
                f"Expected observation of shape (num_channels, num_variables), "
                f"but found {observation.shape}"
            )
        num_channels, num_variables = observation.shape
        if num_channels != layer.num_channels:
            raise ValueError(
                f"Expected an observation with number of channels {layer.num_channels}, "
                f"but found {num_channels}"
            )
        if num_variables != layer.num_variables:
            raise ValueError(
                f"Expected an observation with number of variables {layer.num_variables}, "
                f"but found {num_variables}"
            )
        super().__init__(layer.num_output_units, layer.num_folds, semiring=semiring)
        self.layer = layer
        self.observation = observation

    @property
    def config(self) -> Mapping[str, Any]:
        return {}

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        return {"observation": self.observation}

    @property
    def sub_modules(self) -> Mapping[str, TorchInputLayer]:
        return {"layer": self.layer}

    def forward(self, batch_size: int) -> Tensor:
        obs = self.observation()  # (F, C, D)
        obs = obs.unsqueeze(dim=2)  # (F, C, 1, D)
        x = self.layer(obs)  # (F, 1, K)
        return x.expand(x.shape[0], batch_size, x.shape[2])

    def sample(self, num_samples: int = 1) -> Tensor:
        if self.num_variables != 1:
            raise NotImplementedError("Sampling a multivariate Evidence layer is not implemented")
        # Sampling an evidence layer translates to return the given observation
        obs = self.observation()  # (F, C, D=1)
        obs = obs.unsqueeze(dim=-1)  # (F, C, 1, 1)
        return obs.expand(size=(-1, -1, self.num_output_units, num_samples))

config property ¤

layer = layer instance-attribute ¤

observation = observation instance-attribute ¤

params property ¤

sub_modules property ¤

__init__(layer, *, observation, semiring=None) ¤

Initializes an evidence layer.

Parameters:

Name Type Description Default
layer TorchInputLayer

The input layer on which compute the evidence of.

required
observation TorchParameter

The observation, i.e., the input to pass to the given input layer. It must be a parameter of shape \((F, C, D)\), where \(F\) is the number of folds of the given layer, \(D\) is the number variables the given layer is defined on, and \(C\) is the number channels per variable.

required
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the number of folds or the shape of the given observation is incorrect, with respect to the expected input shape of the given input layer.

Source code in cirkit/backend/torch/layers/input.py
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
def __init__(
    self,
    layer: TorchInputLayer,
    *,
    observation: TorchParameter,
    semiring: Semiring | None = None,
) -> None:
    r"""Initializes an evidence layer.

    Args:
        layer: The input layer on which compute the evidence of.
        observation: The observation, i.e., the input to pass to the given input layer.
            It must be a parameter of shape $(F, C, D)$, where $F$ is the number of folds
            of the given layer, $D$ is the number variables the given layer is defined on,
            and $C$ is the number channels per variable.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the number of folds or the shape of the given observation is incorrect,
            with respect to the expected input shape of the given input layer.
    """
    if observation.num_folds != layer.num_folds:
        raise ValueError(
            f"The number of folds in the observation and in the layer should be the same, "
            f"but found {observation.num_folds} and {layer.num_folds} respectively"
        )
    if len(observation.shape) != 2:
        raise ValueError(
            f"Expected observation of shape (num_channels, num_variables), "
            f"but found {observation.shape}"
        )
    num_channels, num_variables = observation.shape
    if num_channels != layer.num_channels:
        raise ValueError(
            f"Expected an observation with number of channels {layer.num_channels}, "
            f"but found {num_channels}"
        )
    if num_variables != layer.num_variables:
        raise ValueError(
            f"Expected an observation with number of variables {layer.num_variables}, "
            f"but found {num_variables}"
        )
    super().__init__(layer.num_output_units, layer.num_folds, semiring=semiring)
    self.layer = layer
    self.observation = observation

forward(batch_size) ¤

Source code in cirkit/backend/torch/layers/input.py
829
830
831
832
833
def forward(self, batch_size: int) -> Tensor:
    obs = self.observation()  # (F, C, D)
    obs = obs.unsqueeze(dim=2)  # (F, C, 1, D)
    x = self.layer(obs)  # (F, 1, K)
    return x.expand(x.shape[0], batch_size, x.shape[2])

sample(num_samples=1) ¤

Source code in cirkit/backend/torch/layers/input.py
835
836
837
838
839
840
841
def sample(self, num_samples: int = 1) -> Tensor:
    if self.num_variables != 1:
        raise NotImplementedError("Sampling a multivariate Evidence layer is not implemented")
    # Sampling an evidence layer translates to return the given observation
    obs = self.observation()  # (F, C, D=1)
    obs = obs.unsqueeze(dim=-1)  # (F, C, 1, 1)
    return obs.expand(size=(-1, -1, self.num_output_units, num_samples))

TorchExpFamilyLayer ¤

Bases: TorchInputFunctionLayer, ABC

The abstract base class for exponential family distribution layers. An input layer that is an exponential family distribution must define two methods. The first one is the log_unnormalized_likelihood, used to compute the possibly-unnormalized log-likelihood. The second one is the log_partition_function, used to compute the logarithm of the partition function.

Source code in cirkit/backend/torch/layers/input.py
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
class TorchExpFamilyLayer(TorchInputFunctionLayer, ABC):
    """The abstract base class for exponential family distribution layers.
    An input layer that is an exponential family distribution must define two methods.
    The first one is the ```log_unnormalized_likelihood```, used to compute the
    possibly-unnormalized log-likelihood. The second one is the ```log_partition_function```,
    used to compute the logarithm of the partition function."""

    def forward(self, x: Tensor) -> Tensor:
        x = self.log_unnormalized_likelihood(x)
        return self.semiring.map_from(x, LSESumSemiring)

    def integrate(self) -> Tensor:
        log_partition = self.log_partition_function()
        return self.semiring.map_from(log_partition, LSESumSemiring)

    @abstractmethod
    def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
        """Compute the (possibly unnormalized) log-likelihood of the given inputs.

        Args:
            x: The input tensor.

        Returns:
            Tensor: The (possibly unnormalized) log-likelihood as a tensor of shape $(F, K)$,
                where $F$ is the number of folds and $K$ is the number of output units.
        """

    @abstractmethod
    def log_partition_function(self) -> Tensor:
        """Compute the logarithm of the partition function of the layer.

        Returns:
            Tensor: The logarithm of the partition function as a tensor of shape $(F, K)$,
                where $F$ is the number of folds and $K$ is the number of output units.
                Note that it will be a tensor of zeros if the layer encodes already normalized
                exponential family distributions.
        """

forward(x) ¤

Source code in cirkit/backend/torch/layers/input.py
303
304
305
def forward(self, x: Tensor) -> Tensor:
    x = self.log_unnormalized_likelihood(x)
    return self.semiring.map_from(x, LSESumSemiring)

integrate() ¤

Source code in cirkit/backend/torch/layers/input.py
307
308
309
def integrate(self) -> Tensor:
    log_partition = self.log_partition_function()
    return self.semiring.map_from(log_partition, LSESumSemiring)

log_partition_function() abstractmethod ¤

Compute the logarithm of the partition function of the layer.

Returns:

Name Type Description
Tensor Tensor

The logarithm of the partition function as a tensor of shape \((F, K)\), where \(F\) is the number of folds and \(K\) is the number of output units. Note that it will be a tensor of zeros if the layer encodes already normalized exponential family distributions.

Source code in cirkit/backend/torch/layers/input.py
323
324
325
326
327
328
329
330
331
332
@abstractmethod
def log_partition_function(self) -> Tensor:
    """Compute the logarithm of the partition function of the layer.

    Returns:
        Tensor: The logarithm of the partition function as a tensor of shape $(F, K)$,
            where $F$ is the number of folds and $K$ is the number of output units.
            Note that it will be a tensor of zeros if the layer encodes already normalized
            exponential family distributions.
    """

log_unnormalized_likelihood(x) abstractmethod ¤

Compute the (possibly unnormalized) log-likelihood of the given inputs.

Parameters:

Name Type Description Default
x Tensor

The input tensor.

required

Returns:

Name Type Description
Tensor Tensor

The (possibly unnormalized) log-likelihood as a tensor of shape \((F, K)\), where \(F\) is the number of folds and \(K\) is the number of output units.

Source code in cirkit/backend/torch/layers/input.py
311
312
313
314
315
316
317
318
319
320
321
@abstractmethod
def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
    """Compute the (possibly unnormalized) log-likelihood of the given inputs.

    Args:
        x: The input tensor.

    Returns:
        Tensor: The (possibly unnormalized) log-likelihood as a tensor of shape $(F, K)$,
            where $F$ is the number of folds and $K$ is the number of output units.
    """

TorchGaussianLayer ¤

Bases: TorchExpFamilyLayer

The Gaussian distribution layer. Optionally, this layer can encode unnormalized Gaussian distributions with the spefication of a log-partition function parameter.

Source code in cirkit/backend/torch/layers/input.py
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
class TorchGaussianLayer(TorchExpFamilyLayer):
    """The Gaussian distribution layer. Optionally, this layer can encode unnormalized Gaussian
    distributions with the spefication of a log-partition function parameter."""

    def __init__(
        self,
        scope_idx: Tensor,
        num_output_units: int,
        num_channels: int = 1,
        *,
        mean: TorchParameter,
        stddev: TorchParameter,
        log_partition: TorchParameter | None = None,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initialize a Gaussian layer.

        Args:
            scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
                $D$ is the number of variables on which the input layers in each fold are defined on.
                Alternatively, a tensor of shape $(D,)$ can be specified, which will be interpreted
                as a tensor of shape $(1, D)$, i.e., with $F = 1$.
            num_output_units: The number of output units.
            num_channels: The number of channels.
            mean: The mean parameter, having shape $(F, K, C)$, where $K$ is the number of
                output units and $C$ is the number of channels.
            stddev: The standard deviation parameter, having shape $(F, K, C)$, where $K$ is the
                number of output units and $C$ is the number of channels.
            log_partition: An optional parameter of shape $(F, K, C)$, encoding the log-partition.
                function. If this is not None, then the Gaussian layer encodes unnormalized
                Gaussian likelihoods, which are then normalized with the given log-partition
                function.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the scope contains more than one variable.
            ValueError: If the mean and standard deviation parameter shapes are incorrect.
            ValueError: If the log-partition function parameter shape is incorrect.
        """
        num_variables = scope_idx.shape[-1]
        if num_variables != 1:
            raise ValueError("The Gaussian layer encodes a univariate distribution")
        super().__init__(
            scope_idx,
            num_output_units,
            num_channels=num_channels,
            semiring=semiring,
        )
        if not self._valid_mean_stddev_shape(mean):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._mean_stddev_shape} for 'mean', found"
                f"{mean.num_folds} and {mean.shape}, respectively"
            )
        if not self._valid_mean_stddev_shape(stddev):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._mean_stddev_shape} for 'stddev', found"
                f"{stddev.num_folds} and {stddev.shape}, respectively"
            )
        if log_partition is not None and not self._valid_log_partition_shape(log_partition):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._log_partition_shape} for 'log_partition', found"
                f"{log_partition.num_folds} and {log_partition.shape}, respectively"
            )
        self.mean = mean
        self.stddev = stddev
        self.log_partition = log_partition

    def _valid_mean_stddev_shape(self, p: TorchParameter) -> bool:
        if p.num_folds != self.num_folds:
            return False
        return p.shape == self._mean_stddev_shape

    def _valid_log_partition_shape(self, log_partition: TorchParameter) -> bool:
        if log_partition.num_folds != self.num_folds:
            return False
        return log_partition.shape == self._log_partition_shape

    @property
    def _mean_stddev_shape(self) -> tuple[int, ...]:
        return self.num_output_units, self.num_channels

    @property
    def _log_partition_shape(self) -> tuple[int, ...]:
        return self.num_output_units, self.num_channels

    @property
    def config(self) -> Mapping[str, Any]:
        return {"num_output_units": self.num_output_units, "num_channels": self.num_channels}

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        params = {"mean": self.mean, "stddev": self.stddev}
        if self.log_partition is not None:
            params["log_partition"] = self.log_partition
        return params

    def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
        mean = self.mean().unsqueeze(dim=1)  # (F, 1, K, C)
        stddev = self.stddev().unsqueeze(dim=1)  # (F, 1, K, C)
        x = x.permute(0, 2, 3, 1)  # (F, C, B, 1) -> (F, B, 1, C)
        x = distributions.Normal(loc=mean, scale=stddev).log_prob(x)  # (F, B, K, C)
        x = torch.sum(x, dim=3)  # (F, B, K)
        if self.log_partition is not None:
            log_partition = self.log_partition()  # (F, K, C)
            x = x + torch.sum(log_partition, dim=2).unsqueeze(dim=1)
        return x

    def log_partition_function(self) -> Tensor:
        if self.log_partition is None:
            return torch.zeros(
                size=(self.num_folds, 1, self.num_output_units), device=self.mean.device
            )
        log_partition = self.log_partition()  # (F, K, C)
        return torch.sum(log_partition, dim=2).unsqueeze(dim=1)

    def sample(self, num_samples: int = 1) -> Tensor:
        dist = distributions.Normal(loc=self.mean(), scale=self.stddev())
        samples = dist.sample((num_samples,))  # (N, F, K, C)
        samples = samples.permute(1, 3, 2, 0)  # (F, C, K, N)
        return samples

_log_partition_shape property ¤

_mean_stddev_shape property ¤

config property ¤

log_partition = log_partition instance-attribute ¤

mean = mean instance-attribute ¤

params property ¤

stddev = stddev instance-attribute ¤

__init__(scope_idx, num_output_units, num_channels=1, *, mean, stddev, log_partition=None, semiring=None) ¤

Initialize a Gaussian layer.

Parameters:

Name Type Description Default
scope_idx Tensor

A tensor of shape \((F, D)\), where \(F\) is the number of folds, and \(D\) is the number of variables on which the input layers in each fold are defined on. Alternatively, a tensor of shape \((D,)\) can be specified, which will be interpreted as a tensor of shape \((1, D)\), i.e., with \(F = 1\).

required
num_output_units int

The number of output units.

required
num_channels int

The number of channels.

1
mean TorchParameter

The mean parameter, having shape \((F, K, C)\), where \(K\) is the number of output units and \(C\) is the number of channels.

required
stddev TorchParameter

The standard deviation parameter, having shape \((F, K, C)\), where \(K\) is the number of output units and \(C\) is the number of channels.

required
log_partition TorchParameter | None

An optional parameter of shape \((F, K, C)\), encoding the log-partition. function. If this is not None, then the Gaussian layer encodes unnormalized Gaussian likelihoods, which are then normalized with the given log-partition function.

None
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the scope contains more than one variable.

ValueError

If the mean and standard deviation parameter shapes are incorrect.

ValueError

If the log-partition function parameter shape is incorrect.

Source code in cirkit/backend/torch/layers/input.py
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def __init__(
    self,
    scope_idx: Tensor,
    num_output_units: int,
    num_channels: int = 1,
    *,
    mean: TorchParameter,
    stddev: TorchParameter,
    log_partition: TorchParameter | None = None,
    semiring: Semiring | None = None,
) -> None:
    r"""Initialize a Gaussian layer.

    Args:
        scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
            $D$ is the number of variables on which the input layers in each fold are defined on.
            Alternatively, a tensor of shape $(D,)$ can be specified, which will be interpreted
            as a tensor of shape $(1, D)$, i.e., with $F = 1$.
        num_output_units: The number of output units.
        num_channels: The number of channels.
        mean: The mean parameter, having shape $(F, K, C)$, where $K$ is the number of
            output units and $C$ is the number of channels.
        stddev: The standard deviation parameter, having shape $(F, K, C)$, where $K$ is the
            number of output units and $C$ is the number of channels.
        log_partition: An optional parameter of shape $(F, K, C)$, encoding the log-partition.
            function. If this is not None, then the Gaussian layer encodes unnormalized
            Gaussian likelihoods, which are then normalized with the given log-partition
            function.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the scope contains more than one variable.
        ValueError: If the mean and standard deviation parameter shapes are incorrect.
        ValueError: If the log-partition function parameter shape is incorrect.
    """
    num_variables = scope_idx.shape[-1]
    if num_variables != 1:
        raise ValueError("The Gaussian layer encodes a univariate distribution")
    super().__init__(
        scope_idx,
        num_output_units,
        num_channels=num_channels,
        semiring=semiring,
    )
    if not self._valid_mean_stddev_shape(mean):
        raise ValueError(
            f"Expected number of folds {self.num_folds} "
            f"and shape {self._mean_stddev_shape} for 'mean', found"
            f"{mean.num_folds} and {mean.shape}, respectively"
        )
    if not self._valid_mean_stddev_shape(stddev):
        raise ValueError(
            f"Expected number of folds {self.num_folds} "
            f"and shape {self._mean_stddev_shape} for 'stddev', found"
            f"{stddev.num_folds} and {stddev.shape}, respectively"
        )
    if log_partition is not None and not self._valid_log_partition_shape(log_partition):
        raise ValueError(
            f"Expected number of folds {self.num_folds} "
            f"and shape {self._log_partition_shape} for 'log_partition', found"
            f"{log_partition.num_folds} and {log_partition.shape}, respectively"
        )
    self.mean = mean
    self.stddev = stddev
    self.log_partition = log_partition

_valid_log_partition_shape(log_partition) ¤

Source code in cirkit/backend/torch/layers/input.py
659
660
661
662
def _valid_log_partition_shape(self, log_partition: TorchParameter) -> bool:
    if log_partition.num_folds != self.num_folds:
        return False
    return log_partition.shape == self._log_partition_shape

_valid_mean_stddev_shape(p) ¤

Source code in cirkit/backend/torch/layers/input.py
654
655
656
657
def _valid_mean_stddev_shape(self, p: TorchParameter) -> bool:
    if p.num_folds != self.num_folds:
        return False
    return p.shape == self._mean_stddev_shape

log_partition_function() ¤

Source code in cirkit/backend/torch/layers/input.py
694
695
696
697
698
699
700
def log_partition_function(self) -> Tensor:
    if self.log_partition is None:
        return torch.zeros(
            size=(self.num_folds, 1, self.num_output_units), device=self.mean.device
        )
    log_partition = self.log_partition()  # (F, K, C)
    return torch.sum(log_partition, dim=2).unsqueeze(dim=1)

log_unnormalized_likelihood(x) ¤

Source code in cirkit/backend/torch/layers/input.py
683
684
685
686
687
688
689
690
691
692
def log_unnormalized_likelihood(self, x: Tensor) -> Tensor:
    mean = self.mean().unsqueeze(dim=1)  # (F, 1, K, C)
    stddev = self.stddev().unsqueeze(dim=1)  # (F, 1, K, C)
    x = x.permute(0, 2, 3, 1)  # (F, C, B, 1) -> (F, B, 1, C)
    x = distributions.Normal(loc=mean, scale=stddev).log_prob(x)  # (F, B, K, C)
    x = torch.sum(x, dim=3)  # (F, B, K)
    if self.log_partition is not None:
        log_partition = self.log_partition()  # (F, K, C)
        x = x + torch.sum(log_partition, dim=2).unsqueeze(dim=1)
    return x

sample(num_samples=1) ¤

Source code in cirkit/backend/torch/layers/input.py
702
703
704
705
706
def sample(self, num_samples: int = 1) -> Tensor:
    dist = distributions.Normal(loc=self.mean(), scale=self.stddev())
    samples = dist.sample((num_samples,))  # (N, F, K, C)
    samples = samples.permute(1, 3, 2, 0)  # (F, C, K, N)
    return samples

TorchInputFunctionLayer ¤

Bases: TorchInputLayer

An input layer encoding functions defined over a non-empty set of variables.

Source code in cirkit/backend/torch/layers/input.py
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class TorchInputFunctionLayer(TorchInputLayer):
    """An input layer encoding functions defined over a non-empty set of variables."""

    def __call__(self, x: Tensor) -> Tensor:
        # IGNORE: Idiom for nn.Module.__call__.
        return super().__call__(x)  # type: ignore[no-any-return,misc]

    @abstractmethod
    def forward(self, x: Tensor) -> Tensor:
        r"""Invoke the forward function.

        Args:
            x: The tensor input to this layer, having shape $(F, C, B, D)$, where $F$
                is the number of folds, $C$ is the number of channels,
                $B$ is the batch size, and $D$ is the number of variables.

        Returns:
            Tensor: The tensor output of this layer, having shape $(F, B, K)$, where $K$
                is the number of output units.
        """

__call__(x) ¤

Source code in cirkit/backend/torch/layers/input.py
142
143
144
def __call__(self, x: Tensor) -> Tensor:
    # IGNORE: Idiom for nn.Module.__call__.
    return super().__call__(x)  # type: ignore[no-any-return,misc]

forward(x) abstractmethod ¤

Invoke the forward function.

Parameters:

Name Type Description Default
x Tensor

The tensor input to this layer, having shape \((F, C, B, D)\), where \(F\) is the number of folds, \(C\) is the number of channels, \(B\) is the batch size, and \(D\) is the number of variables.

required

Returns:

Name Type Description
Tensor Tensor

The tensor output of this layer, having shape \((F, B, K)\), where \(K\) is the number of output units.

Source code in cirkit/backend/torch/layers/input.py
146
147
148
149
150
151
152
153
154
155
156
157
158
@abstractmethod
def forward(self, x: Tensor) -> Tensor:
    r"""Invoke the forward function.

    Args:
        x: The tensor input to this layer, having shape $(F, C, B, D)$, where $F$
            is the number of folds, $C$ is the number of channels,
            $B$ is the batch size, and $D$ is the number of variables.

    Returns:
        Tensor: The tensor output of this layer, having shape $(F, B, K)$, where $K$
            is the number of output units.
    """

TorchInputLayer ¤

Bases: TorchLayer, ABC

The abstract base class for torch input layers.

Source code in cirkit/backend/torch/layers/input.py
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 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
 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
class TorchInputLayer(TorchLayer, ABC):
    """The abstract base class for torch input layers."""

    def __init__(
        self,
        scope_idx: Tensor,
        num_output_units: int,
        *,
        num_channels: int = 1,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initialize a torch input layer.

        Args:
            scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
                $D$ is the number of variables on which the input layers in each fold are defined
                on. Alternatively, a tensor of shape $(D,)$ can be specified, which will be
                interpreted as a tensor of shape $(1, D)$, i.e., with $F = 1$.
            num_output_units: The number of output units.
            num_channels: The number of channels.
            semiring: The evaluation semiring.
                Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

        Raises:
            ValueError: If the scope index is not a vector or a matrix.
        """
        if len(scope_idx.shape) == 1:
            scope_idx = scope_idx.unsqueeze(dim=0)
        elif len(scope_idx.shape) > 2:
            raise ValueError(f"The scope index must be a matrix, but found shape {scope_idx.shape}")
        num_folds, num_variables = scope_idx.shape
        super().__init__(
            num_variables,
            num_output_units,
            arity=num_channels,
            num_folds=num_folds,
            semiring=semiring,
        )
        self.register_buffer("_scope_idx", scope_idx)

    @property
    def scope_idx(self) -> Tensor:
        """Retrieve the scope index tensor.

        Returns:
            The scope index tensor.
        """
        return self._scope_idx

    @property
    def num_variables(self) -> int:
        """The number of variables the input layer is defined on.

        Returns:
            The number of variables.
        """
        return self.num_input_units

    @property
    def num_channels(self) -> int:
        """The number of channels per variable.

        Returns:
            The number of channels.
        """
        return self.arity

    @property
    @abstractmethod
    def config(self) -> Mapping[str, Any]:
        ...

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        return {}

    @property
    def fold_settings(self) -> tuple[Any, ...]:
        pshapes = [(n, p.shape) for n, p in self.params.items()]
        return self.num_variables, *self.config.items(), *pshapes

    def integrate(self) -> Tensor:
        r"""Integrate an input layer over all its variables' domain.

        Returns:
            Tensor: The tensor result of the integration, having shape $(F, K)$, where
                $F$ is the number of folds and $K$ is the number of output units.

        Raises:
            TypeError: If integration is not supported by the layer.
        """
        raise TypeError(f"Integration is not supported for layers of type {type(self)}")

    def sample(self, num_samples: int = 1) -> Tensor:
        r"""If the input layer encodes a probability distribution, then sample from it.

        Args:
            num_samples: The number of data points to sample.

        Returns:
            Tensor: The tensorized sample, having shape $(F, C, K, N)$, where
                $F$ is the number of folds, $K$ is the number of output units,
                $C$ is the number of channels, and $N$ is the number of samples.

        Raises:
            TypeError: If sampling is not supported by the layer.
        """
        raise TypeError(f"Sampling is not supported for layers of type {type(self)}")

    def extra_repr(self) -> str:
        return (
            "  ".join(
                [
                    f"folds: {self.num_folds}",
                    f"channels: {self.num_channels}",
                    f"variables: {self.num_variables}",
                    f"output-units: {self.num_output_units}",
                ]
            )
            + "\n"
            + f"input-shape: {(self.num_folds, self.arity, -1, self.num_input_units)}"
            + "\n"
            + f"output-shape: {(self.num_folds, -1, self.num_output_units)}"
        )

config abstractmethod property ¤

fold_settings property ¤

num_channels property ¤

The number of channels per variable.

Returns:

Type Description
int

The number of channels.

num_variables property ¤

The number of variables the input layer is defined on.

Returns:

Type Description
int

The number of variables.

params property ¤

scope_idx property ¤

Retrieve the scope index tensor.

Returns:

Type Description
Tensor

The scope index tensor.

__init__(scope_idx, num_output_units, *, num_channels=1, semiring=None) ¤

Initialize a torch input layer.

Parameters:

Name Type Description Default
scope_idx Tensor

A tensor of shape \((F, D)\), where \(F\) is the number of folds, and \(D\) is the number of variables on which the input layers in each fold are defined on. Alternatively, a tensor of shape \((D,)\) can be specified, which will be interpreted as a tensor of shape \((1, D)\), i.e., with \(F = 1\).

required
num_output_units int

The number of output units.

required
num_channels int

The number of channels.

1
semiring Semiring | None

The evaluation semiring. Defaults to SumProductSemiring.

None

Raises:

Type Description
ValueError

If the scope index is not a vector or a matrix.

Source code in cirkit/backend/torch/layers/input.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    scope_idx: Tensor,
    num_output_units: int,
    *,
    num_channels: int = 1,
    semiring: Semiring | None = None,
) -> None:
    r"""Initialize a torch input layer.

    Args:
        scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
            $D$ is the number of variables on which the input layers in each fold are defined
            on. Alternatively, a tensor of shape $(D,)$ can be specified, which will be
            interpreted as a tensor of shape $(1, D)$, i.e., with $F = 1$.
        num_output_units: The number of output units.
        num_channels: The number of channels.
        semiring: The evaluation semiring.
            Defaults to [SumProductSemiring][cirkit.backend.torch.semiring.SumProductSemiring].

    Raises:
        ValueError: If the scope index is not a vector or a matrix.
    """
    if len(scope_idx.shape) == 1:
        scope_idx = scope_idx.unsqueeze(dim=0)
    elif len(scope_idx.shape) > 2:
        raise ValueError(f"The scope index must be a matrix, but found shape {scope_idx.shape}")
    num_folds, num_variables = scope_idx.shape
    super().__init__(
        num_variables,
        num_output_units,
        arity=num_channels,
        num_folds=num_folds,
        semiring=semiring,
    )
    self.register_buffer("_scope_idx", scope_idx)

extra_repr() ¤

Source code in cirkit/backend/torch/layers/input.py
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def extra_repr(self) -> str:
    return (
        "  ".join(
            [
                f"folds: {self.num_folds}",
                f"channels: {self.num_channels}",
                f"variables: {self.num_variables}",
                f"output-units: {self.num_output_units}",
            ]
        )
        + "\n"
        + f"input-shape: {(self.num_folds, self.arity, -1, self.num_input_units)}"
        + "\n"
        + f"output-shape: {(self.num_folds, -1, self.num_output_units)}"
    )

integrate() ¤

Integrate an input layer over all its variables' domain.

Returns:

Name Type Description
Tensor Tensor

The tensor result of the integration, having shape \((F, K)\), where \(F\) is the number of folds and \(K\) is the number of output units.

Raises:

Type Description
TypeError

If integration is not supported by the layer.

Source code in cirkit/backend/torch/layers/input.py
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def integrate(self) -> Tensor:
    r"""Integrate an input layer over all its variables' domain.

    Returns:
        Tensor: The tensor result of the integration, having shape $(F, K)$, where
            $F$ is the number of folds and $K$ is the number of output units.

    Raises:
        TypeError: If integration is not supported by the layer.
    """
    raise TypeError(f"Integration is not supported for layers of type {type(self)}")

sample(num_samples=1) ¤

If the input layer encodes a probability distribution, then sample from it.

Parameters:

Name Type Description Default
num_samples int

The number of data points to sample.

1

Returns:

Name Type Description
Tensor Tensor

The tensorized sample, having shape \((F, C, K, N)\), where \(F\) is the number of folds, \(K\) is the number of output units, \(C\) is the number of channels, and \(N\) is the number of samples.

Raises:

Type Description
TypeError

If sampling is not supported by the layer.

Source code in cirkit/backend/torch/layers/input.py
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def sample(self, num_samples: int = 1) -> Tensor:
    r"""If the input layer encodes a probability distribution, then sample from it.

    Args:
        num_samples: The number of data points to sample.

    Returns:
        Tensor: The tensorized sample, having shape $(F, C, K, N)$, where
            $F$ is the number of folds, $K$ is the number of output units,
            $C$ is the number of channels, and $N$ is the number of samples.

    Raises:
        TypeError: If sampling is not supported by the layer.
    """
    raise TypeError(f"Sampling is not supported for layers of type {type(self)}")

TorchPolynomialLayer ¤

Bases: TorchInputFunctionLayer

The polynomial input layer, evaluating a vector of parameterized polynomials.

Source code in cirkit/backend/torch/layers/input.py
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
class TorchPolynomialLayer(TorchInputFunctionLayer):
    """The polynomial input layer, evaluating a vector of parameterized polynomials."""

    def __init__(
        self,
        scope_idx: Tensor,
        num_output_units: int,
        num_channels: int = 1,
        *,
        degree: int,
        coeff: TorchParameter,
        semiring: Semiring | None = None,
    ) -> None:
        r"""Initialize a polynomial layer.

        Args:
            scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
                $D$ is the number of variables on which the input layers in each fold are defined
                on. Alternatively, a tensor of shape $(D,)$ can be specified, which will be
                interpreted as a tensor of shape $(1, D)$, i.e., with $F = 1$.
            num_output_units: The number of output units.
            num_channels: The number of channels.
            degree: The degree of polynomial.
            coeff: The coefficient parameter, having shape $(F, K, \mathsf{degree} + 1)$, where $K$ is the number
                of output units.

        Raises:
            ValueError: If the scope contains more than one variable.
            ValueError: If the coefficients is not correct.
        """
        num_variables = scope_idx.shape[-1]
        if num_variables != 1:
            raise ValueError("The Polynomial layer encodes a univariate distribution")
        if num_channels != 1:
            raise ValueError("The Polynomial layer encodes a univariate distribution")
        super().__init__(
            scope_idx,
            num_output_units,
            num_channels=num_channels,
            semiring=semiring,
        )
        self.degree = degree
        if not self._valid_parameters_shape(coeff):
            raise ValueError(
                f"Expected number of folds {self.num_folds} "
                f"and shape {self._coeff_shape} for 'coeff', found"
                f"{coeff.num_folds} and {coeff.shape}, respectively"
            )
        self.coeff = coeff

    def _valid_parameters_shape(self, p: TorchParameter) -> bool:
        if p.num_folds != self.num_folds:
            return False
        return p.shape == self._coeff_shape

    @property
    def _coeff_shape(self) -> tuple[int, ...]:
        return self.num_output_units, self.degree + 1

    @staticmethod
    def _polyval(coeff: Tensor, x: Tensor) -> Tensor:
        r"""Evaluate polynomial given coefficients and point, with the shape for PolynomialLayer.

        Args:
            coeff: The coefficients of the polynomial, shape $(F, K_o, \mathsf{degree} + 1)$.
            x: The point of the variable, shape $(F, H, B, K_i)$, where $H=K_i=1$.

        Returns:
            Tensor: The value of the polymonial, shape $(F, B, K_o)$.
        """
        x = x.squeeze(dim=1)  # shape (F, H=1, B, Ki=1) -> (F, B, 1).
        y = x.new_zeros(*x.shape[:-1], coeff.shape[-2])  # shape (F, B, Ko).

        # TODO: iterating over dim=2 is inefficient
        for a_n in reversed(
            coeff.unbind(dim=2)
        ):  # Reverse iterator of the degree axis, shape (F, Ko).
            # a_n shape (F, Ko) -> (F, 1, Ko).
            y = torch.addcmul(a_n.unsqueeze(dim=1), x, y)  # y = a_n + x * y, by Horner's method.
        return y  # shape (F, B, Ko).

    @property
    def config(self) -> Mapping[str, Any]:
        return {
            "num_output_units": self.num_output_units,
            "num_channels": self.num_channels,
            "degree": self.degree,
        }

    @property
    def params(self) -> Mapping[str, TorchParameter]:
        return {"coeff": self.coeff}

    def forward(self, x: Tensor) -> Tensor:
        coeff = self.coeff()  # shape (F, Ko, dp1)
        return self.semiring.map_from(TorchPolynomialLayer._polyval(coeff, x), SumProductSemiring)

_coeff_shape property ¤

coeff = coeff instance-attribute ¤

config property ¤

degree = degree instance-attribute ¤

params property ¤

__init__(scope_idx, num_output_units, num_channels=1, *, degree, coeff, semiring=None) ¤

Initialize a polynomial layer.

Parameters:

Name Type Description Default
scope_idx Tensor

A tensor of shape \((F, D)\), where \(F\) is the number of folds, and \(D\) is the number of variables on which the input layers in each fold are defined on. Alternatively, a tensor of shape \((D,)\) can be specified, which will be interpreted as a tensor of shape \((1, D)\), i.e., with \(F = 1\).

required
num_output_units int

The number of output units.

required
num_channels int

The number of channels.

1
degree int

The degree of polynomial.

required
coeff TorchParameter

The coefficient parameter, having shape \((F, K, \mathsf{degree} + 1)\), where \(K\) is the number of output units.

required

Raises:

Type Description
ValueError

If the scope contains more than one variable.

ValueError

If the coefficients is not correct.

Source code in cirkit/backend/torch/layers/input.py
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
def __init__(
    self,
    scope_idx: Tensor,
    num_output_units: int,
    num_channels: int = 1,
    *,
    degree: int,
    coeff: TorchParameter,
    semiring: Semiring | None = None,
) -> None:
    r"""Initialize a polynomial layer.

    Args:
        scope_idx: A tensor of shape $(F, D)$, where $F$ is the number of folds, and
            $D$ is the number of variables on which the input layers in each fold are defined
            on. Alternatively, a tensor of shape $(D,)$ can be specified, which will be
            interpreted as a tensor of shape $(1, D)$, i.e., with $F = 1$.
        num_output_units: The number of output units.
        num_channels: The number of channels.
        degree: The degree of polynomial.
        coeff: The coefficient parameter, having shape $(F, K, \mathsf{degree} + 1)$, where $K$ is the number
            of output units.

    Raises:
        ValueError: If the scope contains more than one variable.
        ValueError: If the coefficients is not correct.
    """
    num_variables = scope_idx.shape[-1]
    if num_variables != 1:
        raise ValueError("The Polynomial layer encodes a univariate distribution")
    if num_channels != 1:
        raise ValueError("The Polynomial layer encodes a univariate distribution")
    super().__init__(
        scope_idx,
        num_output_units,
        num_channels=num_channels,
        semiring=semiring,
    )
    self.degree = degree
    if not self._valid_parameters_shape(coeff):
        raise ValueError(
            f"Expected number of folds {self.num_folds} "
            f"and shape {self._coeff_shape} for 'coeff', found"
            f"{coeff.num_folds} and {coeff.shape}, respectively"
        )
    self.coeff = coeff

_polyval(coeff, x) staticmethod ¤

Evaluate polynomial given coefficients and point, with the shape for PolynomialLayer.

Parameters:

Name Type Description Default
coeff Tensor

The coefficients of the polynomial, shape \((F, K_o, \mathsf{degree} + 1)\).

required
x Tensor

The point of the variable, shape \((F, H, B, K_i)\), where \(H=K_i=1\).

required

Returns:

Name Type Description
Tensor Tensor

The value of the polymonial, shape \((F, B, K_o)\).

Source code in cirkit/backend/torch/layers/input.py
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
@staticmethod
def _polyval(coeff: Tensor, x: Tensor) -> Tensor:
    r"""Evaluate polynomial given coefficients and point, with the shape for PolynomialLayer.

    Args:
        coeff: The coefficients of the polynomial, shape $(F, K_o, \mathsf{degree} + 1)$.
        x: The point of the variable, shape $(F, H, B, K_i)$, where $H=K_i=1$.

    Returns:
        Tensor: The value of the polymonial, shape $(F, B, K_o)$.
    """
    x = x.squeeze(dim=1)  # shape (F, H=1, B, Ki=1) -> (F, B, 1).
    y = x.new_zeros(*x.shape[:-1], coeff.shape[-2])  # shape (F, B, Ko).

    # TODO: iterating over dim=2 is inefficient
    for a_n in reversed(
        coeff.unbind(dim=2)
    ):  # Reverse iterator of the degree axis, shape (F, Ko).
        # a_n shape (F, Ko) -> (F, 1, Ko).
        y = torch.addcmul(a_n.unsqueeze(dim=1), x, y)  # y = a_n + x * y, by Horner's method.
    return y  # shape (F, B, Ko).

_valid_parameters_shape(p) ¤

Source code in cirkit/backend/torch/layers/input.py
894
895
896
897
def _valid_parameters_shape(self, p: TorchParameter) -> bool:
    if p.num_folds != self.num_folds:
        return False
    return p.shape == self._coeff_shape

forward(x) ¤

Source code in cirkit/backend/torch/layers/input.py
937
938
939
def forward(self, x: Tensor) -> Tensor:
    coeff = self.coeff()  # shape (F, Ko, dp1)
    return self.semiring.map_from(TorchPolynomialLayer._polyval(coeff, x), SumProductSemiring)