cic-cli/cic/contract/contract.py

215 lines
6.5 KiB
Python
Raw Normal View History

# Standard
import importlib
import logging
import os
2022-04-26 13:03:38 +02:00
from typing import List
# External imports
2022-03-01 09:17:17 +01:00
from chainlib.chain import ChainSpec
2022-03-07 10:25:45 +01:00
from chainlib.cli.config import Config
from cic_types.ext.metadata import MetadataRequestsHandler
from cic_types.ext.metadata.signer import Signer as MetadataSigner
2022-04-26 13:03:38 +02:00
# Local Modules
from cic.contract.components.attachment import Attachment
from cic.contract.components.meta import Meta
from cic.contract.components.proof import Proof
from cic.contract.components.token import Token
from cic.contract.helpers import init_writers_from_config
2022-03-07 10:25:45 +01:00
from cic.contract.network import Network
from cic.contract.processor import ContractProcessor
2022-04-26 13:03:38 +02:00
from cic.writers import HTTPWriter, KeyedWriterFactory, MetadataWriter
log = logging.getLogger(__name__)
class Contract:
""" """
def __init__(
self,
token: Token,
proof: Proof,
meta: Meta,
attachment: Attachment,
network: Network,
):
self.token = token
self.proof = proof
self.meta = meta
self.attachment = attachment
self.network = network
def __str__(self):
s = ""
2022-03-07 10:25:45 +01:00
s += f"\n[cic.header]\nversion = {self.proof.version()}\n\n"
s += f"[cic.token]\n{self.token}\n"
s += f"[cic.proof]\n{self.proof}\n"
s += f"[cic.meta]\n{self.meta}\n"
s += f"[cic.attachment]\n{self.attachment}\n"
s += f"[cic.network]\n{self.network}\n"
return s
def load_contract(directory) -> Contract:
token = Token(path=directory)
proof = Proof(path=directory)
meta = Meta(path=directory)
attachment = Attachment(path=directory)
network = Network(directory)
token.load()
proof.load()
meta.load()
attachment.load()
network.load()
return Contract(
token=token, proof=proof, meta=meta, attachment=attachment, network=network
)
def generate_contract(
directory: str, targets: List[str], config, interactive=True
) -> Contract:
os.makedirs(directory)
2022-04-26 13:03:38 +02:00
log.info("Generating token")
token = Token(directory, interactive=interactive)
token.start()
2022-04-26 13:03:38 +02:00
log.info("Generating proof")
proof = Proof(directory, interactive=interactive)
proof.start()
2022-04-26 13:03:38 +02:00
log.info("Generating meta")
meta = Meta(directory, interactive=interactive)
meta.start()
2022-04-26 13:03:38 +02:00
log.info("Generating attachment")
attachment = Attachment(directory, interactive=interactive)
2022-04-26 13:03:38 +02:00
log.info("Generating network")
network = Network(directory, targets=targets)
network.start()
2022-04-26 13:03:38 +02:00
log.info(
f"""Populating infomation from network:
2022-03-01 09:17:17 +01:00
CIC_REGISTRY_ADDRESS: {config.get("CIC_REGISTRY_ADDRESS")}
CHAIN_SPEC: {config.get("CHAIN_SPEC")}
RPC_PROVIDER: {config.get("RPC_PROVIDER")}
"""
)
for target in targets:
# TODO Clean this up
modname = f"cic.ext.{target}"
cmd_mod = importlib.import_module(modname)
signer_hint = config.get("WALLET_KEY_FILE")
keys = cmd_mod.list_keys(config, signer_hint)
2022-03-07 10:25:45 +01:00
if len(keys) > 1:
2022-04-26 13:03:38 +02:00
print("More than one key found, please select one:")
2022-03-07 10:25:45 +01:00
for idx, key in enumerate(keys):
print(f"{idx} - {key} ")
selecting_key = True
while selecting_key:
idx = int(input("Select key: "))
if keys[idx] is not None:
key_account_address = keys[idx]
selecting_key = False
else:
print("Invalid key, try again")
else:
key_account_address = keys[0]
m = importlib.import_module(f"cic.ext.{target}.start")
m.extension_start(
network,
registry_address=config.get("CIC_REGISTRY_ADDRESS"),
2022-03-01 09:17:17 +01:00
chain_spec=ChainSpec.from_chain_str(config.get("CHAIN_SPEC")),
rpc_provider=config.get("RPC_PROVIDER"),
2022-03-07 10:25:45 +01:00
key_account_address=key_account_address,
)
network.load()
return Contract(
token=token, proof=proof, meta=meta, attachment=attachment, network=network
)
2022-03-01 09:17:17 +01:00
def deploy_contract(
config: Config,
target: str,
contract_directory: str,
):
modname = f"cic.ext.{target}"
cmd_mod = importlib.import_module(modname)
writers = init_writers_from_config(config)
output_directory = os.path.join(contract_directory, "out")
output_writer_path_meta = output_directory
metadata_endpoint = config.get("META_URL")
metadata_auth_token = config.get("META_AUTH_TOKEN")
if metadata_endpoint is not None:
MetadataRequestsHandler.base_url = metadata_endpoint
MetadataRequestsHandler.auth_token = metadata_auth_token
MetadataSigner.gpg_path = "/tmp"
2022-03-01 09:17:17 +01:00
MetadataSigner.key_file_path = config.get("AUTH_KEYFILE_PATH")
MetadataSigner.gpg_passphrase = config.get("AUTH_PASSPHRASE")
writers["proof"] = KeyedWriterFactory(MetadataWriter, HTTPWriter).new
writers["attachment"] = KeyedWriterFactory(None, HTTPWriter).new
writers["meta"] = MetadataWriter
output_writer_path_meta = metadata_endpoint
ct = Token(path=contract_directory)
cm = Meta(
path=contract_directory, writer=writers["meta"](path=output_writer_path_meta)
)
ca = Attachment(
path=contract_directory,
writer=writers["attachment"](path=output_writer_path_meta),
)
cp = Proof(
path=contract_directory,
attachments=ca,
writer=writers["proof"](path=output_writer_path_meta),
)
cn = Network(path=contract_directory)
ca.load()
ct.load()
cp.load()
cm.load()
cn.load()
chain_spec = None
try:
chain_spec = config.get("CHAIN_SPEC")
log.debug(f"using CHAIN_SPEC from config: {chain_spec}")
except KeyError:
chain_spec = cn.chain_spec
config.add(chain_spec, "CHAIN_SPEC", exists_ok=True)
log.debug(f"using CHAIN_SPEC: {str(chain_spec)} from network")
signer_hint = config.get("WALLET_KEY_FILE")
(rpc, signer) = cmd_mod.parse_adapter(config, signer_hint)
target_network_reference = cn.resource(target)
chain_spec = cn.chain_spec(target)
log.debug(
f'found reference {target_network_reference["contents"]} chain spec {chain_spec} for target {target}'
)
c = getattr(cmd_mod, "new")(
chain_spec,
target_network_reference["contents"],
cp,
signer_hint=signer,
rpc=rpc,
outputs_writer=writers["ext"](path=output_directory),
)
c.apply_token(ct)
p = ContractProcessor(proof=cp, attachment=ca, metadata=cm, extensions=[c])
p.process()