cic-cli/cic/contract/contract.py

222 lines
6.6 KiB
Python

# Standard
import importlib
import json
import logging
import os
from typing import List, TYPE_CHECKING
import requests
# external imports
from cic_types.ext.metadata import MetadataRequestsHandler
from cic_types.ext.metadata.signer import Signer as MetadataSigner
from chainlib.cli.config import Config
from chainlib.chain import ChainSpec
# Local Modules
from cic.contract.processor import ContractProcessor
from cic.contract.components.attachment import Attachment
from cic.contract.components.meta import Meta
from cic.contract.network import Network
from cic.contract.components.proof import Proof
from cic.contract.components.token import Token
from cic.contract.helpers import init_writers_from_config
from cic.writers import HTTPWriter, KeyedWriterFactory, OutputWriter, 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 = ""
s += f"[cic.header]\nversion = {self.proof.version()}\n"
s += f"[cic.token]\n{self.token}"
s += f"[cic.proof]\n{self.proof}"
s += f"[cic.meta]\n{self.meta}"
s += f"[cic.attachment]\n{self.attachment}"
s += f"[cic.network]\n{self.network}"
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:
if os.path.exists(directory):
contine = input(
"Directory already exists, Would you like to delete it? (y/n): "
)
if contine.lower() != "y":
print("Trying to load existing contract")
return load_contract(directory)
else:
print(f"Deleted {directory}")
os.system(f"rm -rf {directory}")
os.makedirs(directory)
log.debug("Generating token")
token = Token(directory, interactive=interactive)
token.start()
log.debug("Generating proof")
proof = Proof(directory, interactive=interactive)
proof.start()
log.debug("Generating meta")
meta = Meta(directory, interactive=interactive)
meta.start()
log.debug("Generating attachment")
attachment = Attachment(directory, interactive=interactive)
log.debug("Generating network")
network = Network(directory, targets=targets)
network.start()
log.debug(
f"""Populating infomation from network:
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)
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")
m = importlib.import_module(f"cic.ext.{target}.start")
m.extension_start(
network,
registry_address=config.get("CIC_REGISTRY_ADDRESS"),
chain_spec=ChainSpec.from_chain_str(config.get("CHAIN_SPEC")),
rpc_provider=config.get("RPC_PROVIDER"),
key_account_address=key_account_address
)
network.load()
return Contract(
token=token, proof=proof, meta=meta, attachment=attachment, network=network
)
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")
if metadata_endpoint is not None:
MetadataRequestsHandler.base_url = metadata_endpoint
MetadataSigner.gpg_path = "/tmp"
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")
print(chain_spec)
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()