IEC 61850 per Software Engineer: Comunicazione Smart Grid

BAY1_PROT/PROT/XCBR1.Pos.stVal
  |         |     |     |   |
  IED name  LD    LN    DO  DA
            inst  class inst
                  +num

XCBR1 (Logical Node - Circuit Breaker)
|
├── ST (Status Data Class - dati di stato)
│   ├── Pos        - Posizione (DPC: double point control)
│   │   ├── stVal  [BOOLEAN] - Stato aperto/chiuso
│   │   ├── q      [Quality] - Validita del dato
│   │   └── t      [Timestamp] - Timestamp UTC con ms
│   ├── BlkOpn     - Blocco apertura (SPC)
│   ├── BlkCls     - Blocco chiusura (SPC)
│   └── CBOpCap    - capacità operativa (ENS enum)
│
├── MX (Measured Values)
│   └── OpCnt      [INS] - Contatore operazioni
│
└── CF (Configuration)
    ├── CBOpTmms   [INT32U] - Tempo apertura [ms]
    └── ModBlkOpn  [BOOLEAN] - Modalità blocco apertura

Functional Constraints (FC):
  ST = status, MX = measured, CF = config,
  DC = description, EX = extended def,
  CO = control, SP = setpoint

MMXU1 (Measurement Unit - 3-phase)
|
├── MX (Measured Values - Functional Constraint)
│   ├── TotW      [MV] - Potenza attiva totale [W]
│   │   ├── mag.f [FLOAT32] - Valore float
│   │   ├── q     [Quality] - validity, source, detail
│   │   └── t     [Timestamp]
│   ├── TotVAr    [MV] - Potenza reattiva totale [VAr]
│   ├── TotPF     [MV] - Power Factor totale
│   ├── Hz        [MV] - Frequenza [Hz]
│   ├── A         [WYE] - Correnti trifase
│   │   ├── phsA.cVal.mag.f  [FLOAT32] - Fase A [A]
│   │   ├── phsB.cVal.mag.f  [FLOAT32] - Fase B [A]
│   │   └── phsC.cVal.mag.f  [FLOAT32] - Fase C [A]
│   └── PhV       [WYE] - Tensioni di fase [V]
│       ├── phsA.cVal.mag.f
│       ├── phsB.cVal.mag.f
│       └── phsC.cVal.mag.f
│
└── CF (Configuration)
    └── AngRef    - Angular reference (VA, VB, VC, AA...)

Naming esempio completo:
BAY1_PROT/MEAS/MMXU1.MX.TotW.mag.f
  ^IED     ^LD  ^LN   ^FC ^DO  ^DA  ^sub-DA

<?xml version="1.0" encoding="UTF-8"?>
<SCL xmlns="http://www.iec.ch/61850/2003/SCL"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     version="2007" revision="B" release="4">

  <!-- Header: identificazione e versione -->
  <Header id="SCD_Substation_Nord_Milano"
          version="1.0"
          revision="3"
          toolID="OpenSCD v0.22.0"
          nameStructure="IEDName">
    <History>
      <Hitem version="1.0" revision="3" when="2025-03-09"
             who="f.calo@utility.it"
             what="Updated GOOSE subscriptions for BAY3"/>
    </History>
  </Header>

  <!-- Substation: topologia elettrica -->
  <Substation name="SNord" desc="Sottostazione Nord Milano 150/20kV">
    <VoltageLevel name="VL_150kV" desc="Livello 150kV">
      <Voltage unit="V" multiplier="k">150</Voltage>
      <Bay name="BAY1" desc="Feeder 1 - Linea Torino">
        <ConductingEquipment type="CBR" name="QA1"
                              desc="Circuit Breaker Bay1">
          <Terminal name="T1" connectivityNode="SNord/VL_150kV/BAY1/BusbarSection1"/>
        </ConductingEquipment>
        <ConductingEquipment type="DIS" name="QA2"
                              desc="Disconnector Bay1"/>
      </Bay>
    </VoltageLevel>
  </Substation>

  <!-- Communication: indirizzi IP e GOOSE/VLAN -->
  <Communication>
    <SubNetwork name="MMS_LAN" type="8-MMS" desc="MMS Client-Server LAN">
      <ConnectedAP iedName="BAY1_PROT" apName="S1">
        <Address>
          <P type="IP">192.168.1.10</P>
          <P type="IP-MASK">255.255.255.0</P>
          <P type="IP-GATEWAY">192.168.1.1</P>
        </Address>
      </ConnectedAP>
      <ConnectedAP iedName="BAY2_PROT" apName="S1">
        <Address>
          <P type="IP">192.168.1.11</P>
          <P type="IP-MASK">255.255.255.0</P>
        </Address>
      </ConnectedAP>
    </SubNetwork>
    <SubNetwork name="GOOSE_LAN" type="8-MMS"
                desc="GOOSE Process Bus VLAN 100">
      <ConnectedAP iedName="BAY1_PROT" apName="P1">
        <Address>
          <P type="MAC-Address">01-0C-CD-01-00-01</P>
          <P type="VLAN-ID">100</P>
          <P type="VLAN-PRIORITY">4</P>
          <P type="APPID">0001</P>
        </Address>
      </ConnectedAP>
    </SubNetwork>
  </Communication>

  <!-- IED: configurazione dispositivo BAY1_PROT -->
  <IED name="BAY1_PROT" type="SEL-487E"
       manufacturer="Schweitzer Engineering" configVersion="1.2">
    <AccessPoint name="S1">
      <Server>
        <Authentication none="true" password="false" certificate="true"/>
        <LDevice inst="PROT" desc="Protection Logical Device">
          <LN0 lnClass="LLN0" inst="" lnType="LLN0_type1">
            <DataSet name="DSStatusProt" desc="Protection Status Dataset">
              <FCDA ldInst="PROT" prefix="" lnClass="XCBR"
                    lnInst="1" doName="Pos" daName="stVal" fc="ST"/>
              <FCDA ldInst="PROT" prefix="" lnClass="XCBR"
                    lnInst="1" doName="Pos" daName="q" fc="ST"/>
            </DataSet>
            <GSEControl name="GCBStatusProt"
                        desc="GOOSE Control Block - Status"
                        type="GOOSE"
                        appID="GOOSE_BAY1_PROT_STATUS"
                        confRev="1"
                        datSet="DSStatusProt">
            </GSEControl>
          </LN0>
          <LN lnClass="XCBR" inst="1" lnType="XCBR_type1"
              prefix="" desc="Circuit Breaker 1">
          </LN>
          <LN lnClass="MMXU" inst="1" lnType="MMXU_type1"
              prefix="" desc="3-Phase Measurement Feeder1">
          </LN>
          <LN lnClass="PDIF" inst="1" lnType="PDIF_type1"
              prefix="" desc="Differential Protection">
          </LN>
        </LDevice>
      </Server>
    </AccessPoint>
  </IED>

  <!-- DataTypeTemplates: definizione tipi LN, DO, DA -->
  <DataTypeTemplates>
    <LNodeType id="XCBR_type1" lnClass="XCBR">
      <DO name="Pos" type="DPC_type"/>
      <DO name="BlkOpn" type="SPC_type"/>
      <DO name="BlkCls" type="SPC_type"/>
      <DO name="OpCnt" type="INS_type"/>
    </LNodeType>
    <DOType id="DPC_type" cdc="DPC">
      <DA name="stVal" bType="Dbpos" fc="ST"/>
      <DA name="q" bType="Quality" fc="ST"/>
      <DA name="t" bType="Timestamp" fc="ST"/>
      <DA name="ctlVal" bType="BOOLEAN" fc="CO"/>
    </DOType>
  </DataTypeTemplates>

</SCL>

"""
Parser SCL IEC 61850 - Estrae IED, Logical Devices e Logical Nodes
Richiede: pip install lxml
"""
from lxml import etree
from dataclasses import dataclass, field
from typing import Optional

# Namespace SCL standard
SCL_NS = "http://www.iec.ch/61850/2003/SCL"

@dataclass
class LogicalNode:
    ln_class: str
    inst: str
    ln_type: str
    prefix: str = ""
    desc: str = ""

@dataclass
class LogicalDevice:
    inst: str
    desc: str
    logical_nodes: list[LogicalNode] = field(default_factory=list)

@dataclass
class IEDConfig:
    name: str
    manufacturer: str
    ied_type: str
    logical_devices: list[LogicalDevice] = field(default_factory=list)

def parse_scl(file_path: str) -> list[IEDConfig]:
    """Parsa un file SCL e restituisce la lista degli IED configurati."""
    tree = etree.parse(file_path)
    root = tree.getroot()

    ns = {"scl": SCL_NS}
    ieds: list[IEDConfig] = []

    for ied_elem in root.findall("scl:IED", ns):
        ied = IEDConfig(
            name=ied_elem.get("name", ""),
            manufacturer=ied_elem.get("manufacturer", ""),
            ied_type=ied_elem.get("type", ""),
        )

        for ap in ied_elem.findall(".//scl:AccessPoint/scl:Server/scl:LDevice", ns):
            ld = LogicalDevice(
                inst=ap.get("inst", ""),
                desc=ap.get("desc", ""),
            )

            # LN0 (obbligatorio, sempre presente)
            ln0 = ap.find("scl:LN0", ns)
            if ln0 is not None:
                ld.logical_nodes.append(LogicalNode(
                    ln_class="LLN0",
                    inst="",
                    ln_type=ln0.get("lnType", ""),
                    desc=ln0.get("desc", ""),
                ))

            # LN instances
            for ln_elem in ap.findall("scl:LN", ns):
                ld.logical_nodes.append(LogicalNode(
                    ln_class=ln_elem.get("lnClass", ""),
                    inst=ln_elem.get("inst", ""),
                    ln_type=ln_elem.get("lnType", ""),
                    prefix=ln_elem.get("prefix", ""),
                    desc=ln_elem.get("desc", ""),
                ))

            ied.logical_devices.append(ld)

        ieds.append(ied)

    return ieds

def extract_goose_publishers(file_path: str) -> list[dict]:
    """Estrae tutti i GOOSE Control Block dal file SCD."""
    tree = etree.parse(file_path)
    root = tree.getroot()
    ns = {"scl": SCL_NS}

    goose_list = []

    for ied in root.findall("scl:IED", ns):
        ied_name = ied.get("name")
        for gse in ied.findall(".//scl:GSEControl", ns):
            goose_list.append({
                "ied": ied_name,
                "name": gse.get("name"),
                "appID": gse.get("appID"),
                "datSet": gse.get("datSet"),
                "confRev": gse.get("confRev"),
            })

    return goose_list

# Utilizzo
if __name__ == "__main__":
    ieds = parse_scl("substation_nord.scd")
    for ied in ieds:
        print(f"IED: {ied.name} ({ied.manufacturer})")
        for ld in ied.logical_devices:
            print(f"  LD: {ld.inst}")
            for ln in ld.logical_nodes:
                ref = f"{ied.name}/{ld.inst}/{ln.ln_class}{ln.inst}"
                print(f"    {ref}")

"""
Client MMS IEC 61850 con libiec61850 Python bindings
Richiede: pip install pyiec61850 oppure build da sorgenti
Documentazione: https://libiec61850.com/documentation/
"""
import iec61850  # Python wrapper per libiec61850
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class IEC61850Client:
    """Client MMS per comunicazione con IED IEC 61850."""

    def __init__(self, host: str, port: int = 102):
        self.host = host
        self.port = port
        self.connection: Optional[iec61850.IedConnection] = None

    def connect(self) -> bool:
        """Stabilisce connessione MMS con IED."""
        self.connection = iec61850.IedConnection_create()
        error = iec61850.IedClientError()
        iec61850.IedConnection_connect(
            self.connection, error, self.host, self.port
        )
        if error.value != iec61850.IED_ERROR_OK:
            logger.error(f"Connessione fallita a {self.host}:{self.port} "
                        f"- Errore: {error.value}")
            return False
        logger.info(f"Connesso a IED {self.host}:{self.port}")
        return True

    def read_boolean(self, object_ref: str) -> Optional[bool]:
        """Legge un attributo booleano (es. stato interruttore)."""
        error = iec61850.IedClientError()
        value = iec61850.IedConnection_readBooleanValue(
            self.connection, error, object_ref,
            iec61850.IEC61850_FC_ST  # Functional Constraint: ST = Status
        )
        if error.value != iec61850.IED_ERROR_OK:
            logger.error(f"Errore lettura {object_ref}: {error.value}")
            return None
        return bool(value)

    def read_float(self, object_ref: str,
                   fc: int = iec61850.IEC61850_FC_MX) -> Optional[float]:
        """Legge un valore float (es. potenza attiva)."""
        error = iec61850.IedClientError()
        value = iec61850.IedConnection_readFloatValue(
            self.connection, error, object_ref, fc
        )
        if error.value != iec61850.IED_ERROR_OK:
            return None
        return float(value)

    def subscribe_reports(self, rcb_ref: str,
                         callback, client_id: str = "SCADA_01") -> bool:
        """
        Iscrive a un Buffered Report Control Block per ricevere eventi.
        rcb_ref esempio: "BAY1_PROT/PROT/LLN0.BR.brcbStatus01"
        """
        error = iec61850.IedClientError()

        # Abilita il report
        iec61850.IedConnection_setRCBValues(
            self.connection, error, rcb_ref,
            iec61850.RCB_ELEMENT_RPT_ENA | iec61850.RCB_ELEMENT_DATA_SET,
            True, None
        )

        # Installa handler per i report in ingresso
        iec61850.IedConnection_installReportHandler(
            self.connection, rcb_ref, client_id, callback, None
        )

        if error.value != iec61850.IED_ERROR_OK:
            logger.error(f"Errore iscrizione RCB {rcb_ref}")
            return False

        logger.info(f"Iscritto a RCB: {rcb_ref}")
        return True

    def send_select_operate(self, ctrl_ref: str, value: bool) -> bool:
        """
        Invia un comando Select-Before-Operate (SBO) per controllare un
        interruttore. Più sicuro del Direct Operate per comandi critici.
        ctrl_ref esempio: "BAY1_PROT/PROT/CSWI1.Pos"
        """
        error = iec61850.IedClientError()
        ctrl = iec61850.ControlObjectClient_create(ctrl_ref, self.connection)

        # Prima la SELECT
        iec61850.ControlObjectClient_selectWithValue(ctrl, None)
        if error.value != iec61850.IED_ERROR_OK:
            logger.error(f"SELECT fallita per {ctrl_ref}")
            return False

        # Poi OPERATE entro il timeout (tipicamente 30s)
        iec61850.ControlObjectClient_operate(ctrl, value, 0)
        if error.value != iec61850.IED_ERROR_OK:
            logger.error(f"OPERATE fallita per {ctrl_ref}")
            return False

        logger.info(f"Comando SBO inviato: {ctrl_ref} = {value}")
        return True

    def disconnect(self):
        """Chiude la connessione MMS."""
        if self.connection:
            iec61850.IedConnection_destroy(self.connection)
            self.connection = None


# Report handler callback
def on_report_received(report, param):
    """Callback chiamata quando l'IED invia un report."""
    dataset_dir = iec61850.ClientReport_getDataSetDirectory(report)
    values = iec61850.ClientReport_getDataSetValues(report)

    logger.info(f"Report ricevuto - Dataset: {dataset_dir}")
    # Elabora i valori del report
    for i in range(iec61850.MmsValue_getArraySize(values)):
        element = iec61850.MmsValue_getElement(values, i)
        logger.info(f"  Valore[{i}]: {iec61850.MmsValue_printToBuffer(element, 256)}")


# Esempio utilizzo
if __name__ == "__main__":
    client = IEC61850Client(host="192.168.1.10", port=102)

    if client.connect():
        # Leggi stato interruttore
        breaker_pos = client.read_boolean(
            "BAY1_PROT/PROT/XCBR1.ST.Pos.stVal"
        )
        logger.info(f"Posizione interruttore: {'CHIUSO' if breaker_pos else 'APERTO'}")

        # Leggi potenza attiva [MW]
        active_power = client.read_float(
            "BAY1_PROT/MEAS/MMXU1.MX.TotW.mag.f"
        )
        logger.info(f"Potenza attiva: {active_power / 1e6:.2f} MW")

        # Iscrizione report per eventi di protezione
        client.subscribe_reports(
            "BAY1_PROT/PROT/LLN0.BR.brcbProt01",
            on_report_received,
            "SCADA_PRIMARY"
        )

        # Loop principale
        try:
            while True:
                time.sleep(0.1)  # 100ms polling loop
        except KeyboardInterrupt:
            pass
        finally:
            client.disconnect()

Architettura GOOSE Publisher/Subscriber

  ┌─────────────────┐    GOOSE multicast     ┌─────────────────┐
  │  IED Publisher  │ ──────────────────────▶│ IED Subscriber 1│
  │  (BAY1_PROT)   │    EtherType: 0x88B8   │  (BAY2_PROT)    │
  │                 │    MAC: 01-0C-CD-01-*   └─────────────────┘
  │  APPID: 0x0001  │    VLAN: 100           ┌─────────────────┐
  │  confRev: 5     │    PCP: 4 (high prio) ─│ IED Subscriber 2│
  └─────────────────┘                         │  (BAY3_PROT)    │
                                              └─────────────────┘

Frame Ethernet GOOSE:
┌──────────┬──────────┬────────┬───────┬─────────────────────────┐
│ Dest MAC │ Src MAC  │802.1Q  │EthType│     GOOSE PDU (ASN.1)   │
│6 bytes   │6 bytes   │4 bytes │0x88B8 │  appID|len|reserved|PDU │
└──────────┴──────────┴────────┴───────┴─────────────────────────┘

GOOSE PDU (contenuto):
- gocbRef:   riferimento al GOOSE Control Block
- timeAllowedToLive: timeout per ritenere il publisher down [ms]
- datSet:    riferimento al dataset
- goID:      identificatore univoco
- t:         timestamp dell'evento (UTC con ms)
- stNum:     state number (incrementa ad ogni cambio di stato)
- sqNum:     sequence number (incrementa ad ogni retransmission)
- simulation: flag per modalità test
- confRev:   revision numero della configurazione
- ndsCom:    needs commissioning flag
- numDatSetEntries: numero di entry nel dataset
- allData:   i valori del dataset (array MMS)

Retransmission GOOSE dopo un evento (stato cambio a t=0):

  t=0ms:      TX immediata (stNum=N, sqNum=0)
  t=2ms:      TX (stNum=N, sqNum=1)   - T1 = 2ms
  t=6ms:      TX (stNum=N, sqNum=2)   - T1 = 4ms
  t=14ms:     TX (stNum=N, sqNum=3)   - T1 = 8ms
  t=30ms:     TX (stNum=N, sqNum=4)   - T1 = 16ms
  t=62ms:     TX (stNum=N, sqNum=5)   - T1 = 32ms
  ...
  t=MaxTx:    TX periodica con T0 (es. 1000ms, "heartbeat")

  stNum:  incrementa SOLO quando lo stato cambia
  sqNum:  incrementa ad ogni retransmission dello stesso stato

  Un subscriber che riceve GOOSE con sqNum=0 e stNum diverso
  dal precedente sa che c'è stato un cambio di stato.
  Un subscriber che non riceve per più di timeAllowedToLive
  mette il segnale in qualità "LOST".

"""
GOOSE Analyzer - Cattura e analizza frame GOOSE IEC 61850
Richiede: pip install scapy pyasn1
Richiede privilegio root/admin per cattura live
"""
from scapy.all import sniff, Ether, Dot1Q, Raw
from pyasn1.codec.ber import decoder as ber_decoder
from pyasn1.type import univ
import struct
import datetime
import logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)

# EtherType GOOSE
GOOSE_ETHERTYPE = 0x88B8

@dataclass
class GOOSEFrame:
    """Rappresentazione di un frame GOOSE decodificato."""
    src_mac: str
    dst_mac: str
    vlan_id: int
    vlan_priority: int
    app_id: int
    gocb_ref: str
    time_allowed_to_live: int
    dat_set: str
    go_id: str
    timestamp: datetime.datetime
    st_num: int        # State number
    sq_num: int        # Sequence number
    simulation: bool
    conf_rev: int
    num_entries: int
    raw_values: list


def parse_goose_time(raw_t: bytes) -> datetime.datetime:
    """Decodifica timestamp GOOSE (6 byte: 4 secondi + 3 fraz + flags)."""
    if len(raw_t) < 4:
        return datetime.datetime.utcnow()
    seconds = struct.unpack(">I", raw_t[:4])[0]
    # Epoch GOOSE: 1970-01-01 UTC (stesso Unix epoch)
    return datetime.datetime.utcfromtimestamp(seconds)


def decode_goose_pdu(payload: bytes) -> Optional[GOOSEFrame]:
    """
    Decodifica il PDU GOOSE dal payload Ethernet.
    Il formato e: APPID(2) + Length(2) + Reserved1(2) + Reserved2(2) + ASN.1 BER
    """
    if len(payload) < 8:
        return None

    app_id = struct.unpack(">H", payload[0:2])[0]
    length = struct.unpack(">H", payload[2:4])[0]
    # payload[4:8] = Reserved1 + Reserved2

    asn1_data = payload[8:length]

    try:
        # ASN.1 BER decoding del PDU GOOSE
        # Tag 0x61 = IECGoosePdu (context class, primitive)
        goose_pdu = {
            "app_id": app_id,
            "gocb_ref": "",
            "time_allowed_to_live": 0,
            "dat_set": "",
            "go_id": "",
            "st_num": 0,
            "sq_num": 0,
            "simulation": False,
            "conf_rev": 0,
            "num_entries": 0,
        }

        # Parsing manuale TLV (Type-Length-Value) BER per tag GOOSE
        idx = 0
        # Skip outer GOOSE PDU wrapper (tag 0x61)
        if asn1_data[0] == 0x61:
            if asn1_data[1] > 127:  # long form length
                len_bytes = asn1_data[1] - 128
                idx = 2 + len_bytes
            else:
                idx = 2

        while idx < len(asn1_data) - 1:
            tag = asn1_data[idx]
            idx += 1
            length_byte = asn1_data[idx]
            idx += 1

            if length_byte > 127:
                extra = length_byte - 128
                value_length = int.from_bytes(asn1_data[idx:idx+extra], 'big')
                idx += extra
            else:
                value_length = length_byte

            value = asn1_data[idx:idx+value_length]
            idx += value_length

            # Tag mapping IEC 61850 GOOSE PDU
            if tag == 0x80:    # gocbRef [0]
                goose_pdu["gocb_ref"] = value.decode("ascii", errors="replace")
            elif tag == 0x81:  # timeAllowedToLive [1]
                goose_pdu["time_allowed_to_live"] = int.from_bytes(value, 'big')
            elif tag == 0x82:  # datSet [2]
                goose_pdu["dat_set"] = value.decode("ascii", errors="replace")
            elif tag == 0x83:  # goID [3]
                goose_pdu["go_id"] = value.decode("ascii", errors="replace")
            elif tag == 0x84:  # t [4] timestamp
                goose_pdu["timestamp"] = parse_goose_time(value)
            elif tag == 0x85:  # stNum [5]
                goose_pdu["st_num"] = int.from_bytes(value, 'big')
            elif tag == 0x86:  # sqNum [6]
                goose_pdu["sq_num"] = int.from_bytes(value, 'big')
            elif tag == 0x87:  # simulation [7]
                goose_pdu["simulation"] = value[0] != 0
            elif tag == 0x88:  # confRev [8]
                goose_pdu["conf_rev"] = int.from_bytes(value, 'big')
            elif tag == 0x8A:  # numDatSetEntries [10]
                goose_pdu["num_entries"] = int.from_bytes(value, 'big')

        return goose_pdu

    except Exception as e:
        logger.warning(f"Errore parsing GOOSE PDU: {e}")
        return None


def goose_packet_handler(packet):
    """Handler Scapy per ogni pacchetto catturato."""
    if Ether in packet:
        eth = packet[Ether]
        payload = bytes(packet[Raw]) if Raw in packet else b""

        # Verifica EtherType GOOSE (con o senza VLAN tag)
        ethertype = eth.type
        vlan_id = 0
        vlan_prio = 0

        if ethertype == 0x8100 and Dot1Q in packet:  # VLAN tagged
            vlan_id = packet[Dot1Q].vlan
            vlan_prio = packet[Dot1Q].prio
            ethertype = packet[Dot1Q].type
            payload = bytes(packet[Dot1Q].payload)

        if ethertype != GOOSE_ETHERTYPE:
            return

        goose_pdu = decode_goose_pdu(payload)
        if goose_pdu:
            sim_flag = "[SIM]" if goose_pdu.get("simulation") else ""
            print(f"{sim_flag} GOOSE | "
                  f"VLAN={vlan_id}(prio={vlan_prio}) | "
                  f"AppID=0x{goose_pdu['app_id']:04X} | "
                  f"GoCBRef={goose_pdu['gocb_ref']} | "
                  f"stNum={goose_pdu['st_num']} sqNum={goose_pdu['sq_num']} | "
                  f"TAL={goose_pdu['time_allowed_to_live']}ms")


def capture_live(interface: str = "eth0", count: int = 100):
    """Cattura GOOSE live da interfaccia di rete."""
    print(f"Avvio cattura GOOSE su {interface}...")
    sniff(
        iface=interface,
        prn=goose_packet_handler,
        count=count,
        store=False,
    )


def analyze_pcap(file_path: str):
    """Analizza frame GOOSE da file pcap."""
    from scapy.all import rdpcap
    packets = rdpcap(file_path)
    for pkt in packets:
        goose_packet_handler(pkt)


if __name__ == "__main__":
    # Analisi da file pcap (per test senza hardware)
    analyze_pcap("goose_capture_substation.pcap")

"""
Gateway semplificato IEC 61850 MMS -> IEC 104
Gestisce il mapping e la pubblicazione verso SCADA
Dipendenze: pip install pyiec61850 lib60870-python
"""
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Callable
from datetime import datetime

logger = logging.getLogger(__name__)


@dataclass
class DataPoint:
    """Punto dati IEC 104 mappato da un riferimento IEC 61850."""
    # IEC 104
    common_address: int     # ASDU address
    ioa: int                # Information Object Address
    type_id: int            # M_ME_TF_1 (float), M_SP_TB_1 (bool), etc.

    # IEC 61850 sorgente
    ied_host: str
    object_ref: str         # es. BAY1_PROT/PROT/MMXU1.MX.TotW.mag.f
    fc: str                 # ST, MX, CF...

    # Stato corrente
    value: float = 0.0
    quality: int = 0        # 0 = good, 1 = invalid
    timestamp: datetime = field(default_factory=datetime.utcnow)

    # Callback per notifica cambio valore
    change_callbacks: list[Callable] = field(default_factory=list)


class IEC61850ToIEC104Gateway:
    """
    Gateway che mappa dati IEC 61850 verso Information Objects IEC 104.
    Architettura:
      IED IEC61850 --[MMS/GOOSE]--> Gateway --[IEC104]--> SCADA
    """

    def __init__(self):
        self.data_points: dict[int, DataPoint] = {}  # IOA -> DataPoint
        self._61850_clients: dict[str, object] = {}
        self._iec104_server = None
        self._running = False

    def add_mapping(self, point: DataPoint):
        """Aggiunge un mapping IEC61850 -> IEC104."""
        self.data_points[point.ioa] = point
        logger.info(
            f"Mapping aggiunto: IOA={point.ioa} "
            f"<-- {point.ied_host}/{point.object_ref}"
        )

    async def start(self):
        """Avvia il gateway."""
        self._running = True
        # Raggruppa i data points per IED host
        ied_groups: dict[str, list[DataPoint]] = {}
        for dp in self.data_points.values():
            ied_groups.setdefault(dp.ied_host, []).append(dp)

        # Connetti a ogni IED in parallelo
        tasks = [
            self._connect_and_poll_ied(host, points)
            for host, points in ied_groups.items()
        ]
        await asyncio.gather(*tasks)

    async def _connect_and_poll_ied(
        self, host: str, points: list[DataPoint]
    ):
        """Connette a un IED e mantiene i valori aggiornati."""
        logger.info(f"Connessione a IED {host}...")
        # In produzione: usa libiec61850 per connessione reale
        # Qui mostriamo la logica di polling
        while self._running:
            try:
                for dp in points:
                    # Simula lettura MMS
                    new_value = await self._read_ied_value(host, dp)
                    if new_value is not None and new_value != dp.value:
                        old_value = dp.value
                        dp.value = new_value
                        dp.timestamp = datetime.utcnow()
                        dp.quality = 0  # Good quality
                        logger.debug(
                            f"IOA {dp.ioa}: "
                            f"{old_value} -> {new_value}"
                        )
                        # Notifica il server IEC 104
                        await self._publish_to_iec104(dp)

                await asyncio.sleep(0.5)  # 500ms polling cycle

            except ConnectionError as e:
                logger.error(f"IED {host} disconnesso: {e}")
                # Marca tutti i punti come invalid
                for dp in points:
                    dp.quality = 0x20  # INVALID flag IEC 104
                    await self._publish_to_iec104(dp)
                await asyncio.sleep(5)  # Retry dopo 5s

    async def _read_ied_value(
        self, host: str, dp: DataPoint
    ) -> Optional[float]:
        """Legge un valore da IED via MMS."""
        # Implementazione reale usa libiec61850
        # Per test: return mock value
        return None

    async def _publish_to_iec104(self, dp: DataPoint):
        """Pubblica un cambio di valore verso il server IEC 104."""
        if self._iec104_server:
            # lib60870 API per aggiornare il punto nell'ASDU
            logger.debug(
                f"IEC104 update: CA={dp.common_address} "
                f"IOA={dp.ioa} v={dp.value}"
            )

    def get_point_value(self, ioa: int) -> Optional[DataPoint]:
        """Restituisce il valore corrente di un punto per IOA."""
        return self.data_points.get(ioa)


# Configurazione mapping esempio
gateway = IEC61850ToIEC104Gateway()

# Potenza attiva BAY1 -> IOA 1001
gateway.add_mapping(DataPoint(
    common_address=1,
    ioa=1001,
    type_id=0x0D,        # M_ME_TF_1: Short Float + Timestamp
    ied_host="192.168.1.10",
    object_ref="BAY1_PROT/MEAS/MMXU1.MX.TotW.mag.f",
    fc="MX",
))

# Stato interruttore BAY1 -> IOA 2001
gateway.add_mapping(DataPoint(
    common_address=1,
    ioa=2001,
    type_id=0x03,        # M_SP_TB_1: Single-point + Timestamp
    ied_host="192.168.1.10",
    object_ref="BAY1_PROT/PROT/XCBR1.ST.Pos.stVal",
    fc="ST",
))

"""
Configurazione TLS per connessione MMS sicura
Basato su IEC 62351-3 requirements
"""
import ssl
import socket
from pathlib import Path

def create_iec62351_tls_context(
    cert_file: str,
    key_file: str,
    ca_cert_file: str,
    require_client_cert: bool = True
) -> ssl.SSLContext:
    """
    Crea un contesto TLS conforme IEC 62351-3.

    Certificati devono essere X.509 v3 con:
    - Subject: CN=<IED-name>, O=<Utility-name>
    - Extended Key Usage: id-kp-serverAuth (server) / id-kp-clientAuth (client)
    - Key: RSA 2048+ o ECDSA P-256+
    """
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

    # TLS 1.2 minimo, TLS 1.3 preferito
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2

    # Suite crittografiche approvate IEC 62351
    # Priorità: ECDHE > DHE, AES-256-GCM > AES-128-GCM
    ctx.set_ciphers(
        "ECDHE-ECDSA-AES256-GCM-SHA384:"
        "ECDHE-RSA-AES256-GCM-SHA384:"
        "ECDHE-ECDSA-AES128-GCM-SHA256:"
        "DHE-RSA-AES256-GCM-SHA384"
    )

    # Certificato del dispositivo locale
    ctx.load_cert_chain(certfile=cert_file, keyfile=key_file)

    # CA trust store per verificare il peer
    ctx.load_verify_locations(cafile=ca_cert_file)

    # Autenticazione mutua obbligatoria (client cert required)
    if require_client_cert:
        ctx.verify_mode = ssl.CERT_REQUIRED

    # Disabilita verifiche non sicure
    ctx.check_hostname = True
    ctx.verify_mode = ssl.CERT_REQUIRED

    return ctx


def check_tls_compliance(cert_path: str) -> dict:
    """
    Verifica la conformità di un certificato X.509 con IEC 62351.
    Controlla validita, algoritmo chiave, estensioni richieste.
    """
    import subprocess
    import json

    result = subprocess.run(
        ["openssl", "x509", "-in", cert_path,
         "-noout", "-text", "-nameopt", "RFC2253"],
        capture_output=True, text=True
    )

    report = {
        "file": cert_path,
        "tls12_compliant": False,
        "key_length_ok": False,
        "not_expired": False,
        "has_client_auth": False,
        "details": result.stdout[:500]
    }

    output = result.stdout
    # Chiave RSA >= 2048 bit o ECDSA P-256
    if "Public-Key: (2048 bit)" in output or \
       "Public-Key: (3072 bit)" in output or \
       "P-256" in output:
        report["key_length_ok"] = True

    # TLS Client Authentication estensione
    if "TLS Web Client Authentication" in output:
        report["has_client_auth"] = True

    report["tls12_compliant"] = (
        report["key_length_ok"] and report["has_client_auth"]
    )

    return report


# Network Segmentation - zone di sicurezza IEC 62351
SECURITY_ZONES = {
    "zone_0_internet": {
        "access": "none",
        "description": "Internet - accesso vietato dalla sottostazione"
    },
    "zone_1_enterprise": {
        "access": "through_dmz_only",
        "description": "Rete aziendale - solo via DMZ con proxy"
    },
    "zone_2_control": {
        "access": "restricted",
        "description": "SCADA e sistemi di controllo - solo TLS mutual auth"
    },
    "zone_3_substation": {
        "access": "local_only",
        "description": "LAN sottostazione - MMS + GOOSE, firewall dedicato"
    },
    "zone_4_process": {
        "access": "air_gapped_preferred",
        "description": "Process bus - solo GOOSE/SV, nessun IP routing"
    }
}

"""
Suite di test automatizzati per conformità GOOSE IEC 61850
Verifica: timing, retransmission, stNum/sqNum logic, TAL
"""
import time
import threading
import unittest
from dataclasses import dataclass
from scapy.all import sniff, sendp, Ether, Dot1Q, Raw
import struct

@dataclass
class GOOSETestResult:
    test_name: str
    passed: bool
    measured_latency_ms: float = 0.0
    expected_latency_ms: float = 4.0
    details: str = ""


class GOOSEConformanceTest(unittest.TestCase):
    """
    Test di conformità GOOSE secondo IEC 61850-8-1.
    Richiede accesso diretto all'interfaccia di rete.
    """

    def setUp(self):
        """Configura l'interfaccia e i parametri di test."""
        self.interface = "eth0"        # Interfaccia collegata alla rete GOOSE
        self.goose_vlan = 100
        self.goose_appid = 0x0001
        self.captured_frames: list[dict] = []
        self.capture_lock = threading.Lock()

    def _start_capture(self, duration_s: float = 5.0):
        """Avvia cattura GOOSE in background."""
        def capture():
            from et_iec61850_test_utils import goose_packet_handler
            sniff(
                iface=self.interface,
                prn=lambda p: self._capture_handler(p),
                timeout=duration_s,
                store=False
            )

        t = threading.Thread(target=capture, daemon=True)
        t.start()
        return t

    def _capture_handler(self, packet):
        """Cattura frame GOOSE e aggiunge alla lista."""
        if Ether in packet and Raw in packet:
            # Verifica EtherType GOOSE (semplificato)
            raw = bytes(packet)
            if len(raw) > 14:
                with self.capture_lock:
                    self.captured_frames.append({
                        "timestamp": time.time(),
                        "raw": raw,
                        "length": len(raw)
                    })

    def test_retransmission_pattern(self):
        """
        Verifica che il pattern di retransmission segua la
        progressione esponenziale definita in IEC 61850-8-1.
        T1, T2=2*T1, T4=2*T2, fino a T0 (heartbeat).
        """
        # Reset frame list
        self.captured_frames = []

        # Avvia cattura per 2 secondi dopo evento
        capture_thread = self._start_capture(2.0)

        # Attendi evento GOOSE dall'IED (o inietta evento di test)
        # In un test reale si aspetta un cambio di stato noto
        capture_thread.join(timeout=3.0)

        # Analizza intervalli di retransmission
        if len(self.captured_frames) < 3:
            self.skipTest("Insufficienti frame catturati")

        intervals = []
        for i in range(1, len(self.captured_frames)):
            dt_ms = (self.captured_frames[i]["timestamp"] -
                     self.captured_frames[i-1]["timestamp"]) * 1000
            intervals.append(dt_ms)

        # Verifica progressione: ogni intervallo deve essere > del precedente
        # (fase di retransmission) fino al heartbeat
        if len(intervals) >= 2:
            # I primi intervalli dovrebbero essere piccoli (< 50ms)
            self.assertLess(
                intervals[0], 50.0,
                f"Prima retransmission troppo lenta: {intervals[0]:.1f}ms"
            )

    def test_initial_latency_requirement(self):
        """
        Verifica che la latenza iniziale GOOSE sia < 4ms
        per messaggi di categoria P1 (protezione critica).
        Nota: richiede hardware preciso per misurazione accurata.
        """
        # In un test reale si usa un timestamp di riferimento
        # dall'IED (campo 't' nel PDU GOOSE)
        # Qui simuliamo la logica di verifica
        simulated_latency_ms = 2.3  # Da misurare con hardware preciso

        result = GOOSETestResult(
            test_name="Initial Latency P1",
            passed=simulated_latency_ms < 4.0,
            measured_latency_ms=simulated_latency_ms,
            expected_latency_ms=4.0,
            details=f"Latenza: {simulated_latency_ms}ms (max: 4ms)"
        )

        self.assertTrue(
            result.passed,
            f"Latenza GOOSE {result.measured_latency_ms}ms supera 4ms"
        )

    def test_timeallowedtolive_validity(self):
        """
        Verifica che TAL (TimeAllowedToLive) sia configurato
        ragionevolmente rispetto al periodo di heartbeat T0.
        Regola: TAL deve essere > 2 * T0 per tollerare una
        perdita di heartbeat.
        """
        # Valori tipici: T0=1000ms, TAL=2000ms
        t0_ms = 1000   # Periodo heartbeat
        tal_ms = 2000  # TimeAllowedToLive dal GOOSE PDU

        self.assertGreater(
            tal_ms, 2 * t0_ms,
            f"TAL {tal_ms}ms dovrebbe essere > 2*T0={2*t0_ms}ms"
        )

    def test_vlan_priority(self):
        """
        Verifica che i frame GOOSE abbiano la priorità VLAN
        corretta (PCP=4 per protezione, come da IEC 61850-8-1).
        """
        # Costruisci un frame di test GOOSE e verifica
        test_frame = (
            Ether(dst="01:0c:cd:01:00:01", src="00:11:22:33:44:55") /
            Dot1Q(vlan=100, prio=4) /
            Raw(b"\x88\xb8\x00\x01\x00\x10\x00\x00\x00\x00")
        )

        # Estrai VLAN priority dal frame
        if Dot1Q in test_frame:
            pcp = test_frame[Dot1Q].prio
            self.assertEqual(
                pcp, 4,
                f"VLAN PCP deve essere 4 per GOOSE, trovato: {pcp}"
            )


if __name__ == "__main__":
    unittest.main(verbosity=2)

Evoluzione architetturale della sottostazione:

GENERAZIONE 1 (pre-2005): Wired Hardened
┌─────────────────────────────────────────┐
│  CT/VT ──wire──▶ Relay ──wire──▶ CB     │
│  Cavi di rame per ogni segnale          │
│  Manutenzione: sostituzione cavi        │
└─────────────────────────────────────────┘

GENERAZIONE 2 (2005-2015): IEC 61850 Edition 1
┌──────────────────────────────────────────┐
│  CT/VT ──▶ IED ──[MMS/GOOSE Ethernet]──▶│
│  Process Bus opzionale                   │
│  SCADA via MMS                           │
└──────────────────────────────────────────┘

GENERAZIONE 3 (2015-2023): Full Digital
┌──────────────────────────────────────────────┐
│  CT/VT ──▶ Merging Unit ──[SV]──▶ IED       │
│  Process Bus: solo fibra ottica              │
│  Station Bus: Ethernet ridondato (PRP/HSR)   │
│  Gateway verso SCADA/EMS                     │
└──────────────────────────────────────────────┘

GENERAZIONE 4 (2023+): Software-Defined Substation
┌────────────────────────────────────────────────────┐
│  Merging Unit ──[SV via TSN]──▶ COMPUTE CLUSTER   │
│   Funzioni protezione su software (IEC 61850-7-90) │
│   Containerizzazione (K3s edge cluster)            │
│   Digital Twin real-time                           │
│   Edge AI per fault analysis                       │
│   API REST/gRPC verso cloud EMS/DERMS              │
└────────────────────────────────────────────────────┘

# docker-compose.yml per ambiente di sviluppo/test digital substation
# Simula un ambiente di sottostazione con IED virtuali e SCADA

version: "3.8"

services:
  # IED Simulato - Publisher GOOSE e Server MMS
  ied-simulator:
    image: mzautomation/iec61850-simulator:latest
    container_name: ied_bay1_prot
    networks:
      - process_bus
      - station_bus
    environment:
      - IED_NAME=BAY1_PROT
      - IED_IP=192.168.10.10
      - SCL_FILE=/config/bay1_prot.cid
      - GOOSE_INTERFACE=eth1
    volumes:
      - ./config/scl:/config:ro
    cap_add:
      - NET_ADMIN     # Per GOOSE multicast su Ethernet L2

  # Gateway IEC 61850 -> IEC 104
  gateway:
    build: ./gateway
    container_name: iec61850_gateway
    networks:
      - station_bus
      - scada_net
    environment:
      - IED1_HOST=192.168.10.10
      - IED1_PORT=102
      - IEC104_PORT=2404
      - MAPPING_FILE=/config/mapping.json
    volumes:
      - ./config/gateway:/config:ro
    depends_on:
      - ied-simulator

  # SCADA/Historian - InfluxDB + Grafana
  influxdb:
    image: influxdb:2.7
    container_name: historian
    networks:
      - scada_net
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_ORG=utility
      - DOCKER_INFLUXDB_INIT_BUCKET=substation_data
    volumes:
      - influx_data:/var/lib/influxdb2

  grafana:
    image: grafana/grafana:latest
    container_name: scada_dashboard
    networks:
      - scada_net
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=substation123
    depends_on:
      - influxdb

  # SCL Validator - OpenSCD headless
  scl-validator:
    image: openscd/open-scd-cli:latest
    container_name: scl_validator
    volumes:
      - ./config/scl:/workspace
    command: ["validate", "/workspace/substation.scd"]

networks:
  process_bus:
    driver: macvlan          # L2 nativo per GOOSE
    driver_opts:
      parent: eth1
  station_bus:
    driver: bridge
    ipam:
      config:
        - subnet: 192.168.10.0/24
  scada_net:
    driver: bridge
    ipam:
      config:
        - subnet: 192.168.20.0/24

volumes:
  influx_data: