bad_is_date_sequential.py

"""
Uses the Google Cloud Data Labeling API to ask human annotators whether the dates are sequential.
Requires a billing account with an active credit card.
You can use your mother's if you don't have one - just don't tell her about it.
"""

GCLOUD_PROJECT_ID = "<your google cloud project id>"
INPUT_GCS_URI = "gs://<GCS resource file formatted as CSV, with the format below>.csv"
"""Sample format: (you don't really have to use a fixed format, just get every date on the same line and the human will figure it out)
2018-12-21 17:56:37;2018-09-11 19:41:17;2018-10-11 16:34:41;2020-07-27 03:10:17;2019-02-04 19:58:02
..."""

labeling_instruction_name = "Are the following dates in chronological order?"
labeling_instruction_description = "Reply yes or no - if they are chronological reply yes, otherwise reply no."
dataset_display_name = "List of dates"
dataset_description = "A list of dates, you need to check if they are in chronological order."
labels = ["Yes", "No"]
labels_descs = ["Yes, they are in chronological order", "No, they are not in chronological order"]

def is_date_sequential():
    dataset = create_dataset(GCLOUD_PROJECT_ID)
    import_data(dataset.name, "TEXT", INPUT_GCS_URI)
    spec_response = create_annotation_spec_set(GCLOUD_PROJECT_ID)
    instruction = create_instruction(GCLOUD_PROJECT_ID, datalabeling.DataType.TEXT, INPUT_GCS_URI)
    label_text(dataset.name, instruction.name, spec_response.name)
    print("Submit complete! Please go to Google Cloud Console tomorrow morning to see the result.")

def create_instruction(project_id, data_type, instruction_gcs_uri):
    """ Creates a data labeling PDF instruction for the given Google Cloud
    project. The PDF file should be uploaded to the project in
    Google Cloud Storage.
    """
    from google.cloud import datalabeling_v1beta1 as datalabeling

    client = datalabeling.DataLabelingServiceClient()

    project_path = f"projects/{project_id}"

    pdf_instruction = datalabeling.PdfInstruction(gcs_file_uri=instruction_gcs_uri)

    instruction = datalabeling.Instruction(
        display_name=labeling_instruction_name,
        description=labeling_instruction_description,
        data_type=data_type,
        pdf_instruction=pdf_instruction,
    )

    operation = client.create_instruction(
        request={"parent": project_path, "instruction": instruction}
    )

    result = operation.result()

    # The format of the resource name:
    # project_id/{project_id}/instruction/{instruction_id}
    print("The instruction resource name: {}".format(result.name))
    print("Display name: {}".format(result.display_name))
    print("Description: {}".format(result.description))
    print("Create time:")
    print("\tseconds: {}".format(result.create_time.timestamp_pb().seconds))
    print("\tnanos: {}".format(result.create_time.timestamp_pb().nanos))
    print("Data type: {}".format(datalabeling.DataType(result.data_type).name))
    print("Pdf instruction:")
    print("\tGcs file uri: {}\n".format(result.pdf_instruction.gcs_file_uri))

    return result

def create_dataset(project_id):
    """Creates a dataset for the given Google Cloud project."""
    from google.cloud import datalabeling_v1beta1 as datalabeling

    client = datalabeling.DataLabelingServiceClient()

    formatted_project_name = f"projects/{project_id}"

    dataset = datalabeling.Dataset(
        display_name=dataset_display_name, description=dataset_description
    )

    response = client.create_dataset(
        request={"parent": formatted_project_name, "dataset": dataset}
    )

    # The format of resource name:
    # project_id/{project_id}/datasets/{dataset_id}
    print("The dataset resource name: {}".format(response.name))
    print("Display name: {}".format(response.display_name))
    print("Description: {}".format(response.description))
    print("Create time:")
    print("\tseconds: {}".format(response.create_time.timestamp_pb().seconds))
    print("\tnanos: {}\n".format(response.create_time.timestamp_pb().nanos))

    return response

def create_annotation_spec_set(project_id):
    """Creates a data labeling annotation spec set for the given
    Google Cloud project.
    """
    from google.cloud import datalabeling_v1beta1 as datalabeling

    client = datalabeling.DataLabelingServiceClient()

    project_path = f"projects/{project_id}"

    annotation_spec_1 = datalabeling.AnnotationSpec(
        display_name=labels[1], description=labels_descs[1]
    )

    annotation_spec_2 = datalabeling.AnnotationSpec(
        display_name=labels[2], description=labels_descs[2]
    )

    annotation_spec_set = datalabeling.AnnotationSpecSet(
        display_name=dataset_display_name,
        description=dataset_description,
        annotation_specs=[annotation_spec_1, annotation_spec_2],
    )

    response = client.create_annotation_spec_set(
        request={"parent": project_path, "annotation_spec_set": annotation_spec_set}
    )

    # The format of the resource name:
    # project_id/{project_id}/annotationSpecSets/{annotationSpecSets_id}
    print("The annotation_spec_set resource name: {}".format(response.name))
    print("Display name: {}".format(response.display_name))
    print("Description: {}".format(response.description))
    print("Annotation specs:")
    for annotation_spec in response.annotation_specs:
        print("\tDisplay name: {}".format(annotation_spec.display_name))
        print("\tDescription: {}\n".format(annotation_spec.description))

    return response

def import_data(dataset_resource_name, data_type, input_gcs_uri):
    """Imports data to the given Google Cloud project and dataset."""
    from google.cloud import datalabeling_v1beta1 as datalabeling

    client = datalabeling.DataLabelingServiceClient()

    gcs_source = datalabeling.GcsSource(input_uri=input_gcs_uri, mime_type="text/csv")

    csv_input_config = datalabeling.InputConfig(
        data_type=data_type, gcs_source=gcs_source
    )

    response = client.import_data(
        request={"name": dataset_resource_name, "input_config": csv_input_config}
    )

    result = response.result()

    # The format of resource name:
    # project_id/{project_id}/datasets/{dataset_id}
    print("Dataset resource name: {}\n".format(result.dataset))

    return result

def label_text(
    dataset_resource_name, instruction_resource_name, annotation_spec_set_resource_name
):
    """Labels a text dataset."""
    from google.cloud import datalabeling_v1beta1 as datalabeling

    client = datalabeling.DataLabelingServiceClient()

    basic_config = datalabeling.HumanAnnotationConfig(
        instruction=instruction_resource_name,
        annotated_dataset_display_name=dataset_display_name,
        label_group="label group 1",
        replica_count=1,
    )

    feature = datalabeling.LabelTextRequest.Feature.TEXT_CLASSIFICATION

    config = datalabeling.TextEntityExtractionConfig(
        annotation_spec_set=annotation_spec_set_resource_name
    )

    response = client.label_text(
        request={
            "parent": dataset_resource_name,
            "basic_config": basic_config,
            "feature": feature,
            "text_classification_config": config,
        }
    )

    print("Label_text operation name: {}".format(response.operation.name))
    return response