workspace.py

Go to the documentation of this file.

 
 
 
import json
import os
import re
import shutil
from contextlib import contextmanager
from dataclasses import asdict,dataclass
from datetime import datetime
 
 
 
@dataclass
class RunWorkspace:
    repo_root: str
    run_id: str
    command_id: str
    run_root: str
    workspace_root: str
    copy_root: str
    dynamic_root: str
    data_dir: str
    temp_dir: str
    internal_report_dir: str
    report_root: str
    metadata_path: str
 
 
 
def getRepoRoot():
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 
 
def slugifyProjectName(projPath):
    normalized=str(projPath).strip().rstrip('/\\')
    if not normalized:
        return 'project'
    name=re.split(r'[\\/]+',normalized)[-1]
    slug=re.sub(r'[^A-Za-z0-9._-]+','-',name).strip('.-_')
    return slug or 'project'
 
 
 
def _nextRunId(runsRoot,projectSlug,timestamp):
    prefix=f'{projectSlug}__{timestamp}'
    for index in range(1, 1000):
        runId=f'{prefix}-{index:03d}'
        runRoot=os.path.join(runsRoot,runId)
        try:
            os.mkdir(runRoot)
            return runId, runRoot
        except FileExistsError:
            continue
    raise RuntimeError(f'Cannot allocate run id for {prefix}')
 
 
 
def createRunWorkspace(
    repoRoot,
    projPath,
    runCommand,
    runPath,
    libName,
    currentVersion,
    targetVersion,
    currentEnv,
    targetEnv,
    commandId='cmd-001',
    timestamp=None,
    writePatch=False,
):
    repoRoot=os.path.abspath(repoRoot)
    timestamp=timestamp or datetime.now().strftime('%Y%m%d-%H%M%S')
    projectSlug=slugifyProjectName(projPath)
    runsRoot=os.path.join(repoRoot,'PCARTRuns','runs')
    os.makedirs(runsRoot, exist_ok=True)
    runId,runRoot=_nextRunId(runsRoot,projectSlug,timestamp)
 
    workspaceRoot=os.path.join(runRoot,commandId)
    os.mkdir(workspaceRoot)
    copyRoot=os.path.join(workspaceRoot,'Copy')
    dynamicRoot=os.path.join(workspaceRoot,'Dynamic')
    dataDir=os.path.join(workspaceRoot,'data')
    tempDir=os.path.join(workspaceRoot,'temp')
    internalReportDir=os.path.join(workspaceRoot,'Report')
    reportRoot=os.path.join(repoRoot,'Report','runs',runId,commandId)
    metadataPath=os.path.join(workspaceRoot,'metadata.json')
 
    for path in (
        copyRoot,
        dynamicRoot,
        dataDir,
        tempDir,
        internalReportDir,
        reportRoot,
        os.path.join(reportRoot,'patches'),
        os.path.join(reportRoot,'fixed_project'),
    ):
        os.makedirs(path,exist_ok=True)
 
    workspace=RunWorkspace(repo_root=repoRoot,run_id=runId,command_id=commandId,run_root=runRoot,workspace_root=workspaceRoot,copy_root=copyRoot,dynamic_root=dynamicRoot,data_dir=dataDir,temp_dir=tempDir,internal_report_dir=internalReportDir,report_root=reportRoot,metadata_path=metadataPath)
    writeMetadata(
        workspace,
        {
            'project_slug': projectSlug,
            'project_path': projPath,
            'run_command': runCommand,
            'run_file_path': runPath,
            'lib_name': libName,
            'current_version': currentVersion,
            'target_version': targetVersion,
            'current_env': currentEnv,
            'target_env': targetEnv,
            'write_patch': writePatch,
        },
    )
    return workspace
 
 
 
def writeMetadata(workspace,metadata):
    content=asdict(workspace)
    content.update(metadata)
    with open(workspace.metadata_path,'w',encoding='utf-8') as fw:
        json.dump(content,fw,ensure_ascii=False,indent=2)
 
 
 
def getRuntimePaths(workspace):
    return {
        'workspace_root': workspace.workspace_root,
        'copy_root': workspace.copy_root,
        'dynamic_root': workspace.dynamic_root,
        'data_dir': workspace.data_dir,
        'temp_dir': workspace.temp_dir,
        'report_dir': workspace.internal_report_dir,
    }
 
 
 
@contextmanager
def workspaceCwd(path):
    previous=os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(previous)
 
 
 
def exportRunReport(workspace):
    os.makedirs(workspace.report_root, exist_ok=True)
    os.makedirs(os.path.join(workspace.report_root,'patches'),exist_ok=True)
    os.makedirs(os.path.join(workspace.report_root,'fixed_project'),exist_ok=True)
    if not os.path.isdir(workspace.internal_report_dir):
        return
    for name in os.listdir(workspace.internal_report_dir):
        source=os.path.join(workspace.internal_report_dir,name)
        target=os.path.join(workspace.report_root,name)
        if os.path.isdir(source):
            if os.path.exists(target):
                shutil.rmtree(target)
            shutil.copytree(source,target)
        else:
            shutil.copy2(source,target)
 
 
 
def cleanupRunWorkspace(workspace):
    runsRoot=os.path.realpath(os.path.join(workspace.repo_root,'PCARTRuns','runs'))
    runRoot=os.path.realpath(workspace.run_root)
    normalizedRunsRoot=os.path.normcase(os.path.normpath(runsRoot))
    normalizedParent=os.path.normcase(os.path.normpath(os.path.dirname(runRoot)))
 
    if normalizedParent!=normalizedRunsRoot:
        raise ValueError(f'Refuse to remove workspace outside PCARTRuns/runs: {runRoot}')
 
    if os.path.isdir(runRoot):
        shutil.rmtree(runRoot)