Skip to content
Open
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 77 additions & 1 deletion invisible_cities/cities/components.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import warnings
import math
import os
import subprocess

from .. dataflow import dataflow as fl
from .. dataflow.dataflow import sink
Expand Down Expand Up @@ -143,7 +144,8 @@ def proxy(**kwds):
args = vars(conf)
result = check_annotations(city_function)(**args)
if os.path.exists(conf.file_out):
write_city_configuration(conf.file_out, city_function.__name__, args)
git_info = add_git_info()
write_city_configuration(conf.file_out, city_function.__name__, {**args, **git_info})
copy_cities_configuration(conf.files_in[0], conf.file_out)
index_tables(conf.file_out)
return result
Expand Down Expand Up @@ -198,6 +200,80 @@ def create_timestamp_(event_number: Union[int, float]) -> float:
return create_timestamp_


def get_git_tag() -> str:
"""Returns the current tag if on a tagged commit, or '-' if not."""
try:
tag = subprocess.check_output(
['git', 'describe', '--tags', '--exact-match'],
stderr=subprocess.DEVNULL
).decode('ascii').strip()
return tag
except subprocess.CalledProcessError:
print("Warning: No tag found for current commit.")
return "-"
except Exception as e:
print(traceback.format_exc())
return "-"


def get_git_upstream_remote() -> str:
"""Returns the upstream remote name, or '-' if not found."""
try:
upstream = subprocess.check_output(
["git", "rev-parse", "--abbrev-ref", "@{upstream}"],
stderr=subprocess.DEVNULL
).decode("ascii").strip()
remote_name = upstream.split('/')[0] # take the part before '/'
return remote_name
except subprocess.CalledProcessError:
print("Warning: No upstream remote configured for this branch.")
return "-"
except Exception as e:
print(traceback.format_exc())
return "-"


def get_git_branch() -> str:
"""Returns the current branch name, or '-' if not found."""
try:
branch = subprocess.check_output(
['git', 'branch', '--show-current'],
stderr=subprocess.DEVNULL
).decode('ascii').strip()
return branch
except subprocess.CalledProcessError:
print("Warning: No branch found.")
return "-"
except Exception as e:
print(traceback.format_exc())
return "-"


def get_git_commit_hash() -> str:
"""Returns the current commit hash, or '-' if not found."""
try:
hash = subprocess.check_output(
['git', 'rev-parse', 'HEAD'],
stderr=subprocess.DEVNULL
).decode('ascii').strip()
return hash
except subprocess.CalledProcessError:
print("Warning: No commit hash found.")
return "-"
except Exception as e:
print(traceback.format_exc())


def add_git_info():
git_info = dict(
IC_tag = get_git_tag(),
upstream_name = get_git_upstream_remote(),
branch_name = get_git_branch(),
commit_hash = get_git_commit_hash()
)
return git_info


def index_tables(file_out):
"""
-finds all tables in output_file
Expand Down