Módulo 10: Paradigmas & Padrões • Bloco 1 • Trilha em 4 Níveis

🎯 Nível 1: Fundamentos

Problema 04.1 — Motor de Inferência Lógica e Dedução por Unificação

Contexto: No paradigma lógico, o desenvolvedor declara uma base de conhecimento e consulta relações sem prescrever o algoritmo de busca.

Requisitos de Execução:

  1. Crie uma estrutura em Python BaseConhecimentoLogica contendo fatos de parentesco: pai("Carlos", "Ana"), pai("Carlos", "Bruno"), pai("Bruno", "Lucas").
  2. Implemente a regra dedutiva obter_avo(neto: str) -> List[str] e obter_irmaos(pessoa: str) -> List[str].
  3. Retorne as deduções automaticamente a partir da unificação dos fatos.

Resultado Esperado

=== Deduções da Base de Conhecimento ===
Avô de 'Lucas': ['Carlos']
Irmãos de 'Ana': ['Bruno']

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_04_1_SeuNome.py
  2. Envie na tarefa: Atividade Cap 04 - Paradigmas Declarativo, Lógico e Reativo
🔑 Gabarito de Código & Solução Comentada 
# Atividade_04_1_Gabarito.py
from typing import List, Tuple

class BaseConhecimentoLogica:
    def __init__(self):
        # Fatos: (pai, filho)
        self.fatos_pai: List[Tuple[str, str]] = []

    def adicionar_fato_pai(self, pai: str, filho: str):
        self.fatos_pai.append((pai, filho))

    def obter_avo(self, neto: str) -> List[str]:
        """Regra: avo(X, Y) :- pai(X, Z), pai(Z, Y)."""
        pais_do_neto = [p for p, f in self.fatos_pai if f == neto]
        avos = []
        for pai in pais_do_neto:
            avos.extend([avo for avo, f in self.fatos_pai if f == pai])
        return avos

    def obter_irmaos(self, pessoa: str) -> List[str]:
        """Regra: irmao(X, Y) :- pai(Z, X), pai(Z, Y), X \= Y."""
        pais = [p for p, f in self.fatos_pai if f == pessoa]
        irmaos = set()
        for pai in pais:
            for p, f in self.fatos_pai:
                if p == pai and f != pessoa:
                    irmaos.add(f)
        return list(irmaos)

if __name__ == "__main__":
    kb = BaseConhecimentoLogica()
    kb.adicionar_fato_pai("Carlos", "Ana")
    kb.adicionar_fato_pai("Carlos", "Bruno")
    kb.adicionar_fato_pai("Bruno", "Lucas")

    print("=== Deduções da Base de Conhecimento ===")
    print(f"Avô de 'Lucas': {kb.obter_avo('Lucas')}")
    print(f"Irmãos de 'Ana': {kb.obter_irmaos('Ana')}")

🔍 Nível 2: Prática

Problema 04.2 — Observable Reativo com Emissão e Filtragem Push

Contexto: Em programação reativa, consumidores se inscrevem para receber valores propagados ativamente pelo emissor (Push Notification).

Requisitos de Execução:

  1. Implemente a classe StreamObservable.
  2. Adicione os operadores encadeáveis .filter(predicate) e .map(transform).
  3. Conecte um fluxo de telemetria de velocidade veicular, filtrando apenas excessos ($v > 100\text{ km/h}$) e emitindo avisos de infração aos observadores.

Resultado Esperado

=== Monitoramento Reativo de Rodovia ===
Emitindo leituras de radar:
[MULTA DISPARADA]: Veículo a 115.0 km/h na Rodovia SP-330!
[MULTA DISPARADA]: Veículo a 132.5 km/h na Rodovia SP-330!

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_04_2_SeuNome.py
  2. Envie na tarefa: Atividade Cap 04 - Paradigmas Declarativo, Lógico e Reativo
🔑 Gabarito de Código & Solução Comentada 
# Atividade_04_2_Gabarito.py
from typing import Callable, List, Any

class StreamObservable:
    def __init__(self, items: List[Any]):
        self._items = items

    def filter(self, predicate: Callable[[Any], bool]) -> 'StreamObservable':
        return StreamObservable([x for x in self._items if predicate(x)])

    def map(self, transform: Callable[[Any], Any]) -> 'StreamObservable':
        return StreamObservable([transform(x) for x in self._items])

    def subscribe(self, on_next: Callable[[Any], None]):
        for item in self._items:
            on_next(item)

if __name__ == "__main__":
    velocidades = [85.0, 92.0, 115.0, 70.0, 132.5, 99.0]
    print("=== Monitoramento Reativo de Rodovia ===")
    print("Emitindo leituras de radar:")

    (StreamObservable(velocidades)
     .filter(lambda v: v > 100.0)
     .map(lambda v: f"[MULTA DISPARADA]: Veículo a {v:.1f} km/h na Rodovia SP-330!")
     .subscribe(lambda alerta: print(alerta)))

🚀 Nível 3: Integração

Problema 04.3 — Controle de Contrapressão (Backpressure Sampling / Debounce)

Contexto: Em sistemas IoT de alta frequência, receptores lentos podem sofrer esgotamento de recursos (Out Of Memory). A técnica de amostragem (Sampling) descarta dados intermediários em rajadas.

Requisitos de Execução:

  1. Crie uma fila reativa BackpressureSampler com taxa de amostragem parametrizável.
  2. Emita uma rajada de $10\text{ pacotes}$ em sequência rápida.
  3. Configure o consumidor para processar apenas a cada $3\text{ amostras}$ (Sampling Rate = 3), descartando leituras redundantes com segurança.

Resultado Esperado

=== Emissão Reativa com Backpressure Sampler ===
[Produtor] Emitindo pacote #1
[Produtor] Emitindo pacote #2
[Produtor] Emitindo pacote #3 -> [CONSUMIDOR RECEBEU]: Pacote #3
[Produtor] Emitindo pacote #4
[Produtor] Emitindo pacote #5
[Produtor] Emitindo pacote #6 -> [CONSUMIDOR RECEBEU]: Pacote #6

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_04_3_SeuNome.py
  2. Envie na tarefa: Atividade Cap 04 - Paradigmas Declarativo, Lógico e Reativo
🔑 Gabarito de Código & Solução Comentada 
# Atividade_04_3_Gabarito.py
from typing import Callable, Any

class BackpressureSampler:
    def __init__(self, sample_interval: int, on_sample: Callable[[Any], None]):
        self._interval = sample_interval
        self._counter = 0
        self._on_sample = on_sample

    def emit(self, item: Any):
        self._counter += 1
        print(f"[Produtor] Emitindo {item}", end="")
        if self._counter % self._interval == 0:
            print(f" -> [CONSUMIDOR RECEBEU]: {item}")
            self._on_sample(item)
        else:
            print()

if __name__ == "__main__":
    print("=== Emissão Reativa com Backpressure Sampler ===")
    sampler = BackpressureSampler(sample_interval=3, on_sample=lambda _: None)
    for i in range(1, 8):
        sampler.emit(f"pacote #{i}")

🏆 Nível 4: Desafio

Problema 04.4 — Motor Declarativo de Regras de Negócio e Unificação Dinâmica

Contexto: Sistemas bancários corporativos e motores de crédito usam motores declarativos (como Drools / Datalog) para avaliar dezenas de políticas sem código procedural acoplado.

Requisitos de Execução:

  1. Construa um motor declarativo RuleEngine onde políticas de crédito sejam cadastradas como regras (Regra(nome, condicao_lambda, acao_lambda)).
  2. O motor deve avaliar um perfil de cliente e disparar automaticamente todas as regras satisfeitas em lote.
  3. Demonstre a inclusão dinâmica de uma nova regra fiscal sem alterar nenhuma linha do motor de execução.

Resultado Esperado

=== Motor Declarativo de Políticas de Crédito ===
Avaliando Cliente: Mariana (Score: 820, Renda: R$ 12000.00, Idade: 29)
-> [Regra Satisfeita: CartaoBlack] Concedido limite especial de R$ 50.000,00!
-> [Regra Satisfeita: TaxaJurosReduzida] Taxa de financiamento fixada em 0.79% a.m.!

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_04_4_SeuNome.py
  2. Envie na tarefa: Atividade Cap 04 - Paradigmas Declarativo, Lógico e Reativo
🔑 Gabarito de Código & Solução Comentada 
# Atividade_04_4_Gabarito.py
from dataclasses import dataclass
from typing import Callable, List, Dict, Any

@dataclass
class Regra:
    nome: str
    condicao: Callable[[Dict[str, Any]], bool]
    acao: Callable[[Dict[str, Any]], None]

class RuleEngine:
    def __init__(self):
        self._regras: List[Regra] = []

    def registrar_regra(self, regra: Regra):
        self._regras.append(regra)

    def executar(self, contexto: Dict[str, Any]):
        for r in self._regras:
            if r.condicao(contexto):
                r.acao(contexto)

if __name__ == "__main__":
    print("=== Motor Declarativo de Políticas de Crédito ===")
    cliente = {"nome": "Mariana", "score": 820, "renda": 12000.0, "idade": 29}
    print(f"Avaliando Cliente: {cliente['nome']} (Score: {cliente['score']}, Renda: R$ {cliente['renda']:.2f}, Idade: {cliente['idade']})")

    engine = RuleEngine()
    engine.registrar_regra(Regra(
        "CartaoBlack",
        lambda c: c["score"] >= 800 and c["renda"] >= 10000.0,
        lambda _: print("-> [Regra Satisfeita: CartaoBlack] Concedido limite especial de R$ 50.000,00!")
    ))
    engine.registrar_regra(Regra(
        "TaxaJurosReduzida",
        lambda c: c["score"] >= 750,
        lambda _: print("-> [Regra Satisfeita: TaxaJurosReduzida] Taxa de financiamento fixada em 0.79% a.m.!")
    ))

    engine.executar(cliente)