def print_heading()

in src/buildstream/_frontend/widget.py [0:0]

• 99 lines of code
• 25 McCabe index (conditional complexity)

    def print_heading(self, toplevel_project, stream, *, log_file):
        context = self.context
        starttime = datetime.datetime.now()
        text = ""

        def format_spec(spec):
            if spec.instance_name:
                return "{} (instance: {})".format(spec.url, spec.instance_name)
            return spec.url

        self._resolved_keys = {element: element._get_cache_key() for element in stream.session_elements}

        # Main invocation context
        text += "\n"
        text += self.content_profile.fmt("BuildStream Version {}\n".format(bst_version), bold=True)
        values = {}
        values["Session Start"] = starttime.strftime("%A, %d-%m-%Y at %H:%M:%S")
        if toplevel_project:
            values["Project"] = "{} ({})".format(toplevel_project.name, toplevel_project.directory)
        values["Targets"] = ", ".join([t.name for t in stream.targets])
        text += self._format_values(values)

        # User configurations
        text += "\n"
        text += self.content_profile.fmt("User Configuration\n", bold=True)
        values = {}
        values["Configuration File"] = "Default Configuration" if not context.config_origin else context.config_origin
        values["Cache Directory"] = context.cachedir
        values["Log Files"] = context.logdir
        values["Source Mirrors"] = context.sourcedir
        values["Build Area"] = context.builddir
        values["Strict Build Plan"] = "Yes" if context.get_strict() else "No"
        values["Maximum Fetch Tasks"] = context.sched_fetchers
        values["Maximum Build Tasks"] = context.sched_builders
        values["Maximum Push Tasks"] = context.sched_pushers
        values["Maximum Network Retries"] = context.sched_network_retries

        if context.remote_cache_spec:
            values["Cache Storage Service"] = format_spec(context.remote_cache_spec)

        text += self._format_values(values)

        if context.remote_execution_specs:
            specs = context.remote_execution_specs

            text += "\n"
            text += self.content_profile.fmt("Remote Execution Configuration\n", bold=True)
            values = {}
            values["Execution Service"] = format_spec(specs.exec_spec)
            re_storage_spec = specs.storage_spec or context.remote_cache_spec
            values["Storage Service"] = format_spec(re_storage_spec)
            if specs.action_spec:
                values["Action Cache Service"] = format_spec(specs.action_spec)
            text += self._format_values(values)

        # Print information about each loaded project
        #
        text += "\n"
        if toplevel_project:
            loaded_projects = toplevel_project.loaded_projects()
        else:
            loaded_projects = []

        for project_info in loaded_projects:
            project = project_info.project

            # Project title line
            text += (
                self.content_profile.fmt("Project", bold=True)
                + self.format_profile.fmt(": ", bold=True)
                + self.content_profile.fmt(project.name, bold=True)
            )
            text += "\n"

            # Details on how the project was loaded
            #
            values = {}
            if project.junction:
                values["Junction path"] = project_info.project.junction._get_full_name()
            if project_info.provenance:
                values["Loaded by"] = str(project_info.provenance)
            text += self._format_values(values)

            # Print out duplicate declarations
            if project_info.duplicates:
                text += self.format_profile.fmt("{}Declared duplicate by:\n".format(self._indent))
                for duplicate in project_info.duplicates:
                    text += self.content_profile.fmt("{}{}\n".format(self._indent * 2, duplicate))

            # Print out internal declarations
            if project_info.internal:
                text += self.format_profile.fmt("{}Declared internal by:\n".format(self._indent))
                for internal in project_info.internal:
                    text += self.content_profile.fmt("{}{}\n".format(self._indent * 2, internal))

            text += "\n"

            # Project Options
            values = {}
            project.options.printable_variables(values)
            if values:
                text += self.format_profile.fmt("{}Project Options\n".format(self._indent))
                text += self._format_values(values, indent=2)
                text += "\n"

            # Plugins
            text += self._format_plugins(
                {p: d for p, _, _, d in project.element_factory.list_plugins()},
                {p: d for p, _, _, d in project.source_factory.list_plugins()},
            )

            # Artifact cache servers
            specs = context.project_artifact_cache_specs.get(project.name)
            if specs:
                text += self.format_profile.fmt("{}Artifact cache servers\n".format(self._indent))
                text += self._format_list(specs, indent=2)
                text += "\n"

            # Source cache servers
            specs = context.project_source_cache_specs.get(project.name)
            if specs:
                text += self.format_profile.fmt("{}Source cache servers\n".format(self._indent))
                text += self._format_list(specs, indent=2)
                text += "\n"

        # Pipeline state
        text += self.content_profile.fmt("Pipeline\n", bold=True)
        text += self.show_pipeline(stream.total_elements, context.log_element_format)
        text += "\n"

        # Separator line before following output
        text += self.format_profile.fmt("=" * 79 + "\n")

        click.echo(text, nl=False, err=True)
        if log_file:
            click.echo(text, file=log_file, color=False, nl=False)