feat: データモデルを設定駆動の DataSchema に汎用化

config/blockchain.json

@@ -4,5 +4,19 @@
   "export_enabled": true,
   "min_verified_receives": 3,
   "require_content_hash_agreement": true,
-  "min_distinct_devices_for_aggregate": 2
+  "min_distinct_devices_for_aggregate": 2,
+  "data_schema": {
+    "gakuseki_column": "gakuseki",
+    "bt_addr_column": "bt_addrs",
+    "csv_identity_rename": {
+      "学籍番号": "gakuseki"
+    },
+    "output_columns": [
+      "gakuseki",
+      "bt_addrs",
+      "device_name"
+    ],
+    "input_field": "gakuseki",
+    "output_field": "bt_addrs"
+  }
 }

src/ble_blockchain/app/main.py

@@ -28,6 +28,8 @@
 from ble_blockchain.paths import repo_root
 from ble_blockchain.pipeline.delete_excess_data import delete_excess_data
 from ble_blockchain.pipeline.pandas_d_encode import pandas_decode, pandas_encode
+from ble_blockchain.types import ReceivedPayload
+
 RUNTIME_PROFILES_PATH = repo_root() / "config/runtime_profiles.json"
 
 
@@ -70,7 +72,7 @@ def run_communication_steps(
     profile: dict[str, Any],
     settings: DeviceSettings,
     tanmatsu_bt_addrs: list[str],
-    receive_data_list: list[Any],
+    receive_data_list: list[ReceivedPayload],
 ) -> None:
     """Execute discoverable, send, receive, and sleep steps from a profile."""
     defaults = {"sleep_seconds": 30}
@@ -108,7 +110,7 @@ def run_communication_steps(
 
 def process_received_payload(
     raw: bytes, trusted_peer_pems: frozenset[str]
-) -> list[Any]:
+) -> ReceivedPayload:
     """Decrypt, verify, and decode one received payload."""
     try:
         payload = unpack(raw)
@@ -128,19 +130,33 @@ def process_received_payload(
 
         content_hash = payload_content_hash(plaintext)
         if not verified:
-            return [None, public_key, payload.signature, False, None, None]
-
-        return [
-            df,
-            public_key,
-            payload.signature,
-            verified,
-            payload.public_key_pem,
-            content_hash,
-        ]
+            return ReceivedPayload(
+                df=None,
+                public_key=public_key,
+                signature=payload.signature,
+                verified=False,
+                public_key_pem=None,
+                payload_content_hash=None,
+            )
+
+        return ReceivedPayload(
+            df=df,
+            public_key=public_key,
+            signature=payload.signature,
+            verified=verified,
+            public_key_pem=payload.public_key_pem,
+            payload_content_hash=content_hash,
+        )
     except (ValueError, KeyError, json.JSONDecodeError) as exc:
         print(f"受信ペイロードの処理に失敗しました: {exc}")
-        return [None, None, b"", False, None, None]
+        return ReceivedPayload(
+            df=None,
+            public_key=None,
+            signature=b"",
+            verified=False,
+            public_key_pem=None,
+            payload_content_hash=None,
+        )
 
 
 def run_pipeline(  # pylint: disable=too-many-locals
@@ -158,7 +174,7 @@ def run_pipeline(  # pylint: disable=too-many-locals
 
     print(settings.tanmatsu_bt_addrs)
 
-    receive_data_list: list[Any] = []
+    receive_data_list: list[ReceivedPayload] = []
 
     run_communication_steps(
         profile, settings, settings.tanmatsu_bt_addrs, receive_data_list

src/ble_blockchain/blockchain/myblock.py

@@ -8,13 +8,16 @@
 import math
 from collections import Counter
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Callable
 
 import pandas as pd
 
-from ble_blockchain.config.loader import load_blockchain_config
-from ble_blockchain.pipeline.delete_excess_data import filter_registered_data
+from ble_blockchain.config.loader import (
+    BlockchainConfig,
+    load_blockchain_config,
+)
 from ble_blockchain.pipeline.pandas_d_encode import pandas_encode
+from ble_blockchain.types import ReceivedPayload
 
 GENESIS_PREV_HASH = (
     "747bc42088cf0b3915982af289189e8f14d3325a7d594bc2d30a7014a536cb13"
@@ -59,11 +62,25 @@ def payload_content_hash(plaintext: bytes) -> str:
 class MyBlockChain:
     """Build and validate a chain of majority-adopted BLE observations."""
 
-    def __init__(self) -> None:
+    def __init__(
+        self,
+        *,
+        filter_registered: Callable[[pd.DataFrame], pd.DataFrame] | None = None,
+    ) -> None:
         self.chain: list[dict[str, Any]] = []
         self._last_threshold: int = 0
         self._last_verified_count: int = 0
         self._verified_entries: list[VerifiedReceive] = []
+        if filter_registered is not None:
+            self._filter_registered = filter_registered
+        else:
+            # Lazily resolve the default so blockchain does not import
+            # pipeline at module load time (caller may inject a filter).
+            from ble_blockchain.pipeline.delete_excess_data import (  # pylint: disable=import-outside-toplevel
+                filter_registered_data,
+            )
+
+            self._filter_registered = filter_registered_data
 
     @property
     def last_majority_threshold(self) -> int:
@@ -115,10 +132,11 @@ def add_new_block(
         return new_block
 
     def build_from_receives(  # pylint: disable=too-many-locals
-        self, receive_data_list: list[list[Any]]
+        self, receive_data_list: list[ReceivedPayload]
     ) -> None:
-        """Build blocks from verified receive tuples using majority rules."""
+        """Build blocks from verified receive payloads using majority rules."""
         config = load_blockchain_config()
+        schema = config.data_schema
         verified_entries = self._parse_verified_receives(receive_data_list)
         self._verified_entries = verified_entries
 
@@ -135,12 +153,12 @@ def build_from_receives(  # pylint: disable=too-many-locals
 
         bt_addrs_seen: set[str] = set()
         for entry in verified_entries:
-            for bt_addr in entry.df["bt_addrs"].astype(str).unique():
+            for bt_addr in entry.df[schema.bt_addr_column].astype(str).unique():
                 bt_addrs_seen.add(bt_addr)
 
         for bt_addr in sorted(bt_addrs_seen):
             if config.one_block_per_bt_addr and any(
-                block["tran_body"]["output"]["bt_addrs"] == bt_addr
+                block["tran_body"]["output"][schema.output_field] == bt_addr
                 for block in self.chain
             ):
                 continue
@@ -151,48 +169,47 @@ def build_from_receives(  # pylint: disable=too-many-locals
             if adoption is None:
                 continue
 
-            gakuseki, reporter_count, reporters, content_hash, gakuseki_votes = (
+            identity, reporter_count, reporters, content_hash, identity_votes = (
                 adoption
             )
             tran_meta = {
                 "count": reporter_count,
                 "majority_threshold": threshold,
                 "content_hash": content_hash,
-                "gakuseki_votes": gakuseki_votes,
+                "gakuseki_votes": identity_votes,
                 "reporters": reporters,
             }
-            inp = {"gakuseki": gakuseki}
-            out = {"bt_addrs": bt_addr, "count": reporter_count}
+            inp = {schema.input_field: identity}
+            out = {schema.output_field: bt_addr, "count": reporter_count}
             self.add_new_block(inp, out, tran_meta=tran_meta)
 
     def _parse_verified_receives(
-        self, receive_data_list: list[list[Any]]
+        self, receive_data_list: list[ReceivedPayload]
     ) -> list[VerifiedReceive]:
         verified_entries: list[VerifiedReceive] = []
 
         for item in receive_data_list:
-            if len(item) < 4 or not item[3]:
+            if not item.verified:
                 continue
-            if item[0] is None:
+            if item.df is None:
                 continue
 
-            df = filter_registered_data(item[0])
+            df = self._filter_registered(item.df)
             if df.empty:
                 continue
 
-            if len(item) < 6:
+            if item.public_key_pem is None or item.payload_content_hash is None:
                 continue
 
-            public_key_pem = str(item[4])
-            declared_hash = str(item[5])
+            declared_hash = str(item.payload_content_hash)
             encoded_hash = payload_content_hash(pandas_encode(df))
             if declared_hash != encoded_hash:
                 continue
 
             verified_entries.append(
                 VerifiedReceive(
                     df=df,
-                    pubkey_fingerprint=pubkey_fingerprint(public_key_pem),
+                    pubkey_fingerprint=pubkey_fingerprint(str(item.public_key_pem)),
                     payload_content_hash=declared_hash,
                 )
             )
@@ -204,12 +221,13 @@ def _compute_adoption_for_bt_addr(  # pylint: disable=too-many-locals
         verified_entries: list[VerifiedReceive],
         bt_addr: str,
         threshold: int,
-        config: Any,
+        config: BlockchainConfig,
     ) -> tuple[str, int, list[dict[str, str]], str, dict[str, int]] | None:
         """Return adoption data for one bt_addr when threshold is met."""
+        schema = config.data_schema
         reporters_for_addr: list[VerifiedReceive] = []
         for entry in verified_entries:
-            if bt_addr in entry.df["bt_addrs"].astype(str).values:
+            if bt_addr in entry.df[schema.bt_addr_column].astype(str).values:
                 reporters_for_addr.append(entry)
 
         unique_by_fingerprint: dict[str, VerifiedReceive] = {}
@@ -226,21 +244,23 @@ def _compute_adoption_for_bt_addr(  # pylint: disable=too-many-locals
             return None
         content_hash = next(iter(content_hashes))
 
-        gakuseki_votes: Counter[str] = Counter()
+        identity_votes: Counter[str] = Counter()
         for entry in unique_entries:
-            rows = entry.df[entry.df["bt_addrs"].astype(str) == bt_addr]
+            rows = entry.df[
+                entry.df[schema.bt_addr_column].astype(str) == bt_addr
+            ]
             if rows.empty:
                 continue
-            gakuseki_votes[str(rows.iloc[0]["gakuseki"])] += 1
+            identity_votes[str(rows.iloc[0][schema.gakuseki_column])] += 1
 
-        if not gakuseki_votes:
+        if not identity_votes:
             return None
 
-        top_count = max(gakuseki_votes.values())
-        winners = [g for g, c in gakuseki_votes.items() if c == top_count]
+        top_count = max(identity_votes.values())
+        winners = [g for g, c in identity_votes.items() if c == top_count]
         if len(winners) != 1:
             return None
-        gakuseki = winners[0]
+        identity = winners[0]
 
         reporters: list[dict[str, str]] = [
             {
@@ -251,11 +271,11 @@ def _compute_adoption_for_bt_addr(  # pylint: disable=too-many-locals
         ]
 
         return (
-            gakuseki,
+            identity,
             reporter_count,
             reporters,
             content_hash,
-            dict(gakuseki_votes),
+            dict(identity_votes),
         )
 
     def validate_chain(self) -> bool:

src/ble_blockchain/config/loader.py

@@ -2,7 +2,7 @@
 
 import json
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any
 
 from ble_blockchain.paths import repo_root
@@ -39,6 +39,21 @@ class PathsConfig:
     chain_export_dir: str
 
 
+@dataclass(frozen=True)
+class DataSchema:
+    """Column and field names for the observation data model."""
+    gakuseki_column: str = "gakuseki"
+    bt_addr_column: str = "bt_addrs"
+    csv_identity_rename: dict[str, str] = field(
+        default_factory=lambda: {"学籍番号": "gakuseki"}
+    )
+    output_columns: list[str] = field(
+        default_factory=lambda: ["gakuseki", "bt_addrs", "device_name"]
+    )
+    input_field: str = "gakuseki"
+    output_field: str = "bt_addrs"
+
+
 @dataclass(frozen=True)
 class BlockchainConfig:
     """Blockchain majority and export rules."""
@@ -48,6 +63,7 @@ class BlockchainConfig:
     min_verified_receives: int
     require_content_hash_agreement: bool
     min_distinct_devices_for_aggregate: int
+    data_schema: DataSchema
 
 
 def load_l2cap_config() -> L2capConfig:
@@ -78,6 +94,24 @@ def load_paths_config() -> PathsConfig:
     )
 
 
+def load_data_schema(raw: dict[str, Any]) -> DataSchema:
+    """Build a DataSchema from a partial data_schema config block."""
+    rename = raw.get("csv_identity_rename", {"学籍番号": "gakuseki"})
+    return DataSchema(
+        gakuseki_column=str(raw.get("gakuseki_column", "gakuseki")),
+        bt_addr_column=str(raw.get("bt_addr_column", "bt_addrs")),
+        csv_identity_rename={str(key): str(val) for key, val in rename.items()},
+        output_columns=[
+            str(column)
+            for column in raw.get(
+                "output_columns", ["gakuseki", "bt_addrs", "device_name"]
+            )
+        ],
+        input_field=str(raw.get("input_field", "gakuseki")),
+        output_field=str(raw.get("output_field", "bt_addrs")),
+    )
+
+
 def load_blockchain_config() -> BlockchainConfig:
     """Load blockchain rules from config/blockchain.json."""
     data = load_json_config("blockchain.json")
@@ -92,6 +126,7 @@ def load_blockchain_config() -> BlockchainConfig:
         min_distinct_devices_for_aggregate=int(
             data.get("min_distinct_devices_for_aggregate", 2)
         ),
+        data_schema=load_data_schema(data.get("data_schema", {})),
     )